From c1b3a426c6f13240d03f2dd9e49b359b291da0f5 Mon Sep 17 00:00:00 2001 From: "coderabbitai[bot]" <136622811+coderabbitai[bot]@users.noreply.github.com> Date: Tue, 10 Feb 2026 17:44:20 +0000 Subject: [PATCH] =?UTF-8?q?=F0=9F=93=9D=20Add=20docstrings=20to=20`feature?= =?UTF-8?q?/Replication`?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Docstrings generation was requested by @unikdahal. * https://github.com/unikdahal/redis-java/pull/49#issuecomment-3879670413 The following files were modified: * `src/main/java/com/redis/commands/ICommand.java` * `src/main/java/com/redis/commands/generic/DelCommand.java` * `src/main/java/com/redis/commands/generic/ExpireCommand.java` * `src/main/java/com/redis/commands/generic/PExpireAtCommand.java` * `src/main/java/com/redis/commands/list/BLPopCommand.java` * `src/main/java/com/redis/commands/list/LPopCommand.java` * `src/main/java/com/redis/commands/list/LPushCommand.java` * `src/main/java/com/redis/commands/list/RPushCommand.java` * `src/main/java/com/redis/commands/replication/InfoCommand.java` * `src/main/java/com/redis/commands/replication/PsyncCommand.java` * `src/main/java/com/redis/commands/replication/ReplconfCommand.java` * `src/main/java/com/redis/commands/replication/WaitCommand.java` * `src/main/java/com/redis/commands/stream/XAddCommand.java` * `src/main/java/com/redis/commands/string/IncrCommand.java` * `src/main/java/com/redis/commands/string/SetCommand.java` * `src/main/java/com/redis/config/RedisConfig.java` * `src/main/java/com/redis/replication/CommandPropagator.java` * `src/main/java/com/redis/replication/MasterConnection.java` * `src/main/java/com/redis/replication/RdbGenerator.java` * `src/main/java/com/redis/replication/ReplicaConnection.java` * `src/main/java/com/redis/replication/ReplicationLog.java` * `src/main/java/com/redis/replication/ReplicationManager.java` * `src/main/java/com/redis/replication/SnapshotProducer.java` * `src/main/java/com/redis/server/NettyRedisServer.java` * `src/main/java/com/redis/server/RedisCommandHandler.java` * `src/main/java/com/redis/storage/RedisDatabase.java` * `src/main/java/com/redis/storage/RedisValue.java` * `src/test/java/com/redis/integration/ReplicationCommandsIT.java` --- .../java/com/redis/commands/ICommand.java | 44 +-- .../redis/commands/generic/DelCommand.java | 12 +- .../redis/commands/generic/ExpireCommand.java | 47 +-- .../commands/generic/PExpireAtCommand.java | 19 +- .../com/redis/commands/list/BLPopCommand.java | 5 + .../com/redis/commands/list/LPopCommand.java | 5 + .../com/redis/commands/list/LPushCommand.java | 12 +- .../com/redis/commands/list/RPushCommand.java | 5 + .../commands/replication/InfoCommand.java | 58 ++- .../commands/replication/PsyncCommand.java | 39 ++- .../commands/replication/ReplconfCommand.java | 55 ++- .../commands/replication/WaitCommand.java | 19 +- .../redis/commands/stream/XAddCommand.java | 28 +- .../redis/commands/string/IncrCommand.java | 12 +- .../com/redis/commands/string/SetCommand.java | 28 +- .../java/com/redis/config/RedisConfig.java | 79 ++++- .../redis/replication/CommandPropagator.java | 63 ++-- .../redis/replication/MasterConnection.java | 192 ++++++++-- .../com/redis/replication/RdbGenerator.java | 32 +- .../redis/replication/ReplicaConnection.java | 99 ++++-- .../com/redis/replication/ReplicationLog.java | 47 +-- .../redis/replication/ReplicationManager.java | 329 +++++++++++++++--- .../redis/replication/SnapshotProducer.java | 94 ++++- .../com/redis/server/NettyRedisServer.java | 26 +- .../com/redis/server/RedisCommandHandler.java | 40 ++- .../java/com/redis/storage/RedisDatabase.java | 42 +-- .../java/com/redis/storage/RedisValue.java | 89 ++++- .../integration/ReplicationCommandsIT.java | 38 +- 28 files changed, 1240 insertions(+), 318 deletions(-) diff --git a/src/main/java/com/redis/commands/ICommand.java b/src/main/java/com/redis/commands/ICommand.java index e3b386a..301b833 100644 --- a/src/main/java/com/redis/commands/ICommand.java +++ b/src/main/java/com/redis/commands/ICommand.java @@ -64,27 +64,23 @@ public interface ICommand { String execute(List args, ChannelHandlerContext ctx); /** - * Returns the command name (e.g., "SET", "GET", "DEL"). - * Command names are case-insensitive at lookup time. - */ + * Provides the canonical name of the command. + * + * @return the command name (e.g., "SET", "GET", "DEL"); lookup of command names is case-insensitive + */ String name(); /** - * Returns the canonical arguments for replication. - *

- * Override this method when the command needs to be rewritten for replication - * to ensure a consistent state across replicas. Common cases include: - *

- *

- * The default implementation returns null, indicating the original args - * should be used for propagation. + * Provide the canonical argument list to use when propagating this command to replicas. * - * @param originalArgs The original command arguments - * @param response The response from execute() - can be used to extract generated values - * @return Canonical arguments for replication, or null to use original args + *

Override to return a fixed, canonical form when replication must not depend on + * client-provided or runtime-generated values (for example: generated IDs, or relative + * time arguments converted to absolute timestamps). + * + * @param originalArgs the original command arguments (excluding command name) + * @param response the response produced by {@link #execute(List, io.netty.channel.ChannelHandlerContext)}, + * which may contain generated values needed to form canonical args + * @return the canonical arguments to replicate, or `null` to indicate the original arguments should be used */ default List getReplicationArgs(List originalArgs, String response) { return null; // Default: use original args @@ -103,16 +99,14 @@ default boolean isWriteCommand() { } /** - * Returns the command name to use for replication. - *

- * Override when the command should be replicated as a different command. - * For example, EXPIRE should be replicated as PEXPIREAT to use absolute timestamps. - *

- * Default returns null, meaning use the original command name. + * Specifies an alternative command name to use when replicating this command. + * + * Override to substitute a different command name for replication (for example, + * replicate EXPIRE as PEXPIREAT to use absolute timestamps). * - * @return The command name to use for replication, or null to use original + * @return the replication command name, or `null` to use the original command name */ default String getReplicationCommandName() { return null; } -} +} \ No newline at end of file diff --git a/src/main/java/com/redis/commands/generic/DelCommand.java b/src/main/java/com/redis/commands/generic/DelCommand.java index 4ced3dc..ff2adf1 100644 --- a/src/main/java/com/redis/commands/generic/DelCommand.java +++ b/src/main/java/com/redis/commands/generic/DelCommand.java @@ -30,13 +30,23 @@ public String execute(List args, ChannelHandlerContext ctx) { .toString(); } + /** + * Identifier for the DEL command implemented by this class. + * + * @return the command name "DEL" + */ @Override public String name() { return "DEL"; } + /** + * Indicates whether this command performs a write operation on the database. + * + * @return `true` if the command modifies the database, `false` otherwise. + */ @Override public boolean isWriteCommand() { return true; } -} +} \ No newline at end of file diff --git a/src/main/java/com/redis/commands/generic/ExpireCommand.java b/src/main/java/com/redis/commands/generic/ExpireCommand.java index 8ecf4e0..af7aa7a 100644 --- a/src/main/java/com/redis/commands/generic/ExpireCommand.java +++ b/src/main/java/com/redis/commands/generic/ExpireCommand.java @@ -29,16 +29,15 @@ public class ExpireCommand implements ICommand { private static final ThreadLocal lastComputedExpiry = new ThreadLocal<>(); /** - * Set a seconds-based expiration timestamp for the specified key. - * - * Expects {@code args} to contain the key at index 0 and the expiry in seconds at index 1. - * Returns the command error reply {@code ERR_WRONG_ARGS} if fewer than two arguments are provided, - * or {@code ERR_VALUE} if the expiry value is not a valid integer. - * - * @param args the command arguments: {@code [key, seconds]} - * @param ctx the Netty channel context (not used by this implementation) - * @return {@code ":1\r\n"} if the expiry was set, {@code ":0\r\n"} otherwise - */ + * Set an expiration for the given key using a relative seconds value. + * + * Expects {@code args} to contain the key at index 0 and the expiry in seconds at index 1. + * On successful expiry set, records the computed absolute expiry timestamp (milliseconds since epoch) + * in a thread-local for replication rewriting. + * + * @param args the command arguments: {@code [key, seconds]} + * @return {@code ":1\r\n"} if the expiry was set, {@code ":0\r\n"} otherwise + */ @Override public String execute(List args, ChannelHandlerContext ctx) { lastComputedExpiry.remove(); @@ -76,21 +75,26 @@ public String name() { return "EXPIRE"; } + /** + * Indicates this command performs a write operation that modifies the dataset. + * + * @return `true` if the command modifies the data (is a write), `false` otherwise. + */ @Override public boolean isWriteCommand() { return true; } /** - * Returns canonical arguments for replication. - *

- * Converts EXPIRE (relative seconds) to PEXPIREAT (absolute milliseconds) - * to ensure all replicas set the same absolute expiry time. - * - * @param originalArgs The original arguments [key, seconds] - * @param response The response from execute() - * @return PEXPIREAT arguments [key, timestamp_ms], or null if failed - */ + * Produce replication arguments by converting EXPIRE's relative seconds into + * PEXPIREAT's absolute expiration timestamp in milliseconds. + * + *

Only rewrites when the original command succeeded; otherwise returns null. + * + * @param originalArgs the original EXPIRE arguments as [key, seconds] + * @param response the raw response returned by execute() + * @return the replication arguments as [key, timestamp_ms], or null if the command did not succeed or the computed expiry is unavailable + */ @Override public List getReplicationArgs(List originalArgs, String response) { // Only rewrite if successful @@ -114,8 +118,9 @@ public List getReplicationArgs(List originalArgs, String respons } /** - * Returns the rewritten command name for replication. - * EXPIRE becomes PEXPIREAT for absolute time handling. + * Provide the replication command name to use when rewriting EXPIRE for replication. + * + * @return the replication command name "PEXPIREAT", which represents expiry as an absolute Unix-time-millisecond timestamp */ public String getReplicationCommandName() { return "PEXPIREAT"; diff --git a/src/main/java/com/redis/commands/generic/PExpireAtCommand.java b/src/main/java/com/redis/commands/generic/PExpireAtCommand.java index e1b5776..9667607 100644 --- a/src/main/java/com/redis/commands/generic/PExpireAtCommand.java +++ b/src/main/java/com/redis/commands/generic/PExpireAtCommand.java @@ -27,6 +27,13 @@ public class PExpireAtCommand implements ICommand { private static final String ERR_WRONG_ARGS = "-ERR wrong number of arguments for 'PEXPIREAT' command\r\n"; private static final String ERR_VALUE = "-ERR value is not an integer or out of range\r\n"; + /** + * Sets the absolute expiration time (milliseconds since epoch) for the given key and returns a Redis protocol response. + * + * @param args command arguments where args.get(0) is the key and args.get(1) is the expiry timestamp in milliseconds + * @param ctx the Netty channel handler context + * @return ":1\r\n" if the expiry was set, ":0\r\n" if it was not set, ERR_WRONG_ARGS if fewer than two arguments were provided, or ERR_VALUE if the timestamp is not a valid integer + */ @Override public String execute(List args, ChannelHandlerContext ctx) { if (args.size() < 2) { @@ -48,13 +55,23 @@ public String execute(List args, ChannelHandlerContext ctx) { return success ? ":1\r\n" : ":0\r\n"; } + /** + * The Redis command name handled by this command implementation. + * + * @return the command name "PEXPIREAT" + */ @Override public String name() { return "PEXPIREAT"; } + /** + * Indicates that this command modifies the database state. + * + * @return {@code true} if the command performs a write operation, {@code false} otherwise. + */ @Override public boolean isWriteCommand() { return true; } -} +} \ No newline at end of file diff --git a/src/main/java/com/redis/commands/list/BLPopCommand.java b/src/main/java/com/redis/commands/list/BLPopCommand.java index 8c807ba..ee40d5d 100644 --- a/src/main/java/com/redis/commands/list/BLPopCommand.java +++ b/src/main/java/com/redis/commands/list/BLPopCommand.java @@ -249,6 +249,11 @@ public String name() { return "BLPOP"; } + /** + * Indicates that BLPOP modifies the dataset. + * + * @return `true` if the command modifies the dataset, `false` otherwise + */ @Override public boolean isWriteCommand() { return true; diff --git a/src/main/java/com/redis/commands/list/LPopCommand.java b/src/main/java/com/redis/commands/list/LPopCommand.java index 578440c..ce69c1e 100644 --- a/src/main/java/com/redis/commands/list/LPopCommand.java +++ b/src/main/java/com/redis/commands/list/LPopCommand.java @@ -150,6 +150,11 @@ public String name() { return "LPOP"; } + /** + * Indicates whether this command modifies the database state. + * + * @return `true` if the command modifies the database state, `false` otherwise. + */ @Override public boolean isWriteCommand() { return true; diff --git a/src/main/java/com/redis/commands/list/LPushCommand.java b/src/main/java/com/redis/commands/list/LPushCommand.java index eade4e8..172b3c0 100644 --- a/src/main/java/com/redis/commands/list/LPushCommand.java +++ b/src/main/java/com/redis/commands/list/LPushCommand.java @@ -59,13 +59,23 @@ public String execute(List args, ChannelHandlerContext ctx) { return ":" + resultSize.get() + "\r\n"; } + /** + * The registered command name for this implementation. + * + * @return the command name ("LPUSH") + */ @Override public String name() { return "LPUSH"; } + /** + * Indicates that this command performs a write operation on the data store. + * + * @return `true` if the command modifies the data store, `false` otherwise + */ @Override public boolean isWriteCommand() { return true; } -} +} \ No newline at end of file diff --git a/src/main/java/com/redis/commands/list/RPushCommand.java b/src/main/java/com/redis/commands/list/RPushCommand.java index 82f27b8..4717786 100644 --- a/src/main/java/com/redis/commands/list/RPushCommand.java +++ b/src/main/java/com/redis/commands/list/RPushCommand.java @@ -80,6 +80,11 @@ public String name() { return "RPUSH"; } + /** + * Indicates that this command mutates the database state. + * + * @return `true` if the command performs a write operation, `false` otherwise. + */ @Override public boolean isWriteCommand() { return true; diff --git a/src/main/java/com/redis/commands/replication/InfoCommand.java b/src/main/java/com/redis/commands/replication/InfoCommand.java index 2be9272..7381426 100644 --- a/src/main/java/com/redis/commands/replication/InfoCommand.java +++ b/src/main/java/com/redis/commands/replication/InfoCommand.java @@ -30,6 +30,17 @@ public class InfoCommand implements ICommand { STATIC_SERVER_HEADER = sb.toString(); } + /** + * Assembles the requested Redis INFO sections and returns them in Redis Bulk String format. + * + * If no section is specified (or the first argument is "all"), the response includes the + * server, replication, clients, memory, and stats sections. Valid section names are: + * "server", "replication", "clients", "memory", and "stats". + * + * @param args optional arguments where the first element selects which INFO section to include; + * when omitted or "all", all sections are included + * @return a Redis Bulk String containing the requested INFO content (format: $<length>\r\n<content>\r\n) + */ @Override public String execute(List args, ChannelHandlerContext ctx) { String section = args.isEmpty() ? "all" : args.get(0).toLowerCase(); @@ -58,6 +69,14 @@ public String execute(List args, ChannelHandlerContext ctx) { return "$" + content.length() + "\r\n" + content + "\r\n"; } + /** + * Appends the "Server" INFO section to the provided StringBuilder. + * + * The section includes the precomputed server header, `tcp_port`, `uptime_in_seconds`, + * and `uptime_in_days`, and is terminated with an empty line. + * + * @param sb the StringBuilder to append the server information to; this builder is modified + */ private void appendServerInfo(StringBuilder sb) { long uptimeMs = System.currentTimeMillis() - SERVER_START_TIME; sb.append(STATIC_SERVER_HEADER); @@ -67,14 +86,33 @@ private void appendServerInfo(StringBuilder sb) { sb.append("\r\n"); } + /** + * Appends the replication section content to the given StringBuilder. + * + * Appends the replication manager's INFO output followed by a CRLF ("\r\n"). + */ private void appendReplicationInfo(StringBuilder sb) { sb.append(ReplicationManager.getInstance().getInfoReplication()).append("\r\n"); } + /** + * Appends the "Clients" INFO section to the provided StringBuilder. + * + * @param sb the StringBuilder to append the Clients section to; receives `connected_clients` and `blocked_clients` lines + */ private void appendClientsInfo(StringBuilder sb) { sb.append("# Clients\r\nconnected_clients:1\r\nblocked_clients:0\r\n\r\n"); } + /** + * Appends the "Memory" INFO section to the provided StringBuilder. + * + * The section includes `used_memory`, a human-readable `used_memory_human`, + * `total_system_memory`, and a human-readable `total_system_memory_human`, + * followed by a blank line. + * + * @param sb the StringBuilder to append the Memory section to + */ private void appendMemoryInfo(StringBuilder sb) { Runtime rt = Runtime.getRuntime(); long usedMemory = rt.totalMemory() - rt.freeMemory(); @@ -88,14 +126,23 @@ private void appendMemoryInfo(StringBuilder sb) { sb.append("\r\n"); } + /** + * Append the Redis INFO "Stats" section to the provided StringBuilder. + * + * @param sb the StringBuilder to append the Stats section to + */ private void appendStatsInfo(StringBuilder sb) { sb.append("# Stats\r\ntotal_connections_received:1\r\ntotal_commands_processed:0\r\n\r\n"); } /** - * Optimized byte formatter: - * 1. Appends directly to existing StringBuilder to avoid temporary String objects. - * 2. Uses basic math instead of expensive String.format(). + * Append a human-readable representation of a byte count to the given StringBuilder using unit suffixes. + * + * Formats values as bytes ("B") or with one decimal place using "K", "M", or "G" as appropriate and appends + * the result directly to the provided StringBuilder. + * + * @param sb the StringBuilder to append the formatted value to + * @param bytes the byte count to format */ private void formatBytes(StringBuilder sb, long bytes) { if (bytes < 1024) { @@ -109,6 +156,11 @@ private void formatBytes(StringBuilder sb, long bytes) { } } + /** + * Provide the command name for this ICommand implementation. + * + * @return the command name "INFO" + */ @Override public String name() { return "INFO"; diff --git a/src/main/java/com/redis/commands/replication/PsyncCommand.java b/src/main/java/com/redis/commands/replication/PsyncCommand.java index 1fae34a..535d0a9 100644 --- a/src/main/java/com/redis/commands/replication/PsyncCommand.java +++ b/src/main/java/com/redis/commands/replication/PsyncCommand.java @@ -31,6 +31,13 @@ public class PsyncCommand implements ICommand { private static final String ERR_WRONG_ARGS = "-ERR wrong number of arguments for 'PSYNC' command\r\n"; + /** + * Handle the PSYNC command, initiating either a partial or full replication resynchronization for the connected replica. + * + * @param args the PSYNC arguments: expected to contain the replica's requested replication ID at index 0 and the requested offset at index 1 + * @param ctx the Netty channel handler context for the replica connection + * @return an error reply string when the arguments are invalid (e.g., wrong number of arguments); otherwise `null` after the command response has been written to the channel + */ @Override public String execute(List args, ChannelHandlerContext ctx) { if (args.size() < 2) { @@ -71,7 +78,12 @@ public String execute(List args, ChannelHandlerContext ctx) { } /** - * Checks if we can do a partial resync based on replication ID and offset. + * Determine whether a partial resynchronization can be performed for the given replica request. + * + * @param requestedReplId the replication ID provided by the replica (may be "?" to request full sync) + * @param requestedOffset the replication offset provided by the replica (-1 indicates unknown/first sync) + * @param replMgr the replication manager used to evaluate backlog-based partial resync eligibility + * @return `true` if a partial resync can be performed with the given ID and offset, `false` otherwise */ private boolean canDoPartialResync(String requestedReplId, long requestedOffset, ReplicationManager replMgr) { @@ -85,7 +97,16 @@ private boolean canDoPartialResync(String requestedReplId, long requestedOffset, } /** - * Performs a partial resync: sends CONTINUE and backlog data. + * Perform a partial replication synchronization by acknowledging the replica and streaming any missing backlog data. + * + * Updates the replica state to STREAMING, sends a "+CONTINUE \r\n" acknowledgement, writes backlog bytes + * from the given offset when available, and increments the partial resync counter. + * + * @param ctx the Netty channel context to write responses to + * @param replica the replica connection whose state will be updated + * @param replMgr the replication manager used to obtain master repl-id, backlog data, and to update statistics + * @param requestedOffset the replication offset from which to retrieve backlog data + * @return null (response already written to the provided ChannelHandlerContext) */ private String performPartialResync(ChannelHandlerContext ctx, ReplicaConnection replica, ReplicationManager replMgr, long requestedOffset) { @@ -118,7 +139,12 @@ private String performPartialResync(ChannelHandlerContext ctx, ReplicaConnection } /** - * Performs a full resync: sends FULLRESYNC response followed by RDB file. + * Initiates a full replication resynchronization by sending a FULLRESYNC reply and streaming the RDB snapshot to the replica. + * + * @param ctx the channel context for writing responses and RDB data to the replica + * @param replica the replica connection whose state will be updated for the full resync + * @param replMgr the replication manager providing master replication id/offset and statistics tracking + * @return `null` (response and RDB have been written to the channel) */ private String performFullResync(ChannelHandlerContext ctx, ReplicaConnection replica, ReplicationManager replMgr) { @@ -149,8 +175,13 @@ private String performFullResync(ChannelHandlerContext ctx, ReplicaConnection re return null; } + /** + * Command name used to invoke this handler. + * + * @return the literal command name "PSYNC" + */ @Override public String name() { return "PSYNC"; } -} +} \ No newline at end of file diff --git a/src/main/java/com/redis/commands/replication/ReplconfCommand.java b/src/main/java/com/redis/commands/replication/ReplconfCommand.java index 53828e1..f97d9a6 100644 --- a/src/main/java/com/redis/commands/replication/ReplconfCommand.java +++ b/src/main/java/com/redis/commands/replication/ReplconfCommand.java @@ -26,6 +26,16 @@ public class ReplconfCommand implements ICommand { private static final String ERR_WRONG_ARGS = "-ERR wrong number of arguments for 'REPLCONF' command\r\n"; private static final String ERR_UNKNOWN_SUBCOMMAND = "-ERR Unknown REPLCONF subcommand\r\n"; + /** + * Handles the REPLCONF command by dispatching to the appropriate subcommand handler. + * + * @param args the command arguments where args[0] is the REPLCONF subcommand + * (e.g., "LISTENING-PORT", "CAPA", "ACK", "GETACK") and subsequent + * entries are subcommand-specific parameters + * @param ctx the channel handler context for the client connection + * @return a RESP-formatted response string to send to the client (e.g. "+OK\r\n" or an error), + * or `null` when no response should be sent for the subcommand + */ @Override public String execute(List args, ChannelHandlerContext ctx) { if (args.isEmpty()) { @@ -54,8 +64,17 @@ public String execute(List args, ChannelHandlerContext ctx) { } /** - * Handles REPLCONF listening-port from replica. - * Master uses this to know the replica's listening port. + * Process the REPLCONF LISTENING-PORT subcommand from a replica. + * + * Parses the replica's listening port from args, creates or updates the corresponding + * ReplicaConnection with that port, and transitions the replica to the HANDSHAKE state. + * + * @param args command arguments where args.get(1) is the listening port + * @param ctx channel handler context for the replica connection + * @param replMgr replication manager used to lookup or register the replica + * @return {@code +OK\r\n} on success; + * {@code -ERR wrong number of arguments for 'REPLCONF' command\r\n} if args are insufficient; + * {@code -ERR invalid port number\r\n} if the port is not a valid integer */ private String handleListeningPort(List args, ChannelHandlerContext ctx, ReplicationManager replMgr) { @@ -85,8 +104,12 @@ private String handleListeningPort(List args, ChannelHandlerContext ctx, } /** - * Handles REPLCONF capa from replica. - * Replica announces its capabilities (e.g., psync2, eof). + * Process a REPLCONF CAPA subcommand and record the replica's announced capability. + * + * @param args command arguments where args.get(1) is the capability to add + * @param ctx channel handler context identifying the replica connection + * @param replMgr replication manager used to locate and update the ReplicaConnection + * @return RESP_OK after recording the capability, or ERR_WRONG_ARGS if the capability argument is missing */ private String handleCapa(List args, ChannelHandlerContext ctx, ReplicationManager replMgr) { @@ -105,8 +128,16 @@ private String handleCapa(List args, ChannelHandlerContext ctx, } /** - * Handles REPLCONF ACK from replica. - * Replica reports the number of bytes it has processed. + * Process the REPLCONF ACK subcommand from a replica. + * + * Updates the replica's acknowledged replication offset based on the offset value provided in the command. + * + * @param args command arguments where args.get(1) is the acknowledged offset in bytes + * @param ctx the channel handler context for the replica connection + * @param replMgr the replication manager controlling replica state + * @return "-ERR wrong number of arguments for 'REPLCONF' command\r\n" if arguments are missing, + * "-ERR invalid offset\r\n" if the offset cannot be parsed as a number, + * `null` on successful processing to indicate no response should be sent */ private String handleAck(List args, ChannelHandlerContext ctx, ReplicationManager replMgr) { @@ -131,8 +162,9 @@ private String handleAck(List args, ChannelHandlerContext ctx, } /** - * Handles REPLCONF GETACK from master. - * This is received by replicas; they should respond with ACK. + * Responds to a REPLCONF GETACK by returning a RESP multi-bulk containing the current acknowledged offset when running as a replica. + * + * @return the RESP multi-bulk string: ["REPLCONF","ACK",] when this node is a slave; `null` when this node is a master (no response). */ private String handleGetAck(List args, ChannelHandlerContext ctx, ReplicationManager replMgr) { @@ -147,8 +179,13 @@ private String handleGetAck(List args, ChannelHandlerContext ctx, return null; } + /** + * The identifier for this command implementation used to register and look up the command. + * + * @return the command name "REPLCONF" + */ @Override public String name() { return "REPLCONF"; } -} +} \ No newline at end of file diff --git a/src/main/java/com/redis/commands/replication/WaitCommand.java b/src/main/java/com/redis/commands/replication/WaitCommand.java index 17e5baf..041f713 100644 --- a/src/main/java/com/redis/commands/replication/WaitCommand.java +++ b/src/main/java/com/redis/commands/replication/WaitCommand.java @@ -32,6 +32,18 @@ public class WaitCommand implements ICommand { private static final String ERR_INVALID_NUM = "-ERR numreplicas is not a non-negative integer\r\n"; private static final String ERR_INVALID_TIMEOUT = "-ERR timeout is not a non-negative integer\r\n"; + /** + * Execute the WAIT command: validate arguments and wait for the specified number of replicas + * to acknowledge within the given timeout. + * + * @param args a list where args.get(0) is the required number of replicas to wait for (non-negative integer) + * and args.get(1) is the timeout in milliseconds (non-negative long) + * @return `ERR_WRONG_ARGS` if fewer than two arguments are provided, + * `ERR_INVALID_NUM` if the replicas argument is not an integer >= 0, + * `ERR_INVALID_TIMEOUT` if the timeout argument is not a long >= 0, + * otherwise a Redis integer reply of the acknowledged replica count formatted as + * ":" + acknowledged + "\r\n" + */ @Override public String execute(List args, ChannelHandlerContext ctx) { if (args.size() < 2) { @@ -77,8 +89,13 @@ public String execute(List args, ChannelHandlerContext ctx) { return ":" + acknowledged + "\r\n"; } + /** + * Get the command name handled by this ICommand implementation. + * + * @return the command name "WAIT" + */ @Override public String name() { return "WAIT"; } -} +} \ No newline at end of file diff --git a/src/main/java/com/redis/commands/stream/XAddCommand.java b/src/main/java/com/redis/commands/stream/XAddCommand.java index 492ef51..b365f7c 100644 --- a/src/main/java/com/redis/commands/stream/XAddCommand.java +++ b/src/main/java/com/redis/commands/stream/XAddCommand.java @@ -158,29 +158,35 @@ public String execute(List args, ChannelHandlerContext ctx) { return "$" + idStr.length() + "\r\n" + idStr + "\r\n"; } + /** + * Provides the Redis command name handled by this class. + * + * @return the Redis command name "XADD" + */ @Override public String name() { return "XADD"; } + /** + * Indicates that this command modifies the dataset. + * + * @return `true` if the command modifies the dataset, `false` otherwise. + */ @Override public boolean isWriteCommand() { return true; } /** - * Returns canonical arguments for replication. - *

- * For XADD, we must replace auto-generated IDs (* or timestamp-*) with - * the actual ID that was generated. This ensures replicas have the exact - * same entry ID as the master. - *

- * Example: {@code XADD stream * field value} with generated ID "123-0" - * becomes {@code XADD stream 123-0 field value} for replication. + * Produce replication-safe XADD arguments by substituting an auto-generated ID + * ("*" or "timestamp-*") with the actual ID extracted from the command response. * - * @param originalArgs The original arguments including potential * or timestamp-* - * @param response The response containing the generated ID - * @return Arguments with actual ID substituted for auto-generated ones + * @param originalArgs the original XADD arguments (key, id, followed by field/value pairs) + * @param response the RESP bulk-string response from the executed XADD command + * @return a new argument list with the actual ID substituted for the auto-generated one, + * or `null` if no substitution is needed (explicit ID provided), the input is invalid, + * or the response does not contain a parsable ID */ @Override public List getReplicationArgs(List originalArgs, String response) { diff --git a/src/main/java/com/redis/commands/string/IncrCommand.java b/src/main/java/com/redis/commands/string/IncrCommand.java index 9e63b6c..3ec4b71 100644 --- a/src/main/java/com/redis/commands/string/IncrCommand.java +++ b/src/main/java/com/redis/commands/string/IncrCommand.java @@ -74,13 +74,23 @@ public String execute(List args, ChannelHandlerContext ctx) { return result.get(); } + /** + * Returns the command name handled by this implementation. + * + * @return the literal command name "INCR" + */ @Override public String name() { return "INCR"; } + /** + * Indicates whether this command modifies the Redis dataset. + * + * @return `true` if the command modifies the dataset, `false` otherwise. + */ @Override public boolean isWriteCommand() { return true; } -} +} \ No newline at end of file diff --git a/src/main/java/com/redis/commands/string/SetCommand.java b/src/main/java/com/redis/commands/string/SetCommand.java index 94d540b..729d4be 100644 --- a/src/main/java/com/redis/commands/string/SetCommand.java +++ b/src/main/java/com/redis/commands/string/SetCommand.java @@ -37,6 +37,22 @@ public class SetCommand implements ICommand { */ private static final ThreadLocal lastComputedPxat = new ThreadLocal<>(); + /** + * Execute the Redis SET command: store a key with a value, optional expiry, and optional NX/XX modifiers. + * + * Supports the options EX , PX , EXAT , PXAT , + * NX, and XX. When a relative expiry (EX or PX) is used, the computed absolute expiration timestamp is recorded + * in thread-local storage for replication rewriting. + * + * @param args command arguments: at minimum [key, value]; additional elements are parsed as options described above + * @param ctx Netty channel context (not used for command semantics) + * @return one of the Redis protocol responses: + * `+OK\r\n` on success when the key is set; + * `$-1\r\n` when NX/XX conditions prevent a set; + * `-ERR wrong number of arguments for 'SET' command\r\n` when args are insufficient; + * `-ERR invalid expire time in set\r\n` for invalid expiry values; + * `-ERR syntax error\r\n` for unrecognized or conflicting options. + */ @Override public String execute(List args, ChannelHandlerContext ctx) { // Clear any previous value @@ -148,11 +164,21 @@ public String execute(List args, ChannelHandlerContext ctx) { return RESP_OK; } + /** + * The Redis command name handled by this implementation. + * + * @return the command name "SET" + */ @Override public String name() { return "SET"; } + /** + * Indicates that this command modifies the dataset. + * + * @return `true` if the command modifies the dataset, `false` otherwise. + */ @Override public boolean isWriteCommand() { return true; @@ -209,4 +235,4 @@ public List getReplicationArgs(List originalArgs, String respons return replicationArgs; } -} +} \ No newline at end of file diff --git a/src/main/java/com/redis/config/RedisConfig.java b/src/main/java/com/redis/config/RedisConfig.java index 794b454..c4d2362 100644 --- a/src/main/java/com/redis/config/RedisConfig.java +++ b/src/main/java/com/redis/config/RedisConfig.java @@ -37,10 +37,20 @@ public class RedisConfig { private String replicaOfHost; private Integer replicaOfPort; + /** + * Constructs the singleton RedisConfig and initializes configuration from application properties. + */ private RedisConfig() { loadProperties(); } + /** + * Retrieve the singleton RedisConfig instance. + * + * The instance is created on first access and reused for subsequent calls. + * + * @return the shared RedisConfig instance + */ public static RedisConfig getInstance() { if (instance == null) { instance = new RedisConfig(); @@ -49,13 +59,18 @@ public static RedisConfig getInstance() { } /** - * Parses command-line arguments and updates configuration. - *

- * Supported arguments: + * Parse command-line arguments and apply recognized configuration overrides. + * + *

Recognized (case-insensitive) flags: *

+ * + *

On invalid numeric values or unknown {@code --}-prefixed arguments the method will print an + * error message to standard error. If the replica port is invalid the replica host will be cleared. + * + * @param args the command-line arguments to parse */ public void parseArgs(String[] args) { for (int i = 0; i < args.length; i++) { @@ -92,6 +107,12 @@ public void parseArgs(String[] args) { } } + /** + * Loads "application.properties" from the classpath into the instance's properties. + * + * If the resource is not found, the method returns without modifying properties. + * If an I/O error occurs while reading, a warning message is printed to stderr and the method returns. + */ private void loadProperties() { try (InputStream is = getClass().getClassLoader().getResourceAsStream("application.properties")) { if (is != null) { @@ -102,6 +123,11 @@ private void loadProperties() { } } + /** + * Determine the Redis server port using the following precedence: CLI argument, REDIS_PORT environment variable, properties file value, then DEFAULT_PORT. + * + * @return the resolved port number to bind the Redis server to + */ public int getPort() { // Priority: CLI > ENV > Properties > Default if (cliPort != null) { @@ -114,6 +140,11 @@ public int getPort() { return Integer.parseInt(properties.getProperty("redis.port", String.valueOf(DEFAULT_PORT))); } + /** + * Determine the configured number of boss threads for the Redis server. + * + * @return the configured boss thread count: value of the `REDIS_BOSS_THREADS` environment variable if set, otherwise the `redis.boss.threads` property, otherwise the default value. + */ public int getBossThreads() { String envThreads = System.getenv("REDIS_BOSS_THREADS"); if (envThreads != null) { @@ -122,6 +153,11 @@ public int getBossThreads() { return Integer.parseInt(properties.getProperty("redis.boss.threads", String.valueOf(DEFAULT_BOSS_THREADS))); } + /** + * Determines the number of worker threads the server should use. + * + * @return the worker thread count resolved from (in order) the `REDIS_WORKER_THREADS` environment variable, the `redis.worker.threads` property, or the default `DEFAULT_WORKER_THREADS` + */ public int getWorkerThreads() { String envThreads = System.getenv("REDIS_WORKER_THREADS"); if (envThreads != null) { @@ -130,6 +166,13 @@ public int getWorkerThreads() { return Integer.parseInt(properties.getProperty("redis.worker.threads", String.valueOf(DEFAULT_WORKER_THREADS))); } + /** + * Get the interval, in milliseconds, used for periodic cleanup of expired entries. + * + * Resolution order: environment variable `REDIS_CLEANUP_INTERVAL_MS` → property `redis.cleanup.interval.ms` → default `DEFAULT_CLEANUP_INTERVAL_MS`. + * + * @return the cleanup interval in milliseconds + */ public int getCleanupIntervalMs() { String envInterval = System.getenv("REDIS_CLEANUP_INTERVAL_MS"); if (envInterval != null) { @@ -138,6 +181,12 @@ public int getCleanupIntervalMs() { return Integer.parseInt(properties.getProperty("redis.cleanup.interval.ms", String.valueOf(DEFAULT_CLEANUP_INTERVAL_MS))); } + /** + * Determine whether key expiry is enabled by consulting configuration sources in precedence order: + * environment variable `REDIS_EXPIRY_ENABLED`, application properties (`redis.expiry.enabled`), then the built-in default. + * + * @return `true` if expiry is enabled, `false` otherwise. + */ public boolean isExpiryEnabled() { String envExpiry = System.getenv("REDIS_EXPIRY_ENABLED"); if (envExpiry != null) { @@ -147,26 +196,38 @@ public boolean isExpiryEnabled() { } /** - * Returns true if this server should be a replica. + * Indicates whether this server is configured to act as a replica. + * + * @return `true` if both the replica host and replica port are configured, `false` otherwise. */ public boolean isReplica() { return replicaOfHost != null && replicaOfPort != null; } /** - * Gets the master host if configured as replica. + * Returns the configured master host when this instance is set up as a replica. + * + * @return the master host configured via command-line or properties, or `null` if no replica master is configured */ public String getReplicaOfHost() { return replicaOfHost; } /** - * Gets the master port if configured as replica. + * Return the configured master port when this instance is a replica. + * + * @return the master port configured via CLI or properties, or `null` if no replica port is set */ public Integer getReplicaOfPort() { return replicaOfPort; } + /** + * Render a compact single-line representation of the current Redis configuration. + * + * @return a string describing the configured port, bossThreads, workerThreads, cleanupIntervalMs, + * expiryEnabled, and, if configured, the replica master as `host:port` + */ @Override public String toString() { StringBuilder sb = new StringBuilder(); @@ -182,4 +243,4 @@ public String toString() { sb.append('}'); return sb.toString(); } -} +} \ No newline at end of file diff --git a/src/main/java/com/redis/replication/CommandPropagator.java b/src/main/java/com/redis/replication/CommandPropagator.java index 50fac53..908e5ae 100644 --- a/src/main/java/com/redis/replication/CommandPropagator.java +++ b/src/main/java/com/redis/replication/CommandPropagator.java @@ -62,31 +62,22 @@ public class CommandPropagator { ); /** - * Determines if a command should be propagated to replicas. - *

- * Only write commands that modify state need to be propagated. - * Read commands, transaction control (MULTI/EXEC), and replication - * commands (REPLCONF, PSYNC) are not propagated. + * Determines whether the given command modifies dataset state and therefore must be propagated to replicas. * - * @param commandName The uppercase command name - * @return true if the command should be propagated to replicas + * @param commandName the command name (expected in uppercase) + * @return true if the command should be propagated to replicas, false otherwise */ public static boolean shouldPropagate(String commandName) { return WRITE_COMMANDS.contains(commandName); } /** - * Propagates a command to all connected replicas. - *

- * This method converts the command and arguments to RESP format - * and sends to replicas via the ReplicationManager. - *

- * Note: For commands that need rewriting (XADD with *, SET with EX), - * use {@link #propagateRewritten(String, List)} instead to ensure - * consistent state across replicas. + * Propagates the given command and its arguments from the master to all connected replicas. * - * @param commandName The command name (e.g., "SET", "DEL") - * @param args The command arguments (not including command name) + * If the current node is not the master, this method performs no action. + * + * @param commandName the command name (e.g., "SET", "DEL") + * @param args the command arguments (excluding the command name); may contain nulls to represent bulk nils */ public static void propagate(String commandName, List args) { ReplicationManager replMgr = ReplicationManager.getInstance(); @@ -101,13 +92,10 @@ public static void propagate(String commandName, List args) { } /** - * Propagates a rewritten/canonical command to replicas. - *

- * Use this method when the command has been rewritten to its canonical form - * (e.g., XADD with actual ID instead of *, SET with PXAT instead of EX). + * Propagates a command already converted to its canonical RESP arguments to replicas. * - * @param commandName The command name - * @param rewrittenArgs The canonical arguments (already normalized) + * @param commandName the command name + * @param rewrittenArgs canonical, normalized arguments to propagate */ public static void propagateRewritten(String commandName, List rewrittenArgs) { propagate(commandName, rewrittenArgs); @@ -132,18 +120,13 @@ public static void propagateRaw(String respCommand) { } /** - * Builds a RESP Array from command name and arguments. - *

- * RESP Array format: *{count}\r\n${len}\r\n{element}\r\n... - *

- * Example: SET key value becomes: - * 

-     * *3\r\n$3\r\nSET\r\n$3\r\nkey\r\n$5\r\nvalue\r\n
-     *
+ * Constructs a RESP Array–encoded string representing the given command and its arguments. + * + * Null argument elements are encoded as RESP bulk nil entries. * - * @param commandName The command name - * @param args The command arguments - * @return RESP-encoded array string + * @param commandName the command name to encode (e.g., "SET") + * @param args the command arguments in order; elements may be null to produce bulk nil + * @return a RESP Array-encoded string for the command and arguments */ public static String buildRespArray(String commandName, List args) { int totalElements = 1 + args.size(); // command name + args @@ -170,12 +153,12 @@ public static String buildRespArray(String commandName, List args) { } /** - * Appends a bulk string to the StringBuilder. - *

- * Bulk String format: ${length}\r\n{data}\r\n + * Appends the given value to the StringBuilder as a RESP Bulk String; encodes null as a Bulk Nil. + * + * The byte length used for the bulk string header is computed from the UTF-8 encoding of {@code value}. * - * @param sb StringBuilder to append to - * @param value The string value to encode + * @param sb the StringBuilder to append to + * @param value the string to encode; if {@code null}, a RESP Bulk Nil ({@code $-1\r\n}) is appended */ private static void appendBulkString(StringBuilder sb, String value) { if (value == null) { @@ -186,4 +169,4 @@ private static void appendBulkString(StringBuilder sb, String value) { sb.append(value).append("\r\n"); } } -} +} \ No newline at end of file diff --git a/src/main/java/com/redis/replication/MasterConnection.java b/src/main/java/com/redis/replication/MasterConnection.java index cca384d..4005bba 100644 --- a/src/main/java/com/redis/replication/MasterConnection.java +++ b/src/main/java/com/redis/replication/MasterConnection.java @@ -164,12 +164,12 @@ public MasterConnection(String masterHost, int masterPort, int listeningPort) { // ==================== Connection Management ==================== /** - * Initiates asynchronous connection to the master server. + * Establishes a TCP connection to the configured master and begins the replication handshake. * - *

Upon successful connection, automatically begins the - * replication handshake sequence. + *

On successful connection the method starts the handshake sequence (PING → REPLCONF → PSYNC). * - * @return CompletableFuture that completes when connection is established + * @return a CompletableFuture that completes when the connection to the master is established, + * or completes exceptionally if the connection attempt fails */ public CompletableFuture connect() { CompletableFuture future = new CompletableFuture<>(); @@ -183,6 +183,14 @@ public CompletableFuture connect() { .option(ChannelOption.TCP_NODELAY, true) // Disable Nagle for low latency .option(ChannelOption.SO_KEEPALIVE, true) // Enable TCP keepalive .handler(new ChannelInitializer() { + /** + * Initializes a newly accepted SocketChannel's pipeline for communication with the master. + * + * Adds a MasterResponseHandler to the channel pipeline to handle inbound replication messages + * and protocol parsing. + * + * @param ch the SocketChannel whose pipeline will be configured + */ @Override protected void initChannel(SocketChannel ch) { ch.pipeline().addLast(new MasterResponseHandler()); @@ -244,6 +252,20 @@ private class MasterResponseHandler extends ByteToMessageDecoder { /** Buffer for parsing RESP array elements */ private final List argsBuffer = new ArrayList<>(); + /** + * Decodes inbound bytes from the master, handling RDB streaming and RESP responses. + * + * During each invocation this method: + * - If currently receiving an RDB, feeds bytes to the RDB handler until the RDB is complete; if more data is required it resets the reader index and returns. + * - Otherwise parses a single RESP response; if the buffer does not contain a full RESP value it resets the reader index and returns. + * - When a complete RESP response is parsed, routes the parsed arguments to the handshake/streaming response handler. + * + * The method consumes bytes from `in` as responses or RDB data are completed and may update the connection's handshake and streaming state via the response handlers. + * + * @param ctx the Netty channel handler context + * @param in the inbound byte buffer to read from; reader index may be advanced or reset when incomplete data is encountered + * @param out the list to which decoded messages would be added (not used directly by this decoder) + */ @Override protected void decode(ChannelHandlerContext ctx, ByteBuf in, List out) { while (in.readableBytes() > 0) { @@ -273,11 +295,13 @@ protected void decode(ChannelHandlerContext ctx, ByteBuf in, List out) { // ==================== RESP Parsing ==================== /** - * Parses a single RESP value from the buffer. + * Parse a single RESP value from the provided buffer and append its textual parts to result. + * + * If the buffer does not contain a complete RESP value, the reader index is not advanced and the method returns `false`. * - * @param buf Input buffer - * @param result Output list for parsed values - * @return true if complete value was parsed, false if more data needed + * @param buf the ByteBuf containing RESP-encoded data; may be advanced when a full value is parsed + * @param result list to receive parsed string parts (bulk strings, simple strings, integers, and array elements) in encounter order + * @return `true` if a complete RESP value was parsed and appended to result, `false` if more data is required */ private boolean parseRespResponse(ByteBuf buf, List result) { if (buf.readableBytes() < 1) return false; @@ -299,7 +323,11 @@ private boolean parseRespResponse(ByteBuf buf, List result) { } /** - * Parses a simple string (or error/integer) ending in CRLF. + * Parses a RESP simple string, error, or integer terminated by CRLF and appends the parsed text to the result list. + * + * @param buf the ByteBuf positioned at the start of the RESP simple value + * @param result the list to which the parsed string will be appended + * @return `true` if a complete CRLF-terminated simple value was parsed and added to {@code result}, `false` otherwise */ private boolean parseSimpleString(ByteBuf buf, List result) { int start = buf.readerIndex(); @@ -313,7 +341,13 @@ private boolean parseSimpleString(ByteBuf buf, List result) { } /** - * Parses a bulk string: $\r\n\r\n + * Parses a RESP bulk string from the buffer and appends its value to the result list. + * + * If the bulk length is `-1` (null bulk string), `null` is appended to `result`. + * + * @param buf the ByteBuf to read the bulk string from; reader index is advanced on success and reset on incomplete input + * @param result list to which the parsed string (or `null` for a null bulk) will be appended + * @return `true` if a complete bulk string was parsed and appended to `result`, `false` if more bytes are required */ private boolean parseBulkString(ByteBuf buf, List result) { int start = buf.readerIndex(); @@ -341,7 +375,16 @@ private boolean parseBulkString(ByteBuf buf, List result) { } /** - * Parses an array: *\r\n followed by elements. + * Parses a RESP array from the buffer and appends its elements (flattened) to the provided result list. + * + * The method expects an array header of the form `*\r\n` followed by RESP elements. + * If the array count is `-1`, a single `null` is added to `result`. + * On successful parse the buffer's reader index is advanced past the entire array; if data is incomplete the + * reader index is restored to its original position and the method returns `false`. + * + * @param buf the ByteBuf containing RESP data; its reader index is advanced on successful parse + * @param result the list to receive parsed string elements (array elements are flattened into this list) + * @return `true` if a complete array was parsed and appended to `result`, `false` if more data is required or parsing failed */ private boolean parseArray(ByteBuf buf, List result) { int start = buf.readerIndex(); @@ -370,7 +413,11 @@ private boolean parseArray(ByteBuf buf, List result) { } /** - * Parses an inline command (space-separated). + * Parses a CRLF-terminated inline (space-separated) command from the buffer and appends its parts to `result`. + * + * @param buf the ByteBuf to read from; parsing requires a CRLF-terminated line + * @param result destination list to which command parts (split on spaces) will be added + * @return `true` if a complete CRLF-terminated inline command was read and added to `result`, `false` if more bytes are needed */ private boolean parseInlineCommand(ByteBuf buf, List result) { int start = buf.readerIndex(); @@ -406,6 +453,14 @@ private void handleResponse(List response) { } } + /** + * Handle the master's response to the initial PING during the replication handshake. + * + * If the response equals "PONG" (case-insensitive), advance the handshake state to + * REPLCONF_PORT_SENT and send a REPLCONF listening-port command with the local listening port. + * + * @param response the master's reply to PING (expected value: "PONG") + */ private void handlePingResponse(String response) { if ("PONG".equalsIgnoreCase(response)) { System.out.println("[Replication] Master responded to PING"); @@ -414,6 +469,14 @@ private void handlePingResponse(String response) { } } + /** + * Processes the master's reply to the REPLCONF listening-port command. + * + * If the master responds with "OK" (case-insensitive), advances the handshake to + * REPLCONF_CAPA_SENT and sends a REPLCONF capabilities message requesting PSYNC2. + * + * @param response the master's textual response to the REPLCONF listening-port request + */ private void handleReplconfPortResponse(String response) { if ("OK".equalsIgnoreCase(response)) { System.out.println("[Replication] REPLCONF listening-port acknowledged"); @@ -422,6 +485,14 @@ private void handleReplconfPortResponse(String response) { } } + /** + * Handles the server's response to the REPLCONF CAPA command. + * + * If the response equals "OK" (case-insensitive), advances the handshake to + * request PSYNC from the master. + * + * @param response the server reply to the REPLCONF CAPA command + */ private void handleReplconfCapaResponse(String response) { if ("OK".equalsIgnoreCase(response)) { System.out.println("[Replication] REPLCONF capa acknowledged"); @@ -430,6 +501,16 @@ private void handleReplconfCapaResponse(String response) { } } + /** + * Process the master's PSYNC reply and advance the handshake and RDB reception state. + * + *

If the response starts with {@code FULLRESYNC }, the method extracts the + * replication ID and offset, records the master's replication offset, and transitions to RDB + * loading mode (sets {@code loadingRdb=true} and {@code expectedRdbSize=-1}). If the response + * starts with {@code CONTINUE}, the method transitions to streaming mode and disables RDB loading. + * + * @param response the raw PSYNC reply line received from the master (e.g. {@code "FULLRESYNC "} or {@code "CONTINUE"}) + */ private void handlePsyncResponse(String response) { if (response.startsWith("FULLRESYNC")) { // Parse: FULLRESYNC @@ -454,10 +535,15 @@ private void handlePsyncResponse(String response) { // ==================== RDB Handling ==================== /** - * Handles incoming RDB file data. + * Accumulates RDB bytes from the provided input buffer until the full RDB file has been received. * - * @param in Input buffer - * @return true if RDB is complete, false if more data needed + * When the total size is not yet known this method reads the Redis bulk-string size marker and + * initializes an internal buffer for the expected RDB size. It appends available bytes into the + * buffer and, once the expected size is reached, processes the RDB, releases the buffer, clears + * the RDB-loading flag, and transitions the handshake state to STREAMING. + * + * @param in the incoming ByteBuf containing RDB data (may contain partial or multiple chunks) + * @return `true` if the full RDB file has been received and processed, `false` if additional data is required */ private boolean handleRdbData(ByteBuf in) { // First, read the RDB size if not yet known @@ -508,10 +594,11 @@ private boolean handleRdbData(ByteBuf in) { } /** - * Processes the received RDB file. - *

- * For now, just acknowledges receipt. A full implementation - * would parse the RDB format and restore database state. + * Handle a complete RDB file received from the master. + * + *

Currently this implementation only acknowledges receipt and does not restore database state. + * + * @param rdb the ByteBuf containing the complete RDB file bytes */ private void processRdbFile(ByteBuf rdb) { System.out.println("[Replication] RDB processing complete (empty database mode)"); @@ -520,7 +607,14 @@ private void processRdbFile(ByteBuf rdb) { // ==================== Streaming Command Handling ==================== /** - * Handles replicated commands from the master. + * Routes and handles a single replicated command received from the master. + * + * If the command is empty the method returns immediately. If the command is + * "REPLCONF GETACK" the method replies to the master with "REPLCONF ACK " + * using the current replication byte offset; otherwise the command is applied + * locally via executeReplicatedCommand. + * + * @param command a list of RESP-parsed tokens where the first element is the command name */ private void handleStreamingCommand(List command) { if (command.isEmpty()) return; @@ -540,10 +634,18 @@ private void handleStreamingCommand(List command) { } /** - * Executes a replicated write command on the local database. - *

- * This is a simplified implementation. A production version - * would use the command registry for consistency. + * Apply a replicated write command to the local RedisDatabase. + * + *

Handles a minimal set of replication write commands received from the master. + * Supported commands: + *

    + *
  • SET key value [PX milliseconds | EX seconds] — stores a value with optional expiry
    • + *
  • DEL key [key ...] — deletes one or more keys
    • + *
  • INCR key — increments the numeric value of a key (creates key with 0 before increment)
    • + *
+ * Unsupported or malformed commands are ignored. + * + * @param command a List of strings where the first element is the command name and the remaining elements are its arguments (e.g., [\"SET\", \"key\", \"value\"]) */ private void executeReplicatedCommand(List command) { if (command.isEmpty()) return; @@ -585,7 +687,14 @@ private void executeReplicatedCommand(List command) { } } - // ==================== Error Handling ==================== + /** + * Handles exceptions raised by the Netty pipeline for the master connection. + * + * Logs the error, closes the channel context, and marks the connection as not connected. + * + * @param ctx the Netty channel handler context where the exception occurred + * @param cause the exception that was thrown + */ @Override public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) { @@ -595,6 +704,11 @@ public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) { connected = false; } + /** + * Handle the channel becoming inactive by marking the master connection as disconnected and logging the closure. + * + * @param ctx the Netty ChannelHandlerContext for the closed channel + */ @Override public void channelInactive(ChannelHandlerContext ctx) { System.out.println("[Replication] Master connection closed"); @@ -605,7 +719,10 @@ public void channelInactive(ChannelHandlerContext ctx) { // ==================== Lifecycle Management ==================== /** - * Disconnects from the master and releases resources. + * Close the replication connection and release associated network resources. + * + * Marks this connection as disconnected, closes the active Netty channel if present, + * and shuts down the worker event loop group. */ public void disconnect() { connected = false; @@ -617,21 +734,40 @@ public void disconnect() { } } - // ==================== Status Methods ==================== + /** + * Indicates whether the connection to the master is currently active. + * + * @return `true` if the client has an active channel to the master, `false` otherwise. + */ public boolean isConnected() { return connected && channel != null && channel.isActive(); } + /** + * Current handshake state of the connection to the master. + * + * @return the current HandshakeState representing replication handshake progress + */ public HandshakeState getHandshakeState() { return handshakeState; } + /** + * Get the total number of bytes processed from the replication stream. + * + * @return the total number of bytes processed from the replication stream + */ public long getBytesProcessed() { return bytesProcessed.get(); } + /** + * Atomically increments the running total of replication bytes processed. + * + * @param bytes the number of bytes to add to the processed counter (in bytes); can be negative to decrement the counter + */ public void addBytesProcessed(long bytes) { bytesProcessed.addAndGet(bytes); } -} +} \ No newline at end of file diff --git a/src/main/java/com/redis/replication/RdbGenerator.java b/src/main/java/com/redis/replication/RdbGenerator.java index b5aa6b0..ac15eda 100644 --- a/src/main/java/com/redis/replication/RdbGenerator.java +++ b/src/main/java/com/redis/replication/RdbGenerator.java @@ -86,12 +86,12 @@ public class RdbGenerator { // ==================== Public API ==================== /** - * Generates an empty RDB file for full resync. - *

- * This is used when a replica connects and needs initial data. - * For servers with no data, this returns a minimal valid RDB. + * Produce a minimal valid RDB file suitable for a full resynchronization when the server has no data. + * + * The RDB includes the magic header, version, two auxiliary fields (`redis-ver` and `redis-bits`), an EOF marker, + * and an 8-byte zero CRC64 placeholder. * - * @return Byte array containing the RDB file + * @return a byte array containing the generated RDB file */ public static byte[] generateEmptyRdb() { try { @@ -132,14 +132,12 @@ public static byte[] generateEmptyRdb() { } /** - * Gets the RDB file wrapped in RESP bulk string format for transfer. - *

- * Format: $<length>\r\n<rdb-bytes> - *

- * Note: The RDB transfer format does NOT include trailing \r\n after - * the RDB data (unlike normal bulk strings). + * Wraps the generated RDB bytes with a RESP bulk-string header for transfer. * - * @return Byte array ready for network transmission + *

Format: `$<length>\r\n<rdb-bytes>` (no trailing `\r\n` after the RDB data). + * + * @return the RESP bulk-string representation: a literal-length header (`$\r\n`) + * followed immediately by the RDB bytes, without a trailing CRLF */ public static byte[] getRdbTransferFormat() { byte[] rdb = generateEmptyRdb(); @@ -157,11 +155,11 @@ public static byte[] getRdbTransferFormat() { // ==================== Internal Helpers ==================== /** - * Writes a length-prefixed string in RDB format. + * Writes a string to the output stream prefixed by its length using RDB variable-length encoding. * - * @param out Output stream - * @param s String to write - * @throws IOException If write fails + * @param out the output stream to write to + * @param s the string to write + * @throws IOException if an I/O error occurs while writing */ private static void writeString(ByteArrayOutputStream out, String s) throws IOException { byte[] bytes = s.getBytes(); @@ -200,4 +198,4 @@ private static void writeLength(ByteArrayOutputStream out, int length) throws IO out.write(length & 0xFF); } } -} +} \ No newline at end of file diff --git a/src/main/java/com/redis/replication/ReplicaConnection.java b/src/main/java/com/redis/replication/ReplicaConnection.java index 5589f25..0043824 100644 --- a/src/main/java/com/redis/replication/ReplicaConnection.java +++ b/src/main/java/com/redis/replication/ReplicaConnection.java @@ -139,11 +139,14 @@ public enum ReplicaState { // ==================== Constructor ==================== /** - * Creates a new replica connection tracker. + * Create a tracker for a replica connection and initialize its replication state. * - * @param channel The Netty channel to the replica - * @param host The replica's hostname/IP - * @param port The replica's port + *

Initializes the connection state to CONNECTING, acknowledged and expected offsets to 0, + * capabilities to an empty list, and listening port to -1.

+ * + * @param channel the Netty channel to the replica + * @param host the replica's hostname or IP address + * @param port the replica's port number */ public ReplicaConnection(Channel channel, String host, int port) { this.channel = channel; @@ -159,17 +162,14 @@ public ReplicaConnection(Channel channel, String host, int port) { // ==================== Command Propagation ==================== /** - * Propagates a command to this replica using zero-copy ByteBuf. + * Sends a RESP-encoded command to the replica and increments the connection's expected offset. * - *

Performance Notes: - *

    - *
  • Uses Unpooled.wrappedBuffer for zero-copy
    • - *
  • Updates expectedOffset before write (conservative)
    • - *
  • Non-blocking write via Netty's event loop
    • - *
+ *

If the underlying channel is not active or the replica is not in the STREAMING state, the + * method does nothing and returns {@code false}. When a write is attempted, the expected offset + * is incremented by the command byte length before sending. * - * @param respCommand The RESP-encoded command bytes - * @return true if write was attempted, false if channel not ready + * @param respCommand the RESP-encoded command bytes to send + * @return {@code true} if a write was attempted, {@code false} if the channel was not ready or the replica was not streaming */ public boolean propagateCommand(byte[] respCommand) { // Guard: only propagate to active, streaming replicas @@ -188,11 +188,10 @@ public boolean propagateCommand(byte[] respCommand) { } /** - * Propagates a command (string version). - * Convenience method that converts to UTF-8 bytes. + * Propagates a RESP-encoded command to the replica. * - * @param respCommand The RESP-encoded command string - * @return true if write was attempted + * @param respCommand RESP-encoded command as a UTF-8 string + * @return `true` if a write was attempted to the replica, `false` otherwise */ public boolean propagateCommand(String respCommand) { return propagateCommand(respCommand.getBytes(StandardCharsets.UTF_8)); @@ -201,10 +200,9 @@ public boolean propagateCommand(String respCommand) { // ==================== Offset Management ==================== /** - * Updates the acknowledged offset from REPLCONF ACK. - * Called when replica reports bytes it has processed. + * Record the replica's acknowledged replication offset reported via REPLCONF ACK. * - * @param offset The offset value from REPLCONF ACK + * @param offset the replica's acknowledged byte offset; callers should provide a value greater than or equal to the previous acknowledgement */ public void updateAcknowledgedOffset(long offset) { // Use set() for simplicity; ACKs should be monotonically increasing @@ -223,9 +221,9 @@ public boolean hasAcknowledged(long offset) { } /** - * Calculates the replication lag (bytes not yet acknowledged). + * Returns the number of bytes sent to the replica that have not yet been acknowledged. * - * @return Bytes sent but not yet acknowledged (always >= 0) + * @return the number of unacknowledged bytes (zero or greater) */ public long getReplicationLag() { return Math.max(0, expectedOffset.get() - acknowledgedOffset.get()); @@ -234,9 +232,9 @@ public long getReplicationLag() { // ==================== State Management ==================== /** - * Updates the replica's state in the state machine. + * Set the replica's lifecycle state. * - * @param state The new state + * @param state the new lifecycle state for this replica */ public void setState(ReplicaState state) { this.state = state; @@ -254,19 +252,21 @@ public ReplicaState getState() { // ==================== Capability Management ==================== /** - * Adds a capability announced by the replica. + * Record a capability announced by the replica in the negotiated capabilities list. + * + * This method is safe to call concurrently. * - * @param capability The capability name (e.g., "psync2") + * @param capability the capability name (e.g., "psync2") */ public void addCapability(String capability) { capabilities.add(capability); } /** - * Checks if replica has a specific capability. + * Determines if the replica announced the given capability. * - * @param capability The capability to check - * @return true if replica announced this capability + * @param capability capability identifier to check + * @return `true` if the replica announced this capability, `false` otherwise */ public boolean hasCapability(String capability) { return capabilities.contains(capability); @@ -284,33 +284,56 @@ public void setListeningPort(int port) { } /** - * Gets the replica's listening port. - * Returns -1 if not yet reported. + * Get the replica's listening port. * - * @return Listening port or -1 + * @return the listening port, or -1 if the replica has not reported a listening port */ public int getListeningPort() { return listeningPort; } - // ==================== Getters ==================== + /** + * Netty channel used to communicate with the replica. + * + * @return the Netty Channel for this replica connection + */ public Channel getChannel() { return channel; } + /** + * Host name or IP address of the replica. + * + * @return the replica's host name or IP address + */ public String getHost() { return host; } + /** + * Get the configured remote port for this replica connection. + * + * @return the configured remote port (the port provided at construction); note this is not the replica-reported listening port returned by {@code getListeningPort()}. + */ public int getPort() { return port; } + /** + * Gets the latest byte offset the replica has acknowledged processing. + * + * @return the latest acknowledged byte offset from the replica (0 if none reported) + */ public long getAcknowledgedOffset() { return acknowledgedOffset.get(); } + /** + * Get the total number of bytes sent to the replica that the master expects to be acknowledged. + * + * @return the total number of bytes sent to the replica and awaiting acknowledgment + */ public long getExpectedOffset() { return expectedOffset.get(); } @@ -327,11 +350,17 @@ public void close() { } } - // ==================== Object Methods ==================== + /** + * Human-readable representation of the replica connection including host, effective port, + * state, and current replication lag. + * + * @return a string containing the replica's host, effective port (listening port if reported, + * otherwise configured port), current state, and replication lag in bytes + */ @Override public String toString() { return String.format("ReplicaConnection{host=%s, port=%d, state=%s, lag=%d}", host, listeningPort > 0 ? listeningPort : port, state, getReplicationLag()); } -} +} \ No newline at end of file diff --git a/src/main/java/com/redis/replication/ReplicationLog.java b/src/main/java/com/redis/replication/ReplicationLog.java index 32aa48d..04cc9d8 100644 --- a/src/main/java/com/redis/replication/ReplicationLog.java +++ b/src/main/java/com/redis/replication/ReplicationLog.java @@ -214,12 +214,11 @@ public boolean canPartialResync(long fromOffset) { } /** - * Gets backlog data from the specified offset. - *

- * Thread-safe for concurrent reads during partial resync. + * Retrieve backlog bytes starting at the given global offset for partial resynchronization. + * This method is safe for concurrent readers. * - * @param fromOffset The starting offset - * @return The data bytes, or null if offset is not available + * @param fromOffset the global offset to start reading from + * @return the bytes from {@code fromOffset} up to the current global offset; {@code null} if the requested offset is not available for partial resynchronization; an empty array if {@code fromOffset} equals the current global offset */ public byte[] getDataFrom(long fromOffset) { rwLock.readLock().lock(); @@ -259,9 +258,9 @@ public byte[] getDataFrom(long fromOffset) { // ==================== Offset Accessors ==================== /** - * Gets the current global replication offset. + * Retrieves the current global replication offset. * - * @return Total bytes written to the replication stream + * @return the total bytes written to the replication stream */ public long getGlobalOffset() { return globalOffset.get(); @@ -288,9 +287,9 @@ public void setGlobalOffset(long offset) { // ==================== State Accessors ==================== /** - * Checks if the backlog has any data. + * Indicates whether the backlog has ever been written to. * - * @return true if data has been written + * @return `true` if data has been written, `false` otherwise. */ public boolean isActive() { return active; @@ -306,9 +305,9 @@ public int getBufferSize() { } /** - * Gets the amount of data currently in the buffer. + * Number of bytes currently stored in the replication backlog. * - * @return Bytes of valid data (0 to bufferSize) + * @return the number of valid bytes available for reads; between 0 and bufferSize inclusive */ public long getHistoryLength() { long offset = globalOffset.get(); @@ -319,18 +318,18 @@ public long getHistoryLength() { // ==================== Statistics ==================== /** - * Gets total bytes ever written to the backlog. + * Retrieve the cumulative number of bytes written to the backlog. * - * @return Cumulative byte count + * @return the cumulative number of bytes written */ public long getTotalBytesWritten() { return totalBytesWritten.get(); } /** - * Gets the number of eviction events. + * Number of times data was evicted (overwritten) from the ring buffer due to capacity or wraparound. * - * @return Eviction count + * @return the total eviction count */ public long getEvictionCount() { return evictionCount.get(); @@ -339,10 +338,12 @@ public long getEvictionCount() { // ==================== Reset ==================== /** - * Resets the backlog to initial state. - *

- * Warning: This should only be called during server restart - * or testing. Active replicas will need full resync after reset. + * Reset the replication backlog to its initial empty state. + * + *

This marks the backlog inactive, clears global and first-available offsets and the write position, + * and resets statistics (total bytes written and eviction count). Active replicas will require a full + * resynchronization after this operation. This method acquires the write lock to perform the reset + * atomically and thread-safely; it should only be invoked during server restart or testing. */ public void reset() { rwLock.writeLock().lock(); @@ -358,11 +359,15 @@ public void reset() { } } - // ==================== Debug ==================== + /** + * Produce a concise diagnostic string describing the replication log's current state. + * + * @return a formatted string containing the buffer size, global offset, first available offset, history length, and active flag + */ @Override public String toString() { return String.format("ReplicationLog{size=%d, offset=%d, firstAvailable=%d, histLen=%d, active=%s}", bufferSize, globalOffset.get(), firstAvailableOffset.get(), getHistoryLength(), active); } -} +} \ No newline at end of file diff --git a/src/main/java/com/redis/replication/ReplicationManager.java b/src/main/java/com/redis/replication/ReplicationManager.java index 11dfb41..13b2491 100644 --- a/src/main/java/com/redis/replication/ReplicationManager.java +++ b/src/main/java/com/redis/replication/ReplicationManager.java @@ -179,7 +179,14 @@ public class ReplicationManager { /** Shutdown flag for graceful termination */ private volatile boolean shuttingDown; - // ==================== Constructor ==================== + /** + * Initialize a new ReplicationManager with default role, replication identifiers, and + * all internal data structures required for replication management. + * + *

Sets the node role to MASTER, generates the primary replication ID, initializes the + * replication log and replica registry, and creates counters and health-tracking maps + * used for propagation, backlog, and circuit-breaker logic.

+ */ private ReplicationManager() { // Identity @@ -211,6 +218,13 @@ private ReplicationManager() { this.shuttingDown = false; } + /** + * Lazily obtains the globally shared ReplicationManager singleton. + * + * The instance is created on first access and is safe for concurrent use by multiple threads. + * + * @return the shared ReplicationManager singleton instance + */ public static ReplicationManager getInstance() { ReplicationManager instance = INSTANCE; if (instance == null) { @@ -224,7 +238,11 @@ public static ReplicationManager getInstance() { return instance; } - // ==================== Replication ID Generation ==================== + /** + * Generates a cryptographically secure 40-character hexadecimal replication identifier. + * + * @return a 40-character string containing lowercase hexadecimal characters (0-9, a-f) + */ private String generateReplicationId() { SecureRandom random = new SecureRandom(); @@ -236,25 +254,51 @@ private String generateReplicationId() { return sb.toString(); } - // ==================== Role Management ==================== + /** + * Set the server's replication role. + * + * @param newRole the server role to assign (e.g., MASTER or SLAVE) + */ public void setRole(ServerRole newRole) { this.role.set(newRole); } + /** + * Get the current server role of this node. + * + * @return the current {@link ServerRole} indicating whether the node is MASTER or SLAVE + */ public ServerRole getRole() { return role.get(); } + /** + * Determine whether this node currently holds the MASTER role. + * + * @return `true` if the node's role is MASTER, `false` otherwise. + */ public boolean isMaster() { return role.get() == ServerRole.MASTER; } + /** + * Checks whether this node is currently operating in the SLAVE role. + * + * @return `true` if the server role is SLAVE, `false` otherwise. + */ public boolean isSlave() { return role.get() == ServerRole.SLAVE; } - // ==================== Master Info (Slave Mode) ==================== + /** + * Configure the node with the master's address and mark this node as a SLAVE. + * + * Sets the master's host and port used for replication and updates the server role to SLAVE. + * + * @param host the master's hostname or IP address + * @param port the master's TCP port + */ public void setMasterInfo(String host, int port) { this.masterHost = host; @@ -262,23 +306,50 @@ public void setMasterInfo(String host, int port) { this.role.set(ServerRole.SLAVE); } + /** + * The configured master host for this node. + * + * @return the master host, or null if no master is configured + */ public String getMasterHost() { return masterHost; } + /** + * Get the configured master TCP port. + * + * @return the configured master's port number + */ public int getMasterPort() { return masterPort; } + /** + * Set the active master connection for this replication manager. + * + * @param connection the MasterConnection to associate with this manager; may be {@code null} to clear the active connection + */ public void setMasterConnection(MasterConnection connection) { this.masterConnection = connection; } + /** + * Get the currently configured master connection. + * + * @return the active MasterConnection instance, or null if no master connection is set + */ public MasterConnection getMasterConnection() { return masterConnection; } - // ==================== Replica Management (Master Mode) ==================== + /** + * Register a new replica connection and initialize its per-replica health tracking. + * + * @param channel the network channel for the replica + * @param host the replica's host address + * @param port the replica's port number + * @return the created and registered ReplicaConnection + */ public ReplicaConnection addReplica(Channel channel, String host, int port) { ReplicaConnection replica = new ReplicaConnection(channel, host, port); @@ -289,10 +360,24 @@ public ReplicaConnection addReplica(Channel channel, String host, int port) { return replica; } + /** + * Retrieve the replica registration associated with a Netty channel. + * + * @param channel the channel used as the key for the replica + * @return the ReplicaConnection for the given channel, or {@code null} if no replica is registered for that channel + */ public ReplicaConnection getReplica(Channel channel) { return replicas.get(channel); } + /** + * Unregisters the replica associated with the given channel and clears its health tracking. + * + * Removes any replica state tied to the provided Channel (replica registry, failure counts, + * and circuit-breaker trip time). If a replica was removed, logs a disconnection message. + * + * @param channel the channel identifying the replica to remove + */ public void removeReplica(Channel channel) { ReplicaConnection removed = replicas.remove(channel); replicaFailureCounts.remove(channel); @@ -302,10 +387,21 @@ public void removeReplica(Channel channel) { } } + /** + * Retrieves a collection view of all registered replica connections. + * + * @return a collection view of the current {@link ReplicaConnection} instances; the collection is backed by + * the internal registry so changes to the registry are reflected in this collection + */ public Collection getReplicas() { return replicas.values(); } + /** + * Counts replicas that are currently in the STREAMING state. + * + * @return the number of replicas whose state is `ReplicaConnection.ReplicaState.STREAMING` + */ public int getConnectedReplicaCount() { int count = 0; for (ReplicaConnection r : replicas.values()) { @@ -319,8 +415,13 @@ public int getConnectedReplicaCount() { // ==================== Circuit Breaker ==================== /** - * Checks if the circuit breaker is open (tripped) for a replica. - * Uses time-based recovery for self-healing. + * Determine whether the per-replica circuit breaker is currently open. + * + * If the breaker has been tripped but its recovery window has elapsed, this method resets the breaker + * (clears the trip timestamp and per-replica failure count) and returns `false`. + * + * @param channel the replica's Channel used as the circuit-breaker key + * @return `true` if the circuit breaker is currently open for the given channel, `false` otherwise */ private boolean isCircuitBreakerOpen(Channel channel) { AtomicLong tripTime = circuitBreakerTripTimes.get(channel); @@ -341,8 +442,11 @@ private boolean isCircuitBreakerOpen(Channel channel) { } /** - * Records a failure for circuit breaker tracking. - * Trips the breaker after threshold failures. + * Records a propagation failure for the replica identified by the given channel and trips its circuit breaker once failures reach the configured threshold. + * + * If the threshold is reached for the first time, marks the breaker as open by recording the current timestamp. Also increments the global propagation-failure counter. + * + * @param channel the replica channel for which to record the failure */ private void recordReplicaFailure(Channel channel) { AtomicInteger failures = replicaFailureCounts.get(channel); @@ -372,13 +476,14 @@ private void recordReplicaSuccess(Channel channel) { // ==================== Command Propagation ==================== /** - * Propagates a command to all streaming replicas with advanced features: - *
    - *
  • Circuit breaker protection per replica
    • - *
  • Backpressure detection and handling
    • - *
  • Zero-copy ByteBuf writes
    • - *
  • Comprehensive statistics tracking
    • - *
+ * Propagates a RESP-encoded command to all connected replicas in STREAMING state and updates + * replication state, per-replica health, and propagation statistics. + * + * The method appends the command to the replication backlog, advances the master replication + * offset, attempts delivery to each streaming replica, and records successes, failures, + * backpressure events, and circuit-breaker skips. + * + * @param respCommand the command encoded as a RESP byte array to send to replicas */ public void propagateToReplicas(byte[] respCommand) { if (!isMaster() || replicas.isEmpty() || shuttingDown) { @@ -447,11 +552,22 @@ public void propagateToReplicas(byte[] respCommand) { } } + /** + * Propagates a RESP-formatted command string to all connected replicas after encoding it as UTF-8. + * + * @param respCommand the RESP-formatted command to send to replicas + */ public void propagateToReplicas(String respCommand) { propagateToReplicas(respCommand.getBytes(java.nio.charset.StandardCharsets.UTF_8)); } - // ==================== Replication Backlog ==================== + /** + * Determines whether a partial resynchronization can be performed for the given replication id and offset. + * + * @param requestedReplId the replication ID presented by the replica requesting partial resync + * @param requestedOffset the replication offset from which the replica requests backlog data + * @return `true` if the requested replication ID matches one of the master's current IDs and the backlog can serve the requested offset, `false` otherwise + */ public boolean canPartialResync(String requestedReplId, long requestedOffset) { if (!masterReplId.equals(requestedReplId) && !masterReplId2.get().equals(requestedReplId)) { @@ -460,6 +576,12 @@ public boolean canPartialResync(String requestedReplId, long requestedOffset) { return replicationLog.canPartialResync(requestedOffset); } + /** + * Retrieve backlog data starting at a specified replication offset for partial resynchronization. + * + * @param fromOffset the replication offset (inclusive) to read backlog data from + * @return a byte array containing backlog bytes beginning at `fromOffset`, or `null` if partial resynchronization is not possible from that offset + */ public byte[] getBacklogData(long fromOffset) { if (!replicationLog.canPartialResync(fromOffset)) { return null; @@ -470,8 +592,17 @@ public byte[] getBacklogData(long fromOffset) { // ==================== WAIT Implementation ==================== /** - * Waits for replicas to acknowledge with adaptive exponential backoff. - * More efficient than Redis's fixed polling interval. + * Waits until at least {@code numReplicas} replicas have acknowledged the current master replication offset, + * using an adaptive exponential backoff while polling for acknowledgments. + * + * Requests acknowledgments from STREAMING replicas and returns as soon as the required number have acknowledged + * or the timeout elapses. + * + * @param numReplicas the number of replica acknowledgments required + * @param timeoutMs the maximum time to wait in milliseconds + * @return the number of replicas that have acknowledged the master offset when the method returns; + * returns 0 if not running as master or there are no replicas; if the master offset is zero returns + * the current count of connected replicas */ public int waitForReplicas(int numReplicas, long timeoutMs) { if (!isMaster() || replicas.isEmpty()) { @@ -511,6 +642,11 @@ public int waitForReplicas(int numReplicas, long timeoutMs) { return countAcknowledgedReplicas(targetOffset); } + /** + * Sends a REPLCONF GETACK request to all replicas currently in STREAMING state. + * + * This asks each streaming replica to report its replication acknowledgment offset. + */ private void requestAckFromReplicas() { byte[] cmd = "*3\r\n$8\r\nREPLCONF\r\n$6\r\nGETACK\r\n$1\r\n*\r\n" .getBytes(java.nio.charset.StandardCharsets.UTF_8); @@ -523,6 +659,12 @@ private void requestAckFromReplicas() { } } + /** + * Count replicas in STREAMING state that have acknowledged the given replication offset. + * + * @param targetOffset the replication offset to check acknowledgments against + * @return the number of replicas in STREAMING state that have acknowledged at least {@code targetOffset} + */ private int countAcknowledgedReplicas(long targetOffset) { int count = 0; for (ReplicaConnection replica : replicas.values()) { @@ -534,33 +676,130 @@ private int countAcknowledgedReplicas(long targetOffset) { return count; } - // ==================== Accessors ==================== + /** + * Returns the primary replication ID used to identify this master instance. + * + * @return the 40-character hexadecimal master replication ID + */ public String getMasterReplId() { return masterReplId; } - public String getMasterReplId2() { return masterReplId2.get(); } - public long getMasterReplOffset() { return masterReplOffset.get(); } - public void setMasterReplOffset(long offset) { masterReplOffset.set(offset); } - public void incrementMasterReplOffset(long bytes) { masterReplOffset.addAndGet(bytes); } + /** + * Get the secondary replication ID used for PSYNC2. + * + * @return the secondary replication ID as a 40-character hexadecimal string, or {@code null} if not set + */ +public String getMasterReplId2() { return masterReplId2.get(); } + /** + * Gets the current master replication offset. + * + * @return the current master replication offset + */ +public long getMasterReplOffset() { return masterReplOffset.get(); } + /** + * Update the master replication offset used to track replication progress and backlog state. + * + *

This value represents the absolute byte offset of the master's replication stream and + * is used for progress tracking, backlog window calculations, and determining partial resynchronization eligibility. + * + * @param offset the new master replication offset in bytes (absolute offset since replication start) + */ +public void setMasterReplOffset(long offset) { masterReplOffset.set(offset); } + /** + * Adds the specified number of bytes to the master's replication offset. + * + * @param bytes the number of bytes to add to the master's replication offset; may be negative to decrement the offset + */ +public void incrementMasterReplOffset(long bytes) { masterReplOffset.addAndGet(bytes); } - public ReplicationLog getReplicationLog() { return replicationLog; } - public boolean isBacklogActive() { return replicationLog.isActive(); } - public int getBacklogSize() { return replicationLog.getBufferSize(); } - public long getBacklogFirstOffset() { return replicationLog.getFirstAvailableOffset(); } + /** + * Accesses the replication backlog and history manager used for partial resynchronization and backlog operations. + * + * @return the ReplicationLog managing backlog data and partial-resync support + */ +public ReplicationLog getReplicationLog() { return replicationLog; } + /** + * Check whether the replication backlog is currently active. + * + * @return `true` if the replication backlog is active, `false` otherwise. + */ +public boolean isBacklogActive() { return replicationLog.isActive(); } + /** + * Get current size of the replication backlog buffer. + * + * @return the current backlog size in bytes + */ +public int getBacklogSize() { return replicationLog.getBufferSize(); } + /** + * Returns the offset of the earliest byte retained in the replication backlog. + * + * @return the first available backlog offset (the base offset from which backlog data can be read) + */ +public long getBacklogFirstOffset() { return replicationLog.getFirstAvailableOffset(); } - // ==================== Statistics ==================== + /** + * Get the total number of commands that have been propagated to replicas. + * + * @return the total number of propagated commands + */ public long getCommandsPropagated() { return commandsPropagated.sum(); } - public long getBytesPropagated() { return bytesPropagated.sum(); } - public long getPartialResyncs() { return partialResyncs.sum(); } - public long getFullResyncs() { return fullResyncs.sum(); } - public long getPropagationFailures() { return propagationFailures.sum(); } - public long getBackpressureEvents() { return backpressureEvents.sum(); } - public int getReplicasWithBackpressure() { return replicasWithBackpressure.get(); } + /** + * Retrieve the cumulative number of bytes propagated to replicas. + * + * @return the total number of bytes that have been propagated to replicas + */ +public long getBytesPropagated() { return bytesPropagated.sum(); } + /** + * Get the total number of successful partial resynchronizations performed. + * + * @return the total count of successful partial resynchronizations. + */ +public long getPartialResyncs() { return partialResyncs.sum(); } + /** + * Report the total number of full resynchronizations performed. + * + * @return the total count of full resynchronizations recorded + */ +public long getFullResyncs() { return fullResyncs.sum(); } + /** + * Total number of propagation failures recorded by the replication manager. + * + * @return the total count of replication propagation failures + */ +public long getPropagationFailures() { return propagationFailures.sum(); } + /** + * Number of backpressure events recorded. + * + * @return the cumulative count of times replicas were skipped due to backpressure + */ +public long getBackpressureEvents() { return backpressureEvents.sum(); } + /** + * Reports how many connected replicas are currently experiencing backpressure. + * + * @return the number of replicas currently experiencing backpressure + */ +public int getReplicasWithBackpressure() { return replicasWithBackpressure.get(); } - public void incrementPartialResyncs() { partialResyncs.increment(); } - public void incrementFullResyncs() { fullResyncs.increment(); } + /** + * Increment the recorded count of successful partial resynchronizations by one. + */ +public void incrementPartialResyncs() { partialResyncs.increment(); } + /** + * Record a completed full resynchronization by incrementing the full-resync counter. + */ +public void incrementFullResyncs() { fullResyncs.increment(); } - // ==================== INFO Output ==================== + /** + * Builds an INFO-style replication status block reflecting the current replication role, + * connections, backlog state, and propagation statistics. + * + * For master role the block includes connected_slaves, per-slave `slaveN` lines (ip,port,state,offset,lag), + * master replication IDs and offsets, backlog metrics, and enhanced propagation statistics. + * For slave role the block includes master_host, master_port, master_link_status, master_replid, + * and master_repl_offset. + * + * @return the replication INFO block as a CRLF-separated string of key:value lines suitable for monitoring or admin output. + */ public String getInfoReplication() { StringBuilder sb = new StringBuilder(2048); @@ -614,7 +853,12 @@ public String getInfoReplication() { return sb.toString(); } - // ==================== Lifecycle ==================== + /** + * Shuts down the replication manager, closing all replica connections and clearing replication state. + * + * Sets the shutdown flag, closes each registered ReplicaConnection, removes them from the registry, + * and clears per-replica failure counters and circuit-breaker timestamps. + */ public void shutdown() { shuttingDown = true; @@ -627,6 +871,11 @@ public void shutdown() { circuitBreakerTripTimes.clear(); } + /** + * Shuts down the current ReplicationManager instance (if any) and clears the singleton so a fresh instance can be created. + * + * This method is thread-safe; it acquires the initialization lock before invoking shutdown on the existing instance and resetting the internal singleton reference to null. + */ public static void reset() { synchronized (INIT_LOCK) { if (INSTANCE != null) { @@ -635,4 +884,4 @@ public static void reset() { INSTANCE = null; } } -} +} \ No newline at end of file diff --git a/src/main/java/com/redis/replication/SnapshotProducer.java b/src/main/java/com/redis/replication/SnapshotProducer.java index a93caee..3f02cb9 100644 --- a/src/main/java/com/redis/replication/SnapshotProducer.java +++ b/src/main/java/com/redis/replication/SnapshotProducer.java @@ -80,6 +80,15 @@ public class SnapshotProducer { private static volatile SnapshotProducer INSTANCE; private static final Object INIT_LOCK = new Object(); + /** + * Initializes a new SnapshotProducer instance by creating and setting default atomic state fields. + * + * The following fields are initialized: + * - inProgress: false + * - baselineOffset: 0 + * - lastSnapshotTime: 0 + * - snapshotCount: 0 + */ private SnapshotProducer() { this.inProgress = new AtomicBoolean(false); this.baselineOffset = new AtomicLong(0); @@ -87,6 +96,11 @@ private SnapshotProducer() { this.snapshotCount = new AtomicLong(0); } + /** + * Retrieve the singleton SnapshotProducer instance, creating it lazily in a thread-safe manner. + * + * @return the singleton SnapshotProducer instance + */ public static SnapshotProducer getInstance() { SnapshotProducer instance = INSTANCE; if (instance == null) { @@ -137,9 +151,9 @@ public byte[] generateSnapshot(long replicationOffset) { } /** - * Generates an empty RDB file for replicas connecting to an empty master. + * Create a minimal RDB file representing an empty dataset. * - * @return Empty RDB file bytes + * @return the bytes of a minimal RDB file suitable for replica synchronization (contains header, auxiliary fields, EOF opcode, and an 8-byte CRC64 placeholder) */ public byte[] generateEmptySnapshot() { try { @@ -187,7 +201,14 @@ public byte[] wrapForTransfer(byte[] rdb) { // ==================== Internal RDB Generation ==================== /** - * Creates the actual RDB snapshot from database state. + * Builds a Redis RDB-format snapshot representing the current in-memory database state. + * + * The returned byte array is a complete RDB file containing header and auxiliary fields, + * a SELECTDB opcode for database 0, a database size hint, serialized key-value entries + * (including expiry timestamps when present), an EOF opcode, and an 8-byte CRC64 placeholder. + * + * @return a byte array containing the serialized RDB snapshot + * @throws RuntimeException if snapshot serialization fails */ private byte[] createRdbSnapshot() { try { @@ -276,7 +297,15 @@ private byte[] createRdbSnapshot() { } } - // ==================== RDB Encoding Helpers ==================== + /** + * Writes an AUX field entry to the RDB output stream: emits the AUX opcode followed by the + * provided key and value encoded as RDB strings. + * + * @param out the output stream to write the AUX entry to + * @param key the auxiliary field name + * @param value the auxiliary field value + * @throws IOException if an I/O error occurs while writing to the stream + */ private void writeAuxField(ByteArrayOutputStream out, String key, String value) throws IOException { out.write(RDB_OPCODE_AUX); @@ -284,12 +313,33 @@ private void writeAuxField(ByteArrayOutputStream out, String key, String value) writeString(out, value); } + /** + * Writes a UTF-8 encoded string to the given output stream, preceded by its Redis RDB length encoding. + * + * @param out the target ByteArrayOutputStream to write the length prefix and UTF-8 bytes into + * @param s the string to encode and write + * @throws IOException if an I/O error occurs while writing to the stream + */ private void writeString(ByteArrayOutputStream out, String s) throws IOException { byte[] bytes = s.getBytes(StandardCharsets.UTF_8); writeLength(out, bytes.length); out.write(bytes); } + /** + * Encodes an integer using Redis RDB length encoding and writes the resulting bytes to the output stream. + * + *

Encoding forms: + *

    + *
  • 0 <= length < 64: single byte containing the length.
    • + *
  • 64 <= length < 16384: two bytes with 0x40 prefix in the first byte followed by the low 8 bits.
    • + *
  • length >= 16384: marker byte 0x80 followed by a 4-byte big-endian length.
    • + *
+ * + * @param out the output stream to write encoded length bytes to + * @param length the integer length to encode + * @throws IOException if an I/O error occurs while writing to the stream + */ private void writeLength(ByteArrayOutputStream out, int length) throws IOException { if (length < 64) { out.write(length); @@ -305,6 +355,13 @@ private void writeLength(ByteArrayOutputStream out, int length) throws IOExcepti } } + /** + * Writes the given long as an 8-byte little-endian integer to the provided output stream. + * + * @param out the stream to write the bytes to + * @param value the long value to encode + * @throws IOException if an I/O error occurs while writing to the stream + */ private void writeLongLE(ByteArrayOutputStream out, long value) throws IOException { // Little-endian 8-byte integer for (int i = 0; i < 8; i++) { @@ -313,29 +370,52 @@ private void writeLongLE(ByteArrayOutputStream out, long value) throws IOExcepti } } - // ==================== State Accessors ==================== + /** + * Indicates whether a snapshot is currently being produced. + * + * @return `true` if a snapshot is in progress, `false` otherwise. + */ public boolean isInProgress() { return inProgress.get(); } + /** + * Replication offset recorded when the most recent snapshot generation began. + * + * @return the baseline replication offset captured at snapshot start + */ public long getBaselineOffset() { return baselineOffset.get(); } + /** + * Returns the timestamp when the most recent snapshot was produced. + * + * @return the last snapshot time in milliseconds since the Unix epoch, or 0 if no snapshot has been produced yet + */ public long getLastSnapshotTime() { return lastSnapshotTime.get(); } + /** + * The total number of snapshots produced by this SnapshotProducer. + * + * @return the total number of snapshots produced + */ public long getSnapshotCount() { return snapshotCount.get(); } - // ==================== Reset (Testing) ==================== + /** + * Reset the SnapshotProducer singleton, clearing the cached instance so a new instance will be created on the next call to getInstance(). + * + * This operation acquires the initialization lock to perform the reset in a thread-safe manner (intended for use in tests). + */ public static void reset() { synchronized (INIT_LOCK) { INSTANCE = null; } } -} +} \ No newline at end of file diff --git a/src/main/java/com/redis/server/NettyRedisServer.java b/src/main/java/com/redis/server/NettyRedisServer.java index 85a96fe..1cae4ad 100644 --- a/src/main/java/com/redis/server/NettyRedisServer.java +++ b/src/main/java/com/redis/server/NettyRedisServer.java @@ -34,10 +34,22 @@ public class NettyRedisServer { private final RedisConfig config; + /** + * Creates a NettyRedisServer configured with the provided RedisConfig. + * + * @param config the server configuration to use for startup and runtime settings + */ public NettyRedisServer(RedisConfig config) { this.config = config; } + /** + * Starts the Netty-based Redis server: initializes replication state, configures networking, + * binds to the configured port, optionally connects to a master when running as a replica, + * blocks until the server channel closes, and performs orderly shutdown and replication cleanup. + * + * @throws Exception if server startup, binding, connection to master, or shutdown encounters an error + */ public void run() throws Exception { // Initialize replication initReplication(); @@ -104,7 +116,10 @@ public void initChannel(SocketChannel ch) { } /** - * Initializes replication based on configuration. + * Configure the replication role and master endpoint from the server configuration. + * + * If the server is configured as a replica, sets the replication role to SLAVE and + * records the configured master host and port; otherwise sets the role to MASTER. */ private void initReplication() { ReplicationManager replMgr = ReplicationManager.getInstance(); @@ -139,6 +154,13 @@ private void connectToMaster() { }); } + /** + * Application entry point; parses command-line arguments, prints the resolved configuration, + * and starts the Netty-based Redis server. + * + * @param args command-line arguments forwarded to the configuration parser + * @throws Exception if configuration parsing or server startup fails + */ public static void main(String[] args) throws Exception { RedisConfig config = RedisConfig.getInstance(); @@ -149,4 +171,4 @@ public static void main(String[] args) throws Exception { new NettyRedisServer(config).run(); } -} +} \ No newline at end of file diff --git a/src/main/java/com/redis/server/RedisCommandHandler.java b/src/main/java/com/redis/server/RedisCommandHandler.java index 139d56c..84c8e5d 100644 --- a/src/main/java/com/redis/server/RedisCommandHandler.java +++ b/src/main/java/com/redis/server/RedisCommandHandler.java @@ -83,11 +83,25 @@ public class RedisCommandHandler extends ByteToMessageDecoder { private static final int INCOMPLETE = Integer.MIN_VALUE; /** - * The main entry point called by Netty whenever new data arrives from the network. - * * @param ctx Context to interact with the channel pipeline (e.g., writing responses). + * Decode incoming TCP bytes into complete Redis (RESP) commands, execute or queue them + * (including transaction handling), enforce replica read-only rules, and write responses + * and replication propagation as needed. * - * @param in The input ByteBuf containing raw bytes received from the OS. - * @param out (Unused) We write responses directly to ctx, rather than passing objects up the pipeline. + * Detailed behavior: + * - Supports pipelined commands in the input buffer and handles fragmented packets by + * deferring processing until a full command is available. + * - Resolves commands from the registry; unknown commands produce an error and mark a + * transaction as errored when applicable. + * - If the server is a replica, rejects write commands that are not allowed for replication. + * - Queues non-control commands when inside a MULTI/EXEC transaction and acknowledges with + * RESP_QUEUED; otherwise executes commands and writes their responses. + * - On successful execution of propagatable commands, derives replication arguments and + * propagates the command via the CommandPropagator. + * + * @param ctx the Netty channel context used to read channel state, write responses, and + * access per-channel transaction state + * @param in the input ByteBuf containing raw bytes to parse as RESP commands + * @param out unused; responses are written directly to the channel via {@code ctx} */ @Override protected void decode(ChannelHandlerContext ctx, ByteBuf in, List out) { @@ -310,21 +324,31 @@ private int readInteger(ByteBuf buf) { } /** - * Helper to write a response string back to the client. - * Uses an unpooled buffer because response strings are typically short lived. + * Writes the given response string to the channel and flushes it. + * + * @param ctx the Netty channel handler context to write to + * @param response the response string to send (encoded as UTF-8) */ private void writeResponse(ChannelHandlerContext ctx, String response) { ctx.writeAndFlush(Unpooled.copiedBuffer(response, StandardCharsets.UTF_8)); } /** - * Checks if a command is a replication-related command. - * These commands are allowed on replicas even though they may modify internal state. + * Determine whether the provided command name identifies a replication-related command. + * + * @param commandName the command name (expected in upper-case) + * @return `true` if the name is "REPLCONF" or "PSYNC", `false` otherwise */ private boolean isReplicationCommand(String commandName) { return "REPLCONF".equals(commandName) || "PSYNC".equals(commandName); } + /** + * Handles uncaught exceptions from the Netty pipeline by logging the error and closing the channel. + * + * @param ctx the ChannelHandlerContext for the current channel + * @param cause the thrown exception that triggered this handler + */ @Override public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) { // Standard Netty error handling: log and close connection on fatal errors diff --git a/src/main/java/com/redis/storage/RedisDatabase.java b/src/main/java/com/redis/storage/RedisDatabase.java index 79652b3..4a5e8d0 100644 --- a/src/main/java/com/redis/storage/RedisDatabase.java +++ b/src/main/java/com/redis/storage/RedisDatabase.java @@ -262,20 +262,16 @@ public void shutdown() { // ==================== Atomic Operations ==================== /** - * The "Holy Grail" of Thread-Safe Read-Modify-Write. - *

- * Why use this? - * If you want to append a string, you cannot do: - * 

-     * val = get(k);
-     * put(k, val + "new");
-     *
- * Between the get and put, another thread might have changed the value. - *

- * How it works: - * ConcurrentHashMap locks the specific key bucket. It feeds the current value - * to your function, and atomically updates the map with your result. - * No other thread can touch this key until the function finishes. + * Atomically applies a read-modify-write operation to the value stored for a key. + * + *

The provided remapping function is invoked with the current value, or `null` if the key + * does not exist or has expired. If the function returns `null`, the key is removed; otherwise + * the returned `RedisValue` is stored. When updating an existing entry, its expiry time is + * preserved; a newly created value has no expiry. + * + * @param key the key to update + * @param remappingFunction function that accepts the current value (or `null`) and returns the + * new value to store, or `null` to remove the key */ public void compute(String key, java.util.function.Function remappingFunction) { map.compute(key, (k, existingEntry) -> { @@ -307,12 +303,12 @@ public void compute(String key, java.util.function.Function - * This method is used for RDB generation during full resync. - * The returned map is a shallow copy suitable for iteration. + * Creates a point-in-time snapshot of all non-expired keys and their values. + * + * The snapshot is a shallow, independent map suitable for iteration and RDB serialization. + * Values retain expiry information when an expiry is set. * - * @return Map of key to RedisValue with expiry information + * @return a map from key to RedisValue containing all non-expired entries; expiry is preserved on values when applicable */ public Map getSnapshot() { Map snapshot = new HashMap<>(); @@ -334,11 +330,11 @@ public Map getSnapshot() { } /** - * Returns all keys in the database (for KEYS command). - *

- * Note: This may include keys that are technically expired but not yet cleaned up. + * Provides a collection view of all keys currently stored in the database. + * + * Note: the returned view may include keys that have expired but have not yet been removed. * - * @return Collection of all key names + * @return a `Collection` view of the current keys; may include keys that are expired but not yet cleaned up */ public Collection keys() { return map.keySet(); diff --git a/src/main/java/com/redis/storage/RedisValue.java b/src/main/java/com/redis/storage/RedisValue.java index 828939b..a3ee0dc 100644 --- a/src/main/java/com/redis/storage/RedisValue.java +++ b/src/main/java/com/redis/storage/RedisValue.java @@ -140,7 +140,10 @@ default Map> asStream() { } /** - * Utility to check type equality. + * Check whether this value's Redis type matches the given type. + * + * @param expectedType the Redis type to compare against + * @return {@code true} if the value's type equals {@code expectedType}, {@code false} otherwise */ default boolean isType(Type expectedType) { return getType() == expectedType; @@ -161,9 +164,9 @@ default Long getExpiryTime() { } /** - * Checks if this value is expired (for snapshot consistency). + * Determine whether the value has passed its expiry time. * - * @return true if expired + * @return `true` if the value has an expiry time and that time is less than or equal to the current system time, `false` otherwise. */ default boolean isExpired() { Long expiry = getExpiryTime(); @@ -171,12 +174,12 @@ default boolean isExpired() { } /** - * Creates a copy of this value with an expiry time set. - *

- * Used during snapshot creation to preserve expiry metadata. + * Creates a copy of this value that carries an expiry timestamp. + * + * Used when creating snapshots to preserve expiry metadata. * - * @param expiryTimeMillis The expiry timestamp in milliseconds - * @return A new RedisValue with expiry set + * @param expiryTimeMillis the expiry timestamp in milliseconds since the Unix epoch + * @return a RedisValue that reports the given expiry time and delegates type/data access to this value */ default RedisValue withExpiry(long expiryTimeMillis) { return new ExpiringValue(this, expiryTimeMillis); @@ -340,6 +343,11 @@ public Object getData() { return stream; } + /** + * String representation of this RedisValue when it holds a stream. + * + * @return a string describing the value's type (`STREAM`) and its underlying stream data + */ @Override public String toString() { return "RedisValue{type=STREAM, data=" + stream + "}"; @@ -353,61 +361,126 @@ public String toString() { * This is a transient wrapper, not used for normal storage. */ record ExpiringValue(RedisValue wrapped, long expiryTimeMillis) implements RedisValue { + /** + * Get the runtime RedisValue.Type of the wrapped value. + * + * @return the RedisValue.Type of the wrapped value + */ @Override public Type getType() { return wrapped.getType(); } + /** + * Get the underlying Java object used to represent this value for serialization and debugging. + * + * @return the wrapped value's underlying data object + */ @Override public Object getData() { return wrapped.getData(); } + /** + * The expiry time in milliseconds since the Unix epoch, or null when no expiry is set. + * + * @return the expiry time in milliseconds since epoch, or {@code null} if not set + */ @Override public Long getExpiryTime() { return expiryTimeMillis; } + /** + * Checks whether the stored expiry time has been reached. + * + * @return `true` if the current system time is greater than or equal to the expiry time, `false` otherwise + */ @Override public boolean isExpired() { return expiryTimeMillis <= System.currentTimeMillis(); } + /** + * Create a new RedisValue wrapper that associates the same wrapped value with a different expiry timestamp. + * + * @param newExpiryTimeMillis the expiry time in milliseconds since the epoch + * @return a new RedisValue whose wrapped value is the same as this instance's wrapped value and whose expiry time is `newExpiryTimeMillis` + */ @Override public RedisValue withExpiry(long newExpiryTimeMillis) { return new ExpiringValue(wrapped, newExpiryTimeMillis); } + /** + * Retrieve the stored string value from the wrapped RedisValue. + * + * @return the wrapped value's string + * @throws IllegalStateException if the wrapped value is not of type STRING + */ @Override public String asString() { return wrapped.asString(); } + /** + * Exposes the wrapped value as a list view. + * + * @return the underlying value as a List + * @throws IllegalStateException if the wrapped value is not a list (WRONGTYPE) + */ @Override public List asList() { return wrapped.asList(); } + /** + * Provides the set view of this value. + * + * @return the set of strings represented by this value + */ @Override public Set asSet() { return wrapped.asSet(); } + /** + * Return the wrapped value as a hash map. + * + * @return the underlying map of field to value for the wrapped hash value + * @throws IllegalStateException if the wrapped value is not of type HASH + */ @Override public Map asHash() { return wrapped.asHash(); } + /** + * Get the wrapped value as a sorted-set view. + * + * @return a map of members to their scores + * @throws IllegalStateException if the value's type is not SORTED_SET + */ @Override public Map asSortedSet() { return wrapped.asSortedSet(); } + /** + * Provides the stream entries indexed by StreamId for this value. + * + * @return a map from each `StreamId` to the entry's field-value map (`Map`). + */ @Override public Map> asStream() { return wrapped.asStream(); } + /** + * Provide a string representation of the expiring RedisValue that includes its runtime type, expiry timestamp, and wrapped data. + * + * @return a string containing the value's runtime type, the expiryTimeMillis, and the wrapped data + */ @Override public String toString() { return "RedisValue{type=" + getType() + ", expiry=" + expiryTimeMillis + ", data=" + wrapped.getData() + "}"; diff --git a/src/test/java/com/redis/integration/ReplicationCommandsIT.java b/src/test/java/com/redis/integration/ReplicationCommandsIT.java index efc3954..31dced1 100644 --- a/src/test/java/com/redis/integration/ReplicationCommandsIT.java +++ b/src/test/java/com/redis/integration/ReplicationCommandsIT.java @@ -13,6 +13,9 @@ @DisplayName("Replication Commands Integration Tests") public class ReplicationCommandsIT extends BaseIntegrationTest { + /** + * Reset the global replication state to a clean baseline before each test. + */ @BeforeEach void resetReplication() { // Ensure we're starting fresh @@ -64,6 +67,11 @@ void testInfoMemory() { assertTrue(result.contains("used_memory:")); } + /** + * Verifies that the server's replication INFO identifies it as the master with no connected replicas. + * + * Sends an INFO replication request and asserts the response contains `role:master` and `connected_slaves:0`. + */ @Test @DisplayName("INFO shows master role by default") void testInfoMasterRole() { @@ -128,6 +136,13 @@ void setupReplica() { sendCommand("REPLCONF", "capa", "psync2"); } + /** + * Verifies that issuing "PSYNC ? -1" triggers a full resynchronization and does not produce an error. + * + * Sends the PSYNC replica handshake and asserts that the server either reports `FULLRESYNC`, + * returns a non-error reply, or yields no direct textual result (embedding channels may not + * surface the response). + */ @Test @DisplayName("PSYNC ? -1 triggers full resync") void testPsyncFullResync() { @@ -153,6 +168,11 @@ void testPsyncWrongArgs() { @DisplayName("WAIT Command") class WaitCommandTests { + /** + * Verifies that sending "WAIT 0 0" returns immediately with a RESP integer reply. + * + * Asserts the response is non-null and starts with ':' (RESP integer prefix). + */ @Test @DisplayName("WAIT with 0 replicas returns immediately") void testWaitZeroReplicas() { @@ -199,6 +219,11 @@ void testWaitNegativeNum() { assertTrue(result.startsWith("-")); } + /** + * Verifies that the WAIT command returns an error when the timeout is negative. + * + * Asserts the server response begins with '-' for the invocation WAIT 1 -100. + */ @Test @DisplayName("WAIT with negative timeout") void testWaitNegativeTimeout() { @@ -238,6 +263,11 @@ void testMasterOffsetStartsZero() { assertEquals(0, mgr.getMasterReplOffset()); } + /** + * Verifies the replication manager starts with no connected replicas. + * + * Asserts that the connected replica count is zero and the replicas collection is empty. + */ @Test @DisplayName("No connected replicas initially") void testNoReplicasInitially() { @@ -247,8 +277,14 @@ void testNoReplicasInitially() { } } + /** + * Removes test keys from the datastore after each test execution. + * + * This teardown deletes the keys "key" and "test_key" to ensure a clean state + * between tests. + */ @AfterEach void cleanup() { cleanupKeys("key", "test_key"); } -} +} \ No newline at end of file