Replication Added - #49
Conversation
📝 WalkthroughWalkthroughAdds a production-grade master–replica replication system: role management, propagation pipeline, bounded backlog + snapshots, master/replica connection implementations, replication-aware command canonicalization and replica write-protection, CLI/env config for replica mode, extensive tests, docs, and an end-to-end test script. Changes
Sequence Diagram(s)sequenceDiagram
participant Client as Client
participant Handler as RedisCommandHandler
participant Cmd as ICommand
participant CP as CommandPropagator
participant RM as ReplicationManager
participant Replica as ReplicaConnection
Client->>Handler: send command args
Handler->>Cmd: execute(args)
Cmd-->>Handler: result
Handler->>Handler: if isWriteCommand()
alt node is MASTER
Handler->>Cmd: getReplicationArgs(originalArgs, result)
Cmd-->>Handler: rewrittenArgs / null
Handler->>CP: propagate(commandName, rewrittenArgs or originalArgs)
CP->>RM: propagateToReplicas(respBytes)
RM->>Replica: send RESP (propagate)
Replica->>Replica: apply command locally
Replica->>RM: ACK offset
RM->>RM: update ack/lag/stats
end
Handler-->>Client: response
sequenceDiagram
participant ReplicaClient as Replica
participant MasterServer as Master
participant RM as ReplicationManager
participant Backlog as ReplicationLog/SnapshotProducer
ReplicaClient->>MasterServer: PSYNC [repl-id] [offset]
MasterServer->>RM: canPartialResync(repl-id, offset)?
alt partial resync possible
RM-->>MasterServer: true
MasterServer->>ReplicaClient: +CONTINUE [repl-id] [offset]
MasterServer->>Backlog: getDataFrom(offset)
Backlog-->>ReplicaClient: backlog bytes
ReplicaClient->>ReplicaClient: apply backlog
RM->>RM: increment partial resync stat
else full resync required
RM-->>MasterServer: false
MasterServer->>ReplicaClient: +FULLRESYNC [repl-id] [offset]
MasterServer->>Backlog: generateSnapshot()
Backlog-->>ReplicaClient: RDB transfer (bulk)
ReplicaClient->>ReplicaClient: load RDB
RM->>RM: increment full resync stat
end
MasterServer->>ReplicaClient: begin streaming commands
ReplicaClient->>MasterServer: periodic ACKs (offset)
Estimated code review effort🎯 4 (Complex) | ⏱️ ~60 minutes Possibly related PRs
Poem
🚥 Pre-merge checks | ✅ 1 | ❌ 2❌ Failed checks (1 warning, 1 inconclusive)
✅ Passed checks (1 passed)
✏️ Tip: You can configure your own custom pre-merge checks in the settings. ✨ Finishing touches
🧪 Generate unit tests (beta)
Thanks for using CodeRabbit! It's free for OSS, and your support helps us grow. If you like it, consider giving us a shout-out. Comment |
![coderabbitai[bot]](https://avatars.githubusercontent.com/in/347564?s=60&v=4){kind=small}
There was a problem hiding this comment.
Actionable comments posted: 14
Note
Due to the large number of review comments, Critical, Major severity comments were prioritized as inline comments.
🤖 Fix all issues with AI agents
In `@src/main/java/com/redis/commands/list/BLPopCommand.java`:
- Around line 251-255: Implement an override of getReplicationCommandName() in
the BLPopCommand class to return the non-blocking replica-safe command name
"LPOP"; locate BLPopCommand (which already overrides isWriteCommand()) and add a
public String getReplicationCommandName() method that returns "LPOP" so replicas
receive the deterministic non-blocking operation instead of BLPOP with timeout.
In `@src/main/java/com/redis/commands/replication/InfoCommand.java`:
- Around line 56-58: The RESP bulk string length is computed using Java char
count; update InfoCommand to compute the byte length of the payload using UTF-8
encoding (e.g., derive byte[] from info.toString() with StandardCharsets.UTF_8
and use its length) and write that byte length into the "$<len>\r\n" header,
ensuring any necessary import (StandardCharsets) is added and the same UTF-8
bytes are used when sending the content to avoid mismatched lengths.
In `@src/main/java/com/redis/commands/replication/WaitCommand.java`:
- Around line 53-60: The timeout parsing treats timeoutMs==0 as a valid finite
timeout, causing waitForReplicas to compute deadline =
System.currentTimeMillis() + 0 and return immediately; change the handling in
WaitCommand so that when timeoutMs == 0 you treat it as an infinite wait (e.g.,
set deadline to Long.MAX_VALUE or pass a sentinel to waitForReplicas) so the
polling loop in waitForReplicas blocks indefinitely until replicas ack; update
any callers and document that timeoutMs==0 means "wait forever".
- Around line 74-77: The current synchronous call to
ReplicationManager.waitForReplicas(...) in WaitCommand blocks Netty's
event-loop; instead offload the blocking polling (or the Thread.sleep-based
logic) off the event-loop by either (A) creating an async API (e.g., add
ReplicationManager.waitForReplicasAsync(...) that returns a
CompletableFuture<Integer> and use it from WaitCommand) or (B) submit a task to
a dedicated ExecutorService from WaitCommand to call waitForReplicas(...) and
then write the response back to the channel from the event-loop (use
ctx.executor().execute(...) or ctx.channel().eventLoop().execute(...) to marshal
the write). If you keep the polling inside ReplicationManager, replace
Thread.sleep-based polling with a non-blocking scheduled check
(ctx.executor().schedule(...) / ScheduledExecutorService) or expose an async
callback; ensure InterruptedException is handled and the final write (returning
":" + acknowledged + "\r\n") is performed on the Netty event-loop to avoid
cross-thread channel access.
In `@src/main/java/com/redis/config/RedisConfig.java`:
- Around line 110-112: The environment-variable parsing in the four getters
(getPort, getBossThreads, getWorkerThreads, getCleanupIntervalMs) can throw
NumberFormatException; update each getter to wrap Integer.parseInt(envVar) in a
try-catch that catches NumberFormatException and falls back to the existing
CLI/default behavior (matching the CLI parsing pattern already used), optionally
logging a debug/warn with the invalid value and key; ensure you reference and
fix Integer.parseInt calls in getPort (REDIS_PORT), getBossThreads,
getWorkerThreads, and getCleanupIntervalMs so no uncaught NumberFormatException
can propagate.
In `@src/main/java/com/redis/replication/MasterConnection.java`:
- Around line 224-233: sendCommand currently uses String.length() for the RESP
bulk string length which is wrong for multi-byte UTF-8 characters; update
MasterConnection.sendCommand to compute the byte length with
arg.getBytes(StandardCharsets.UTF_8).length (or CharsetEncoder) and use that
value in the "$<length>\r\n" prefix, then write the UTF-8 bytes for the argument
into the ByteBuf (instead of relying on the StringBuilder's char count); ensure
the ByteBuf written to channel contains the correct CRLF framing and UTF-8
encoded bytes for each arg.
- Around line 548-586: executeReplicatedCommand currently implements a few
hardcoded commands and misses canonicalized forms (e.g., master converts SET
EX/PX to SET PXAT), silently drops many write commands, and can crash in INCR on
non-numeric values; update executeReplicatedCommand to recognize canonicalized
command names (handle "SET PXAT" and other master-canonicalized variants), add
handlers or a pluggable dispatch for other replicated write commands (LPUSH,
RPUSH, EXPIRE, PEXPIREAT, HSET, SADD, XADD, etc.) that apply the equivalent
mutations to RedisDatabase, and make INCR robust by validating/parsing the
current value with try/catch (or NumberFormat handling) and defining the
semantics for non-numeric values (e.g., treat as 0 or log and skip). Also ensure
unrecognized commands are logged (not dropped) so divergence can be detected.
- Around line 248-271: In MasterConnection.decode, the current mark/reset
pattern used for RESP parsing causes bytes consumed by the streaming RDB loader
to be replayed when handleRdbData(in) returns false; update the logic so that
when loadingRdb is true you do not call
in.markReaderIndex()/in.resetReaderIndex() around the RDB handling (or move the
mark to occur only for the RESP path) — specifically adjust the decode method
around the loadingRdb check and calls to handleRdbData, ensuring partial reads
written into rdbBuffer are not undone by resetReaderIndex; keep
parseRespResponse, argsBuffer.clear(), and handleResponse unchanged.
In `@src/main/java/com/redis/replication/ReplicaConnection.java`:
- Around line 174-188: The current propagateCommand method advances
expectedOffset before the write completes, inflating getReplicationLag if write
fails; change it so expectedOffset is only updated on successful write by
attaching a ChannelFuture listener to channel.writeAndFlush(buf) inside
propagateCommand (or move the addAndGet call after a synchronous success),
incrementing expectedOffset.addAndGet(respCommand.length) in the listener on
success and handling/rolling back on failure (and keep the initial
isActive()/state check and use ReplicaState.STREAMING, channel, writeAndFlush,
expectedOffset, and respCommand to locate the code).
In `@src/main/java/com/redis/replication/ReplicationLog.java`:
- Around line 144-195: In append(...) the large-data branch calls
globalOffset.addAndGet(data.length) then execution continues to the
unconditional globalOffset.addAndGet(data.length) later, causing double
increment; modify append so the data.length >= bufferSize branch computes the
newOffset once (e.g. newOffset = globalOffset.addAndGet(data.length)) and skips
the later unconditional increment (either by returning newOffset immediately
after setting writePosition/firstAvailableOffset/evictionCount, or by setting a
flag to avoid the second addAndGet), ensuring firstAvailableOffset and
evictionCount are updated exactly once and totalBytesWritten is still
incremented for the written bytes.
In `@src/main/java/com/redis/replication/ReplicationManager.java`:
- Around line 476-512: The blocking Thread.sleep loop inside waitForReplicas is
running on the Netty event loop (invoked from WaitCommand.execute via
RedisCommandHandler) and prevents ReplconfCommand.handleAck from updating
acknowledgedOffset; change waitForReplicas to perform its blocking/polling off
the I/O thread by moving the loop into a dedicated background executor or by
using Netty scheduling on the channel/eventLoop (e.g., schedule polls), and
update the call site in WaitCommand.execute to offload or await the async result
so the event loop is never blocked; keep identifiers: waitForReplicas,
requestAckFromReplicas, countAcknowledgedReplicas,
masterReplOffset/acknowledgedOffset, WaitCommand.execute, RedisCommandHandler,
and ReplconfCommand.handleAck while ensuring the final API returns the
acknowledged count to the caller without blocking the Netty worker thread.
In `@src/main/java/com/redis/replication/SnapshotProducer.java`:
- Around line 234-264: The switch in SnapshotProducer that handles
value.getType() currently serializes only STRING, LIST, and STREAM and drops
SET, HASH, and SORTED_SET; add explicit cases for these types: for SET write
RDB_TYPE_SET, writeString(out, key), cast value.getData() to
java.util.Set<String>, writeLength(out, set.size()) and writeString for each
member; for HASH write RDB_TYPE_HASH, writeString(out, key), cast
value.getData() to java.util.Map<String,String>, writeLength(out, map.size())
and for each entry writeString(field) then writeString(value); for SORTED_SET
write RDB_TYPE_ZSET (or the existing sorted-set constant), writeString(out,
key), cast value.getData() to a map or list representing member->score (e.g.,
java.util.Map<String,Double>), writeLength(out, size) and for each entry
writeString(member) then writeString(String.valueOf(score)). Ensure you
reference the same helper methods used elsewhere (writeString, writeLength) and
the constants (RDB_TYPE_SET, RDB_TYPE_HASH, RDB_TYPE_ZSET) so these types are
not silently skipped.
In `@test_replication.sh`:
- Line 23: The script uses set -e which causes the script to abort when the
arithmetic post-increment ((failed++)) returns a non-zero exit status; replace
that pattern with a safe assignment so increments never produce a failing exit
code (e.g., use failed=$((failed + 1)) instead of ((failed++)) wherever tests
increment the failed counter, and update test invocation patterns so they use
"command || failed=$((failed + 1))" rather than relying on ((failed++)); apply
the same change to the other occurrences mentioned (around the block covering
lines 365-372).
- Around line 129-146: send_command is collapsing and re-splitting arguments
which breaks values containing spaces; replace the string-based handling with an
argument array: inside send_command capture all command args as an array (e.g.
local args=( "$@" ) or similar) and use "${args[@]}" when invoking redis-cli
(redis-cli -p "$port" "${args[@]}") and when building the RESP body iterate over
"${args[@]}" to compute each arg length and content; also quote "$port" and
other expansions to avoid word-splitting and remove the local cmd/args=($cmd)
re-splitting logic so values like "Hello, Replication!" remain a single RESP
token.
🟡 Minor comments (25)
README.md-58-105 (1)
58-105:⚠️ Potential issue | 🟡 MinorAdd blank lines around tables to satisfy markdownlint.
Multiple tables are adjacent to headings or text without blank separators, triggering MD058. Please add a blank line before and after each table in these sections.✅ Example fix pattern
-### 🔑 Connection & Utility -| Command | Usage | Description | +### 🔑 Connection & Utility + +| Command | Usage | Description | |:--------|:------|:------------| | `PING` | `PING [message]` | Test connection, returns PONG or echoes message | ... -| `INFO` | `INFO [section]` | Get server information | +| `INFO` | `INFO [section]` | Get server information | +Also applies to: 224-234, 354-361, 446-451
README.md-171-171 (1)
171-171:⚠️ Potential issue | 🟡 MinorSpecify a language for fenced code blocks (MD040).
These fenced blocks should include a language for linting and readability. Usetextfor diagrams andbashfor shell examples where applicable.✅ Example fix pattern
-``` +```text ┌─────────────────────────────────────────────────────────────────────────────┐ ... -``` +``` -``` +```text ┌─────────────────────────────────────────────────────────────────────────────┐ ... -``` +``` -``` +```text parse → validate → canonicalize → execute → append to log → send to replicas -``` +``` -``` +```text Replica connects → sends replicationId + offset ... -``` +```Also applies to: 243-243, 291-291, 312-312
README.md-476-479 (1)
476-479:⚠️ Potential issue | 🟡 MinorConsider a more formal term than “amazing.”
Minor tone/style improvement for contributing guidance.✅ Example fix
-2. Create a feature branch (`git checkout -b feature/amazing-feature`) -3. Commit your changes (`git commit -m 'Add amazing feature'`) +2. Create a feature branch (`git checkout -b feature/new-feature`) +3. Commit your changes (`git commit -m 'Add new feature'`)src/main/java/com/redis/config/RedisConfig.java-75-85 (1)
75-85:⚠️ Potential issue | 🟡 MinorSilent failure when
--replicaofis given with insufficient arguments.If a user passes
--replicaof host(missing port) or just--replicaof, the conditioni + 2 < args.lengthfails silently with no error message. This can be confusing to debug.Proposed fix
case "--replicaof": - if (i + 2 < args.length) { + if (i + 2 < args.length) { replicaOfHost = args[++i]; try { replicaOfPort = Integer.parseInt(args[++i]); } catch (NumberFormatException e) { System.err.println("[RedisConfig] Invalid replica port: " + args[i]); replicaOfHost = null; } + } else { + System.err.println("[RedisConfig] --replicaof requires <host> <port>"); + i = args.length; // skip remaining } break;src/main/java/com/redis/commands/generic/PExpireAtCommand.java-30-48 (1)
30-48:⚠️ Potential issue | 🟡 MinorExtra arguments are silently ignored.
args.size() < 2rejects too few arguments, but if the client sendsPEXPIREAT key timestamp extra_arg, the extra arguments are silently accepted. Real Redis rejects extra arguments with a wrong-number-of-arguments error.Proposed fix
- if (args.size() < 2) { + if (args.size() != 2) {src/test/java/com/redis/commands/generic/PExpireAtCommandTest.java-52-66 (1)
52-66:⚠️ Potential issue | 🟡 MinorIncomplete test — no assertion after the sleep.
The test sleeps for 100ms waiting for lazy expiration but never asserts that the key is actually expired. Either add
assertFalse(db.exists(key))after the sleep, or remove the sleep if lazy expiration can't be reliably tested this way.Proposed fix
// Key should be expired (may need small wait for lazy expiration) Thread.sleep(100); - // The key may still exist but will report as expired on access + assertFalse(db.exists(key), "Key should be expired after past timestamp");src/test/java/com/redis/integration/ReplicationCommandsIT.java-131-140 (1)
131-140:⚠️ Potential issue | 🟡 MinorPSYNC full resync assertion is too lenient — it accepts almost anything.
The condition
result == null || result.contains("FULLRESYNC") || !result.startsWith("-")passes fornull, any FULLRESYNC response, and any non-error string (e.g.,"+OK\r\n"). This doesn't meaningfully validate that a full resync was triggered. Consider tightening the assertion — if the EmbeddedChannel doesn't capture the direct-write response, at least verify the replica was registered or some observable side effect occurred.src/main/java/com/redis/commands/replication/ReplconfCommand.java-70-74 (1)
70-74:⚠️ Potential issue | 🟡 Minor
remoteAddress()can return null on a closing channel — NPE risk.If the channel is in the process of disconnecting,
ctx.channel().remoteAddress()may returnnull, causing aNullPointerExceptionon.toString().🛡️ Suggested fix
- String host = ctx.channel().remoteAddress().toString(); + var remoteAddr = ctx.channel().remoteAddress(); + String host = remoteAddr != null ? remoteAddr.toString() : "unknown";src/main/java/com/redis/storage/RedisDatabase.java-336-345 (1)
336-345:⚠️ Potential issue | 🟡 Minor
keys()returns a live mutable view of the internal map.
map.keySet()returns a view backed by theConcurrentHashMap. Any structural modifications by the caller (e.g.,remove()during iteration) will mutate the database. Return a defensive copy or unmodifiable wrapper.♻️ Suggested fix
public Collection<String> keys() { - return map.keySet(); + return java.util.Collections.unmodifiableCollection(map.keySet()); }src/main/java/com/redis/commands/replication/ReplconfCommand.java-35-36 (1)
35-36:⚠️ Potential issue | 🟡 MinorPotential NPE if the first argument is a null bulk string.
The RESP parser can produce
nullentries for$-1bulk strings.args.get(0).toUpperCase()will throw aNullPointerExceptionin that case.🛡️ Suggested fix
- String subcommand = args.get(0).toUpperCase(); + String subArg = args.get(0); + if (subArg == null) { + return ERR_WRONG_ARGS; + } + String subcommand = subArg.toUpperCase();src/test/java/com/redis/replication/ReplicationLogTest.java-158-178 (1)
158-178:⚠️ Potential issue | 🟡 MinorWraparound read test doesn't actually trigger a wraparound.
fillerisbufferSize - 100bytes andwrapDatais 4 bytes, totalingbufferSize - 96, which fits entirely within the buffer without wrapping. To actually exercise the wraparound read path, the combined write should exceedbufferSize.♻️ Suggested fix
- // Fill to near capacity - byte[] filler = new byte[bufferSize - 100]; + // Fill to near capacity, leaving only 2 bytes before wrap + byte[] filler = new byte[bufferSize - 2]; java.util.Arrays.fill(filler, (byte) 'X'); smallLog.append(filler); // Write data that will wrap byte[] wrapData = "wrap".getBytes(StandardCharsets.UTF_8); smallLog.append(wrapData); - // Read the wrapped data - it should be at the end - long startOffset = bufferSize - 100; + // Read the wrapped data - part is at end of buffer, part at beginning + long startOffset = bufferSize - 2;src/main/java/com/redis/server/RedisCommandHandler.java-45-61 (1)
45-61:⚠️ Potential issue | 🟡 MinorRemove
REPLICA_ALLOWED_COMMANDSconstant; it's unused dead code.The set defined at lines 49–61 is never referenced anywhere in the codebase. The actual replica write protection at line 142 relies on
cmd.isWriteCommand()instead of this allowlist. Either remove this constant or refactor the protection logic to use it for better defensiveness.src/main/java/com/redis/commands/string/SetCommand.java-95-106 (1)
95-106:⚠️ Potential issue | 🟡 MinorPotential
longoverflow with EXAT multiplication.Line 99:
Long.parseLong(args.get(++i)) * 1000Lcan overflow if the parsed value exceedsLong.MAX_VALUE / 1000. While unlikely in practice, Redis itself validates the range. Consider adding an upper-bound check or usingMath.multiplyExactto detect overflow.src/main/java/com/redis/commands/replication/InfoCommand.java-40-54 (1)
40-54:⚠️ Potential issue | 🟡 MinorUnknown INFO sections fall through to "all" — Redis returns empty for unknown sections.
The
defaultcase falls through to the"all"branch, soINFO foobarreturns all sections. Real Redis returns an empty bulk string for unrecognized section names. This is a minor behavioral divergence.src/main/java/com/redis/commands/replication/PsyncCommand.java-46-51 (1)
46-51:⚠️ Potential issue | 🟡 MinorPotential NPE from
ctx.channel().remoteAddress().
remoteAddress()can returnnullfor certain channel types (e.g.,EmbeddedChannelin tests, or if the channel is disconnected). Calling.toString()on it would throw aNullPointerException.Proposed fix
ReplicaConnection replica = replMgr.getReplica(ctx.channel()); if (replica == null) { // Create one if not exists (shouldn't happen in normal flow) - String host = ctx.channel().remoteAddress().toString(); + var remoteAddr = ctx.channel().remoteAddress(); + String host = remoteAddr != null ? remoteAddr.toString() : "unknown"; replica = replMgr.addReplica(ctx.channel(), host, 0); }src/main/java/com/redis/commands/string/SetCommand.java-138-143 (1)
138-143:⚠️ Potential issue | 🟡 MinorTimestamp drift between master storage and replication PXAT value.
db.put(key, value, ttlMillis)at Line 139 likely computes the absolute expiry internally usingSystem.currentTimeMillis() + ttlMillis. Then Line 142 callsSystem.currentTimeMillis()again, producing a slightly different absolute timestamp. This means the replicated PXAT may not exactly match the master's actual expiry.Consider computing the absolute expiry once before the
db.putcall and using it for both storage and the ThreadLocal.Proposed fix
// Set the value if (ttlMillis > 0) { + long absoluteExpiry = System.currentTimeMillis() + ttlMillis; db.put(key, value, ttlMillis); // Store the absolute expiry time for replication if relative time was used if (hasRelativeExpiry) { - lastComputedPxat.set(System.currentTimeMillis() + ttlMillis); + lastComputedPxat.set(absoluteExpiry); }src/main/java/com/redis/commands/generic/ExpireCommand.java-120-122 (1)
120-122:⚠️ Potential issue | 🟡 MinorMissing
@Overrideannotation ongetReplicationCommandName().Since
getReplicationCommandName()overrides the default method from theICommandinterface (line 115 of ICommand.java), it should be annotated with@Overridefor consistency with the other overridden methods in this class (execute(),name(),isWriteCommand(),getReplicationArgs()).Proposed fix
- public String getReplicationCommandName() { + `@Override` + public String getReplicationCommandName() {src/test/java/com/redis/replication/SnapshotProducerTest.java-167-183 (1)
167-183:⚠️ Potential issue | 🟡 Minor
testPreventsConcurrentGenerationdoes not actually test concurrent snapshot rejection.The thread created at line 171 only sleeps — it never calls
generateSnapshot. The assertion at line 182 (assertFalse(producer.isInProgress())) simply verifies the default state. This test provides no coverage for thecompareAndSetguard ingenerateSnapshot.To properly test this, you'd need one thread to hold
inProgress(e.g., via a slow snapshot on a large dataset) while the second thread attemptsgenerateSnapshotand verifies it returnsnull.src/main/java/com/redis/replication/MasterConnection.java-87-88 (1)
87-88:⚠️ Potential issue | 🟡 Minor
channelfield lacksvolatile— unsafe publication across threads.
channelis written on the Netty event loop thread (line 195) but read from arbitrary threads viaisConnected()(line 623) anddisconnect()(line 612). Whileconnectedis volatile, a volatile write toconnecteddoesn't guarantee that thechannelreference written before it is visible to other threads (theconnected = trueat line 196 happens after thechannelassignment at line 195, so in theory volatile ordering helps, but relying on this implicit ordering is fragile).Proposed fix
/** Netty channel to master */ - private Channel channel; + private volatile Channel channel;src/main/java/com/redis/replication/MasterConnection.java-610-618 (1)
610-618:⚠️ Potential issue | 🟡 Minor
disconnect()doesn't releaserdbBufferif RDB loading was in progress.If
disconnect()is called whileloadingRdbis true,rdbBuffer(allocated at line 485) is never released, causing a direct memory leak. Netty'sUnpooled.buffer()uses heap memory by default so it will eventually be GC'd, but explicit release is best practice.Proposed fix
public void disconnect() { connected = false; + loadingRdb = false; + if (rdbBuffer != null) { + rdbBuffer.release(); + rdbBuffer = null; + } if (channel != null && channel.isActive()) { channel.close(); } if (workerGroup != null) { workerGroup.shutdownGracefully(); } }src/test/java/com/redis/replication/MasterReplicaIntegrationTest.java-455-482 (1)
455-482:⚠️ Potential issue | 🟡 MinorConcurrent snapshot test does not actually test concurrent rejection.
The background thread (lines 467-471) starts, counts down the latch, and calls
generateSnapshot, but there's a race: by the time line 476 (producer.generateSnapshot(0)) executes, the first snapshot may have already completed. The test assertion on line 481 (successCount.get() >= 1 || second != null) is always true since at least one call will succeed regardless of timing.To reliably test concurrent rejection, you would need to make the first snapshot take longer (e.g., populate a very large database) or use synchronization barriers inside the snapshot path.
src/test/java/com/redis/replication/MasterReplicaIntegrationTest.java-47-65 (1)
47-65:⚠️ Potential issue | 🟡 Minor
RedisDatabasestate is not reset between tests — data leaks across test methods.
setUpresetsReplicationManagerandSnapshotProducersingletons but doesn't clearRedisDatabase. SinceRedisDatabaseuses a holder-pattern singleton without areset()method, keys inserted by one test persist into subsequent tests. This causes data leakage, particularly evident in the high-throughput test (lines 493-503) which uses non-prefixed keys like "key0", "key1", etc., that can carry over to other tests and affect snapshot-size assertions.src/main/java/com/redis/replication/ReplicationManager.java-565-614 (1)
565-614:⚠️ Potential issue | 🟡 MinorINFO output for slave mode may emit
master_host:nullif role is set withoutsetMasterInfo.If
setRole(ServerRole.SLAVE)is called directly (bypassingsetMasterInfo),masterHostremainsnull, andgetInfoReplication()would outputmaster_host:null. Consider adding a null guard or ensuringsetMasterInfois the only way to enter slave mode.src/main/java/com/redis/replication/ReplicationManager.java-259-263 (1)
259-263:⚠️ Potential issue | 🟡 MinorNon-atomic compound update in
setMasterInfo— visible to concurrent readers.
masterHostandmasterPortare written beforeroleis set toSLAVE. A concurrent thread callingisSlave()could seeSLAVErole whilemasterHost/masterPortare still stale or default values. SincegetInfoReplication()readsmasterHostunder the slave branch, this could produce incorrect INFO output briefly during role transition.This is likely acceptable during startup, but worth noting if role transitions are ever triggered at runtime.
test_replication.sh-285-296 (1)
285-296:⚠️ Potential issue | 🟡 MinorPotential division by zero if
duration_msis 0.If the system clock resolution causes
startandendto be equal,duration_mswill be 0 and line 296 will fail with a division-by-zero error.🛡️ Proposed fix
local duration_ms=$(( (end - start) / 1000000 )) + if [ "$duration_ms" -eq 0 ]; then duration_ms=1; fi local ops_per_sec=$(( count * 1000 / duration_ms ))
🧹 Nitpick comments (31)
src/main/java/com/redis/config/RedisConfig.java (1)
44-49: SingletongetInstance()is not thread-safe.If two threads call
getInstance()concurrently before it's initialized, two instances can be created. Given this is server config likely initialized once at startup, this may be fine in practice, but worth noting.Option: use holder idiom for lazy thread-safe init
- private static RedisConfig instance; - - public static RedisConfig getInstance() { - if (instance == null) { - instance = new RedisConfig(); - } - return instance; - } + private static class Holder { + private static final RedisConfig INSTANCE = new RedisConfig(); + } + + public static RedisConfig getInstance() { + return Holder.INSTANCE; + }Note: if
parseArgsneeds to be called before first use, you'd need to keep the current approach but addsynchronized.src/main/java/com/redis/replication/RdbGenerator.java (1)
102-105: Consider specifying charset explicitly ingetBytes()calls.Multiple
getBytes()calls (Lines 102, 105, 148, 167) use the platform default charset. While UTF-8 is the default since Java 18 and these are all ASCII strings, explicitStandardCharsets.US_ASCIImakes the binary-protocol intent clearer and guards against any edge cases.Also applies to: 148-148, 167-167
src/test/java/com/redis/integration/ReplicationCommandsIT.java (1)
16-20:ReplicationManager.reset()in@BeforeEachaffects global singleton state.This is fine for sequential test execution, but be aware that if tests ever run in parallel (e.g., via
maven-surefire-pluginwithforkCount), the shared singleton reset could cause flaky failures. The@AfterEachcleanup should ideally also reset or the test class should be annotated with@Isolatedor equivalent to prevent parallel execution issues.src/main/java/com/redis/commands/stream/XAddCommand.java (1)
207-207: Use an import instead of the fully qualifiedjava.util.ArrayList.
java.util.ArrayListis used with its fully qualified name while otherjava.utiltypes (List,Map,LinkedHashMap) are imported at the top of the file.Proposed fix
Add to imports:
import java.util.ArrayList;Then at line 207:
- List<String> replicationArgs = new java.util.ArrayList<>(originalArgs.size()); + List<String> replicationArgs = new ArrayList<>(originalArgs.size());src/test/java/com/redis/replication/ReplicationLogTest.java (1)
304-326: Thread safety test validates serialization but uses a small data volume.The test is good for verifying
totalBytesWrittenconsistency under contention. However, 4000 total bytes on a 64KB buffer won't trigger any contention on the ring-buffer wraparound or eviction path. Consider adding a variant that exceeds the buffer size to stress the eviction logic under concurrency.src/main/java/com/redis/server/RedisCommandHandler.java (1)
167-187: Propagation uses a livesubListview of the reusableargsBuffer.
commandArgs(line 149) is asubListview ofargsBuffer, which is cleared and reused for each command. Currently this is safe becausepropagateis called synchronously. However, ifCommandPropagator.propagate()orreplMgr.propagateToReplicas()ever becomes asynchronous, the buffer will be mutated underneath. Consider passingList.copyOf(commandArgs)topropagatefor safety.♻️ Defensive copy
if (replicationArgs != null) { CommandPropagator.propagate(replicationCmdName, replicationArgs); } else { - CommandPropagator.propagate(replicationCmdName, commandArgs); + CommandPropagator.propagate(replicationCmdName, List.copyOf(commandArgs)); }src/main/java/com/redis/server/NettyRedisServer.java (1)
109-118: Remove redundantsetRole(ServerRole.SLAVE)call.The
setMasterInfo(host, port)method already sets the role toSLAVE(line 262 of ReplicationManager). The explicitsetRole()call on line 113 is unnecessary.♻️ Suggested simplification
private void initReplication() { ReplicationManager replMgr = ReplicationManager.getInstance(); if (config.isReplica()) { - replMgr.setRole(ServerRole.SLAVE); replMgr.setMasterInfo(config.getReplicaOfHost(), config.getReplicaOfPort()); } else { replMgr.setRole(ServerRole.MASTER); } }src/main/java/com/redis/commands/string/SetCommand.java (1)
195-204: Option skipping logic uses two independentifblocks—considerelse iffor clarity.Lines 197 and 201 are independent
ifstatements. While functionally correct (NX/XX won't match EX/PX/PXAT/EXAT), anelse ifwould make the mutual-exclusivity explicit and easier to reason about.src/test/java/com/redis/replication/CommandPropagatorTest.java (1)
166-270: Tests mutate singletonRedisDatabasewithout cleanup — risk of test pollution.
SetCommand.execute()writes to the globalRedisDatabase.getInstance()singleton (e.g., keys"key","nx_test_key","xx_test_key"). Without@BeforeEach/@AfterEachcleanup, leftover state from one test can affect others, especially if test ordering changes or tests run concurrently.Consider adding teardown that removes keys used by these tests after each test, or isolating the database state.
src/main/java/com/redis/commands/replication/InfoCommand.java (1)
8-8: Unused import:java.lang.management.ManagementFactory.This import is not referenced anywhere in the file.
src/main/java/com/redis/commands/replication/PsyncCommand.java (1)
107-107: ReplaceSystem.out.printlnwith structured logging.Multiple
System.out.printlncalls for replication events. These should use SLF4J orjava.util.loggingfor proper log levels, filtering, and production observability.Also applies to: 115-115, 145-146
src/main/java/com/redis/replication/SnapshotProducer.java (3)
213-216: RESIZEDB expires hint is always zero, even when keys have expiry set.The second argument to
RESIZEDBis meant to hint the number of keys with an expiry. Hardcoding0here means RDB loaders won't pre-size the expires hash table. Since you already have the snapshot, you could count entries with expiry.This is a hint (won't break loading), so it's low priority.
Proposed fix
// Database size hint out.write(RDB_OPCODE_RESIZEDB); writeLength(out, snapshot.size()); // db size - writeLength(out, 0); // expires size (simplified) + int expiresCount = (int) snapshot.values().stream() + .filter(v -> v.getExpiryTime() != null && v.getExpiryTime() > 0) + .count(); + writeLength(out, expiresCount); // expires size
114-137:generateSnapshotreturnsnullsilently on concurrent calls — ensure callers handle this.The CAS guard on
inProgressis correct for preventing concurrent snapshots. However, returningnullon contention could lead to NPE in callers if they don't check. TheSystem.out.printlnon line 117 is the only signal. Consider whether logging at a higher severity or throwing would be safer.
293-305:writeLengthonly handlesint— large values aboveInteger.MAX_VALUEwould silently overflow.The RDB length-encoding spec supports lengths up to 2³² - 1 for the 5-byte encoding. Since the parameter is
int, it can't represent values above ~2.1B anyway, but negativeintvalues (from overflow) could produce corrupt output. For the current use cases (string lengths, list sizes), this is unlikely to be hit.src/test/java/com/redis/replication/MasterReplicaIntegrationTest.java (2)
73-83: RESP bulk string length usesString.length()instead of UTF-8 byte length.Line 77 uses
arg.length()which returns the number of Javachars, not the number of UTF-8 bytes. For multi-byte characters (e.g., emoji, CJK), the RESP frame would be malformed. This is fine for ASCII test data but could cause confusing failures if non-ASCII test keys are introduced later.Proposed fix
private String sendCommandOn(EmbeddedChannel channel, String... args) { StringBuilder cmd = new StringBuilder(); cmd.append("*").append(args.length).append("\r\n"); for (String arg : args) { - cmd.append("$").append(arg.length()).append("\r\n").append(arg).append("\r\n"); + cmd.append("$").append(arg.getBytes(StandardCharsets.UTF_8).length).append("\r\n").append(arg).append("\r\n"); }
429-439: Snapshot test uses non-prefixed keys ("key1", "key2") which may collide across tests.Lines 431-432 use
db.put("key1", ...)anddb.put("key2", ...)without thetestPrefix, which could leak state between test classes if they run in the same JVM. ThetestPrefixis defined (line 53) but not used here.Proposed fix
- db.put("key1", "value1"); - db.put("key2", "value2"); + db.put(testPrefix + "key1", "value1"); + db.put(testPrefix + "key2", "value2");src/main/java/com/redis/replication/CommandPropagator.java (3)
180-188:appendBulkStringallocates a byte array per call just to compute length.
value.getBytes(StandardCharsets.UTF_8)creates a temporary array only to read.length. For ASCII-only values (the common case in Redis),value.length()would suffice. For full correctness with multi-byte chars, this allocation is necessary, but it's on the hot propagation path.Consider caching or using a
CharsetEncoderto compute byte length without allocation if this becomes a bottleneck.
112-114:propagateRewrittenis a no-op wrapper aroundpropagate.It provides no additional logic. If the intent is to signal semantic difference at call sites, a comment on
propagatenoting it handles both original and rewritten args would be simpler. That said, it does make call-site intent clearer.
74-76:shouldPropagateassumes uppercase input but doesn't enforce it.The Javadoc states the parameter should be uppercase, but there's no normalization. If a caller passes lowercase, the lookup silently fails and the write command isn't propagated. Consider
commandName.toUpperCase()or at least an assertion.Proposed fix
public static boolean shouldPropagate(String commandName) { - return WRITE_COMMANDS.contains(commandName); + return commandName != null && WRITE_COMMANDS.contains(commandName.toUpperCase()); }src/test/java/com/redis/replication/SnapshotProducerTest.java (1)
18-18: StatickeyCounteris shared across test instances and not reset.If JUnit parallel execution is enabled,
++keyCounteris a data race. Even sequentially, the counter grows across@BeforeEachcalls, so keys are globally unique — which is fine for isolation but the counter never resets. Not a bug currently, but fragile.src/test/java/com/redis/replication/ReplicationScaleTest.java (2)
66-174: Throughput assertions with hard thresholds may cause flaky CI failures.Hard-coded thresholds like 50K ops/sec (line 95), 100K ops/sec (line 124), and 500K ops/sec (line 172) will fail on constrained CI runners (e.g., shared Docker containers, GitHub Actions). The
@Tag("scale")annotation helps by allowing exclusion, but consider adding a profile or system property gate, or relaxing the thresholds by 2-5x as a safety margin.
34-46: NoRedisDatabasecleanup between tests — same concern as inMasterReplicaIntegrationTest.Keys like
"warmup0","key0","counter","prekey0"persist across tests. This is especially problematic fordatabaseHandlesLargeKeyCount(line 382) which assertsdb.size() >= initialSize + keyCount— if prior tests left many keys,initialSizeabsorbs them, but the assertion still holds. However, snapshot tests elsewhere could produce unexpected results from leftover data.src/main/java/com/redis/replication/ReplicaConnection.java (2)
209-212:acknowledgedOffset.set()can regress on out-of-order ACKs.If a stale or reordered REPLCONF ACK arrives with a lower offset than the current value,
set()moves the acknowledged offset backward, which could causeWAITto report fewer synced replicas than actual.Proposed fix
public void updateAcknowledgedOffset(long offset) { - acknowledgedOffset.set(offset); + acknowledgedOffset.accumulateAndGet(offset, Math::max); }
323-328:close()doesn't checkchannel.isOpen()— callingclose()on an already-closed channel is safe in Netty butisActive()may return false for a channel that's still open but not yet connected.Consider using
channel.isOpen()instead for a broader check, or just always callingchannel.close()since Netty'sclose()is idempotent..github/copilot-instructions.md (2)
146-148: Add language specifiers to fenced code blocks.The fenced code blocks on lines 146 and 183 lack language specifiers, which aids syntax highlighting and satisfies markdown linting rules (MD040).
📝 Proposed fix
-``` +```text parse → validate → canonicalize → execute → append to log → send to replicas -``` +```And similarly for the block at line 183:
-``` +```text ID mismatch → FULL SYNC
149-174: Add blank lines around tables for markdown compliance.Static analysis (MD058) flags that tables on lines 151 and 167 should be surrounded by blank lines to render correctly in all Markdown parsers.
📝 Proposed fix (example for line 149-155)
Core Roles:
+
Role Responsibilities </details> </blockquote></details> <details> <summary>test_replication.sh (2)</summary><blockquote> `42-46`: **`pkill -f "redis-server.jar"` may kill unrelated Java processes.** The `-f` flag matches against the full command line. Any Java process whose command line contains `redis-server.jar` (e.g., an IDE indexing the file) would be killed. Consider using the PID file or a more specific pattern (e.g., include the port arguments or track PIDs from the `start_master`/`start_replica` functions). --- `97-127`: **Server startup detection relies on `pgrep` pattern matching and a fixed `sleep 2`, which is fragile.** `pgrep -f "redis-server.jar.*$port"` may not match if the JVM reorders arguments. A more robust approach would be to save the PID from the backgrounded `java` command (`$!`), then poll the log file for a "ready" marker or try connecting to the port. </blockquote></details> <details> <summary>src/main/java/com/redis/replication/ReplicationManager.java (3)</summary><blockquote> `283-303`: **Replace `System.out.println` with a proper logging framework.** `System.out.println` is used throughout for replica lifecycle events (lines 288, 301, 356, 423, 445). This mixes logging with stdout, offers no log levels, and can't be filtered or redirected independently. Consider using `java.util.logging` or SLF4J, consistent with the rest of the server. --- `92-97`: **Singleton makes the `ReplicationManager` hard to unit-test and wire.** The global mutable singleton (with `reset()` for tests) couples all consumers tightly to a static accessor. This works for now, but consider accepting `ReplicationManager` as a constructor parameter in dependents (e.g., `RedisCommandHandler`, command classes) to improve testability and allow parallel test execution without interference. Also applies to: 214-225 --- `619-637`: **`shutdown()` is not idempotent — calling it twice would re-close already-closed channels.** `ReplicaConnection.close()` sets state to `DISCONNECTED` and calls `channel.close()` on an already-closed channel, which is a no-op in Netty. So this is safe in practice, but guarding with `if (!shuttingDown)` or a CAS on a shutdown flag would make the intent clearer. </blockquote></details> </blockquote></details> <!-- This is an auto-generated comment by CodeRabbit for review status -->
| String content = info.toString(); | ||
| // Redis Bulk String format: $[len]\r\n[data]\r\n | ||
| return "$" + content.length() + "\r\n" + content + "\r\n"; |
There was a problem hiding this comment.
RESP Bulk String uses content.length() (char count) instead of byte length.
RESP protocol specifies $<byte_count>\r\n<data>\r\n. Using String.length() returns the number of Java chars (UTF-16 code units), not bytes. If the OS name or version contains multi-byte UTF-8 characters, the reported length will be wrong, causing protocol deserialization errors on the client.
Proposed fix
String content = info.toString();
- // Redis Bulk String format: $[len]\r\n[data]\r\n
- return "$" + content.length() + "\r\n" + content + "\r\n";
+ // Redis Bulk String format: $[byte_len]\r\n[data]\r\n
+ return "$" + content.getBytes(java.nio.charset.StandardCharsets.UTF_8).length + "\r\n" + content + "\r\n";📝 Committable suggestion
‼️ IMPORTANT
Carefully review the code before committing. Ensure that it accurately replaces the highlighted code, contains no missing lines, and has no issues with indentation. Thoroughly test & benchmark the code to ensure it meets the requirements.
| String content = info.toString(); | |
| // Redis Bulk String format: $[len]\r\n[data]\r\n | |
| return "$" + content.length() + "\r\n" + content + "\r\n"; | |
| String content = info.toString(); | |
| // Redis Bulk String format: $[byte_len]\r\n[data]\r\n | |
| return "$" + content.getBytes(java.nio.charset.StandardCharsets.UTF_8).length + "\r\n" + content + "\r\n"; |
🤖 Prompt for AI Agents
In `@src/main/java/com/redis/commands/replication/InfoCommand.java` around lines
56 - 58, The RESP bulk string length is computed using Java char count; update
InfoCommand to compute the byte length of the payload using UTF-8 encoding
(e.g., derive byte[] from info.toString() with StandardCharsets.UTF_8 and use
its length) and write that byte length into the "$<len>\r\n" header, ensuring
any necessary import (StandardCharsets) is added and the same UTF-8 bytes are
used when sending the content to avoid mismatched lengths.
| try { | ||
| timeoutMs = Long.parseLong(args.get(1)); | ||
| if (timeoutMs < 0) { | ||
| return ERR_INVALID_TIMEOUT; | ||
| } | ||
| } catch (NumberFormatException e) { | ||
| return ERR_INVALID_TIMEOUT; | ||
| } |
There was a problem hiding this comment.
timeout=0 should block indefinitely, not return immediately.
In real Redis, WAIT numreplicas 0 means "wait forever (no timeout)." Here, timeoutMs=0 is passed to waitForReplicas, where deadline = System.currentTimeMillis() + 0 causes the polling loop to exit immediately. This means WAIT 1 0 would return 0 instead of blocking until a replica acknowledges.
If you intend to match Redis semantics, treat timeoutMs == 0 as an infinite wait (e.g., use Long.MAX_VALUE as deadline).
🤖 Prompt for AI Agents
In `@src/main/java/com/redis/commands/replication/WaitCommand.java` around lines
53 - 60, The timeout parsing treats timeoutMs==0 as a valid finite timeout,
causing waitForReplicas to compute deadline = System.currentTimeMillis() + 0 and
return immediately; change the handling in WaitCommand so that when timeoutMs ==
0 you treat it as an infinite wait (e.g., set deadline to Long.MAX_VALUE or pass
a sentinel to waitForReplicas) so the polling loop in waitForReplicas blocks
indefinitely until replicas ack; update any callers and document that
timeoutMs==0 means "wait forever".
|
Note Docstrings generation - SUCCESS |
Docstrings generation was requested by @unikdahal. * #49 (comment) 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`
There was a problem hiding this comment.
Actionable comments posted: 9
🤖 Fix all issues with AI agents
In `@src/main/java/com/redis/commands/list/BLPopCommand.java`:
- Around line 251-268: BLPOP currently overrides getReplicationCommandName() but
not getReplicationArgs(), so BLPOP's multi-key+timeout args are forwarded to
replicas as LPOP (wrong). Add an override of getReplicationArgs(Response reply)
in BLPopCommand that extracts the popped key from the execute() response (the
reply is the [key, element] array) and returns a single-key canonical arg list
for replication (e.g., ["<key>"] or ["<key>","1"]). Implement this in the
BLPopCommand class using the same response-parsing logic as execute() to safely
build and return the LPOP args.
In `@src/main/java/com/redis/config/RedisConfig.java`:
- Around line 105-118: The properties-based port parsing in getPort is still
calling Integer.parseInt(properties.getProperty("redis.port",
String.valueOf(DEFAULT_PORT))) without handling NumberFormatException; update
getPort to first read the property value into a String, attempt Integer.parseInt
inside a try-catch, and on NumberFormatException log a warning (e.g., System.err
or logger) and return DEFAULT_PORT; apply the same pattern to the other getters
mentioned (the methods that call properties.getProperty(...) and parse ints on
lines referenced) so every properties-derived integer parse is guarded and falls
back to its respective default constant.
In `@src/main/java/com/redis/replication/MasterConnection.java`:
- Around line 109-116: The rdbBuffer ByteBuf allocated (via
Unpooled.buffer(expectedRdbSize) when starting RDB transfer) can leak if the
connection is closed or an exception occurs while loadingRdb is true; update the
cleanup paths (channelInactive and exceptionCaught handlers and any
disconnect/cleanup method called at connection teardown) to check if loadingRdb
is true and if rdbBuffer != null then release the buffer and set rdbBuffer =
null (and clear/reset loadingRdb/expectedRdbSize as appropriate) to ensure the
ByteBuf is always released on error or disconnect; reference the rdbBuffer,
loadingRdb, expectedRdbSize variables and the Unpooled.buffer allocation site
when applying the fix.
- Around line 458-477: The handlePsyncResponse method currently calls
Long.parseLong(parts[2]) without validation which can throw
NumberFormatException for malformed FULLRESYNC responses; catch
NumberFormatException around parsing (in handlePsyncResponse), log the malformed
value and the full response, and bail out of the FULLRESYNC branch without
changing state (do not call
ReplicationManager.getInstance().setMasterReplOffset, do not set handshakeState,
loadingRdb or expectedRdbSize) so the Netty pipeline is not disrupted; ensure
the log includes the offending parts[2] and the response string to aid
debugging.
- Around line 550-565: handleStreamingCommand currently returns
bytesProcessed.get() for REPLCONF ACK but bytesProcessed is never incremented;
locate the RESP decoding logic in the decode method and, for each decoded RESP
message/command (the same messages later passed to executeReplicatedCommand),
compute the exact number of bytes consumed by that RESP-encoded message and
increment the AtomicLong bytesProcessed by that amount so subsequent REPLCONF
GETACK responses reflect the true processed offset; ensure the increment happens
immediately after successful decode (before calling executeReplicatedCommand)
and use the same byte-counting logic for multi-bulk and inline RESP forms so
ACKs are accurate.
- Around line 487-511: handleRdbData currently consumes a byte when checking the
RDB marker, losing that byte on protocol error; change handleRdbData to accept
ChannelHandlerContext (or use stored ctx) and use in.getByte(in.readerIndex())
instead of in.readByte() to peek without consuming, and if the marker is not '$'
log a clear protocol error and close the connection via ctx.close() (and
set/cleanup any RDB state like expectedRdbSize/rdbBuffer as needed); update the
decode caller to pass the ChannelHandlerContext to handleRdbData.
In `@src/main/java/com/redis/replication/ReplicationManager.java`:
- Around line 631-649: shutdown() currently only closes downstream
ReplicaConnection instances; when running in slave mode you must also disconnect
and release the upstream MasterConnection to avoid leaving its
EventLoopGroup/channel open. Update ReplicationManager.shutdown() to check the
masterConnection (or MasterConnection instance) for non-null, call its
disconnect/close/shutdown methods (the public API on MasterConnection that
closes the channel and shuts down its EventLoopGroup), and handle exceptions;
after closing, set the masterConnection reference to null so reset() and future
init will recreate it. Ensure the same synchronization/visibility as used for
replicas (use existing shutdown flag or INIT_LOCK if needed) to avoid races.
In `@src/main/java/com/redis/replication/SnapshotProducer.java`:
- Around line 291-297: The current STREAM branch in SnapshotProducer (case
STREAM) writes RDB_TYPE_STREAM, the key via writeString, then drops all entries
by calling writeLength(out, 0); update this so streams are not silently
discarded: either implement full stream entry serialization (serialize stream
metadata, entries and consumer groups) instead of writeLength(out, 0), or at
minimum call the class logger (e.g., logger.warn) before writing an empty marker
to record the key and that stream data is being skipped (include key and context
like RDB_TYPE_STREAM and SnapshotProducer). Ensure you modify the STREAM case
handling and any helper used for writing lengths so the warning is emitted for
each skipped stream key.
In `@test_replication.sh`:
- Around line 279-313: The stress_test function spawns many concurrent
background send_command jobs which can exhaust file descriptors/ports; modify
stress_test to limit concurrency (introduce a small concurrency variable like
concurrency=10 and use it instead of 100 when batching/waiting, or replace the
loop with a throttling mechanism/semaphore to ensure no more than concurrency
background jobs run), optionally expose concurrency as a parameter or env var,
and add a short comment/doc note referencing send_command, MASTER_PORT and
REPLICA1_PORT recommending increasing ulimit -n if larger concurrency is
required.
🧹 Nitpick comments (6)
src/main/java/com/redis/commands/replication/WaitCommand.java (2)
41-45: Thread factory produces identically-named threads.Every thread from
WAIT_EXECUTORis named"wait-cmd-worker", making it hard to distinguish concurrent WAIT operations in thread dumps. Consider appending anAtomicIntegercounter.Proposed fix
+ private static final java.util.concurrent.atomic.AtomicInteger THREAD_COUNTER = new java.util.concurrent.atomic.AtomicInteger(0); + private static final ExecutorService WAIT_EXECUTOR = Executors.newCachedThreadPool(r -> { - Thread t = new Thread(r, "wait-cmd-worker"); + Thread t = new Thread(r, "wait-cmd-worker-" + THREAD_COUNTER.getAndIncrement()); t.setDaemon(true); return t; });
93-101: Use top-level imports instead of fully-qualified class names.Lines 95–96 use
io.netty.buffer.Unpooledandjava.nio.charset.StandardCharsetsinline. These should be imported at the top of the file for consistency with the rest of the codebase.Proposed fix
Add to the import section:
import io.netty.buffer.Unpooled; import java.nio.charset.StandardCharsets;Then replace the fully-qualified references in the lambdas with the short names.
src/main/java/com/redis/commands/replication/InfoCommand.java (1)
76-78: Hardcoded stub values for clients and stats sections.
connected_clients:1,blocked_clients:0,total_connections_received:1, andtotal_commands_processed:0are all hardcoded. This is fine as a starting point, but these will produce misleading output for operators.Would you like me to open an issue to track wiring these to actual server metrics?
Also applies to: 93-95
src/main/java/com/redis/replication/SnapshotProducer.java (1)
306-310: Zeroed CRC64 checksum may cause compatibility issues.The placeholder
new byte[8]checksum will fail validation by any Redis-compatible tool or client that verifies RDB integrity. This is acceptable for a first pass but should be documented as a known limitation.src/main/java/com/redis/replication/ReplicaConnection.java (1)
217-220:acknowledgedOffset.set()can regress on reordered ACKs.If two REPLCONF ACK responses arrive out of order (e.g., due to concurrent processing),
set()could overwrite a higher offset with a lower one. UsingupdateAndGetwithMath.maxwould be a safer monotonic update.Proposed fix
public void updateAcknowledgedOffset(long offset) { - // Use set() for simplicity; ACKs should be monotonically increasing - acknowledgedOffset.set(offset); + // Ensure monotonic progress even if ACKs arrive out of order + acknowledgedOffset.updateAndGet(current -> Math.max(current, offset)); }test_replication.sh (1)
42-46:pkill -f "redis-server.jar"may kill unrelated JVM processes.The pattern
redis-server.jarmatches any process whose command line contains this string, including other users' or CI instances. Consider using a PID file or tracking PIDs from thejava ... &launches to kill only the processes this script started.♻️ Suggested approach
+PIDS=() + cleanup() { echo -e "\n${YELLOW}Cleaning up...${NC}" - pkill -f "redis-server.jar" 2>/dev/null || true + for pid in "${PIDS[@]}"; do + kill "$pid" 2>/dev/null || true + done + wait 2>/dev/null rm -rf "$LOG_DIR" }Then capture PIDs when starting servers:
- java $JAVA_OPTS -jar "$JAR_PATH" --port $MASTER_PORT > "$LOG_DIR/master.log" 2>&1 & + java $JAVA_OPTS -jar "$JAR_PATH" --port $MASTER_PORT > "$LOG_DIR/master.log" 2>&1 & + PIDS+=($!)
|
|
||
| @Override | ||
| public boolean isWriteCommand() { | ||
| return true; | ||
| } | ||
|
|
||
| /** | ||
| * Returns the non-blocking command name for replication. | ||
| * <p> | ||
| * BLPOP is a blocking command that should be replicated as LPOP | ||
| * to ensure deterministic behavior on replicas. | ||
| * | ||
| * @return "LPOP" for replica-safe non-blocking operation | ||
| */ | ||
| @Override | ||
| public String getReplicationCommandName() { | ||
| return "LPOP"; | ||
| } |
There was a problem hiding this comment.
🧩 Analysis chain
🏁 Script executed:
#!/bin/bash
echo "=== Checking BLPopCommand.java for getReplicationArgs override ==="
grep -n "getReplicationArgs" src/main/java/com/redis/commands/list/BLPopCommand.java
echo ""
echo "=== ICommand interface default methods ==="
cat src/main/java/com/redis/commands/ICommand.java
echo ""
echo "=== CommandPropagator usage of replication args ==="
grep -A 10 -B 5 "getReplicationArgs\|getReplicationCommandName" src/main/java/com/redis/replication/CommandPropagator.java | head -100Repository: unikdahal/redis-java
Length of output: 4641
🏁 Script executed:
cat src/main/java/com/redis/replication/CommandPropagator.javaRepository: unikdahal/redis-java
Length of output: 6454
🏁 Script executed:
#!/bin/bash
echo "=== Searching for calls to getReplicationArgs() ==="
rg "getReplicationArgs" --type java -B 2 -A 2
echo ""
echo "=== Searching for calls to getReplicationCommandName() ==="
rg "getReplicationCommandName" --type java -B 2 -A 2Repository: unikdahal/redis-java
Length of output: 13626
🏁 Script executed:
sed -n '1,500p' src/main/java/com/redis/server/RedisCommandHandler.java | grep -A 30 "Get canonical args for replication"Repository: unikdahal/redis-java
Length of output: 1603
getReplicationArgs() must be overridden — BLPOP args are incompatible with LPOP.
BLPOP accepts key1 [key2 ...] timeout (multiple keys and a timeout), but LPOP expects key [count] (single key and optional count). Without overriding getReplicationArgs(), the original BLPOP arguments are forwarded unchanged to replicas, causing the replica to receive LPOP key1 key2 ... timeout. The timeout value will be misinterpreted as a count, and LPOP will fail due to receiving multiple keys.
The getReplicationArgs() method receives the response from execute(), which contains the specific key that was actually popped ([key, element]). Use this to return canonical args with only that single key, e.g., ["key"] or ["key", "1"] (with explicit count).
🤖 Prompt for AI Agents
In `@src/main/java/com/redis/commands/list/BLPopCommand.java` around lines 251 -
268, BLPOP currently overrides getReplicationCommandName() but not
getReplicationArgs(), so BLPOP's multi-key+timeout args are forwarded to
replicas as LPOP (wrong). Add an override of getReplicationArgs(Response reply)
in BLPopCommand that extracts the popped key from the execute() response (the
reply is the [key, element] array) and returns a single-key canonical arg list
for replication (e.g., ["<key>"] or ["<key>","1"]). Implement this in the
BLPopCommand class using the same response-parsing logic as execute() to safely
build and return the LPOP args.
There was a problem hiding this comment.
Actionable comments posted: 12
🤖 Fix all issues with AI agents
In `@src/main/java/com/redis/commands/list/BLPopCommand.java`:
- Around line 297-306: The RESP bulk string marker currently uses Java char
counts which breaks multi-byte UTF-8 data; update BLPopCommand.formatResult to
compute lengths with key.getBytes(StandardCharsets.UTF_8).length and
element.getBytes(StandardCharsets.UTF_8).length (mirror the approach used in
CommandPropagator.java at the byte-length calculation), and similarly fix
formatResult implementations in XReadCommand, XRangeCommand, and LPopCommand so
all RESP bulk-string length prefixes use UTF-8 byte lengths rather than
String.length(); ensure the produced RESP lengths match what the parser (which
reads a byte count) expects.
- Around line 283-314: The async BLPOP path in BLPopCommand currently skips
replication because execute() returns null for deferred pops and schedulePolling
writes responses directly; modify schedulePolling (the callback that currently
calls writeResponse) to invoke the same replication hook used for immediate
responses—use CommandPropagator (or the same propagation method used when
execute() returns a non-null resp) to propagate the LPOP to replicas when a
delayed pop actually succeeds (ensure you pass the same replication args as
produced by getReplicationArgs / LPOP propagation). Also fix formatResult to
compute bulk-string lengths in bytes (use
key.getBytes(StandardCharsets.UTF_8).length and
element.getBytes(StandardCharsets.UTF_8).length) so getReplicationArgs can
reliably parse RESP responses.
In `@src/main/java/com/redis/replication/MasterConnection.java`:
- Around line 352-404: Wrap the Integer.parseInt calls in parseBulkString and
parseArray with a try-catch for NumberFormatException: before parsing capture
the start index (already present as start), attempt
Integer.parseInt(lenStr.toString()) inside try, and on NumberFormatException
restore the reader index via buf.readerIndex(start) and return false; also
validate parsed values (e.g., ensure len and numElements are not negative except
for the -1 sentinel) before proceeding so the buffer isn’t left partially
consumed (apply same pattern in both parseBulkString and parseArray).
- Around line 523-535: Wrap the Integer.parseInt(sizeStr.toString()) call in a
try/catch for NumberFormatException inside the MasterConnection read path: on
NumberFormatException, log the malformed sizeStr value (include
sizeStr.toString()), revert any readerIndex adjustments (as done earlier when
end == -1), perform the same error handling used by the FULLRESYNC offset
parsing path (e.g., log error, close/abort the replication handshake or return
false to wait for more data), and do not allocate rdbBuffer or set
expectedRdbSize when parsing fails; keep references to expectedRdbSize,
rdbBuffer and in.readerIndex to implement the rollback/early-return behavior.
In `@src/main/java/com/redis/replication/ReplicationManager.java`:
- Around line 483-524: The waitForReplicas method currently treats timeoutMs ==
0 as Long.MAX_VALUE which can block forever; modify waitForReplicas to cap
"infinite" waits by introducing a MAX_INFINITE_WAIT_MS (e.g., 5 minutes) and use
Math.min(deadline, System.currentTimeMillis() + MAX_INFINITE_WAIT_MS) when
timeoutMs == 0, and also inside the while loop periodically re-check replica
connectivity (use getConnectedReplicaCount() and the replicas collection) and
break/return early if there are no connected replicas or replica set changed;
keep existing calls to requestAckFromReplicas(),
countAcknowledgedReplicas(targetOffset) and honor InterruptedException as
before.
- Around line 526-536: requestAckFromReplicas writes the raw REPLCONF GETACK
bytes directly to replica channels, causing masterReplOffset to not reflect
those bytes and creating an offset mismatch with replicas; modify
requestAckFromReplicas to send the same GETACK via the existing
propagateToReplicas path (or, if propagateToReplicas cannot be used, compute the
exact byte length of the cmd and add that amount to masterReplOffset each time
before/after writing) so the masterReplOffset is updated consistently; touch
symbols: requestAckFromReplicas, propagateToReplicas, masterReplOffset, and
ReplicaConnection to locate the fix.
- Around line 383-448: The backpressure path in propagateToReplicas currently
calls recordReplicaFailure when !channel.isWritable(), which wrongly counts
transient writability into the circuit breaker; remove the recordReplicaFailure
call from that branch and only increment
backpressureEvents/replicasWithBackpressure and skippedBackpressure (and
optionally add a dedicated backpressure-only counter) so those transient
backpressure occurrences do not contribute to the circuit-breaker failure count
tracked by recordReplicaFailure/isCircuitBreakerOpen.
In `@src/main/java/com/redis/replication/SnapshotProducer.java`:
- Around line 383-387: reset() currently just nulls INSTANCE and leaks any
in-progress state; capture the current INSTANCE into a local variable, set
INSTANCE = null inside the synchronized (INIT_LOCK) block, then if the captured
instance is non-null call a cleanup/shutdown method on it (e.g.,
cancelInProgress() / shutdown() / clearInProgress()) that clears its inProgress
flag and notifies any threads waiting on that state (or otherwise
interrupts/joins in-progress work). Add/implement that cleanup method on
SnapshotProducer so it atomically clears inProgress and signals waiting threads
to avoid holding references to the stale instance.
- Around line 216-219: The RESIZEDB expires count is hardcoded to 0 in
SnapshotProducer around the RDB_OPCODE_RESIZEDB write, producing incorrect RDB
metadata; modify SnapshotProducer to compute the actual number of keys with
expirations (e.g., by pre-scanning snapshot or incrementing an expiresCount
during iteration) and replace the hardcoded writeLength(out, 0) with
writeLength(out, expiresCount) so the RDB_OPCODE_RESIZEDB record correctly
reflects snapshot expirations; ensure the computed expiresCount is available
where writeLength is called and use the existing writeLength and
RDB_OPCODE_RESIZEDB symbols.
In `@test_replication.sh`:
- Around line 352-354: The scale_test function currently only logs success when
success==3 but never returns a non-zero exit code on failure; update the
scale_test function to return a non-zero status when the replica check fails
(e.g., if the variable success is not 3) by adding an explicit return 1 (or exit
1 if the function is executed in subshell) after logging the error case so
callers like main can detect failures; reference the scale_test function and the
success variable and ensure log_success remains for the success path while
adding a failing return for the other path.
- Around line 42-46: The cleanup() function currently uses pkill -f
"redis-server.jar" which can kill unrelated processes; instead maintain a global
PIDS array populated when starting servers in start_master and start_replica by
appending the background PID (use $! after launching java) and then in cleanup()
iterate over that PIDS array to kill only those PIDs (e.g., kill -TERM <pid> and
optionally wait or fallback to kill -KILL), then remove "$LOG_DIR"; update
start_master/start_replica to push PIDs into PIDS and ensure PIDS is initialized
(e.g., PIDS=()) so only processes started by this script are terminated.
- Around line 306-311: The script can divide by zero when computing ops_per_sec
because duration_ms may be 0; add a guard around the variables used
(start,end,duration_ms,ops_per_sec) so that if duration_ms is 0 or unset you set
duration_ms=1 (or otherwise avoid the division) before computing ops_per_sec.
Locate the block using start, end, duration_ms, count and log_success and update
it so duration_ms is coerced to a minimum of 1 (or compute ops_per_sec
conditionally) to prevent the arithmetic error.
🧹 Nitpick comments (10)
src/main/java/com/redis/config/RedisConfig.java (4)
44-49: Thread-unsafe lazy singleton.
getInstance()has a classic check-then-act race. Two threads calling it concurrently could create two instances. For a config object this is low-risk, but since this is a server startup class, consider a simple fix.Option: holder idiom (lazy + thread-safe, no synchronization overhead)
- private static RedisConfig instance; + private static final class Holder { + private static final RedisConfig INSTANCE = new RedisConfig(); + } public static RedisConfig getInstance() { - if (instance == null) { - instance = new RedisConfig(); - } - return instance; + return Holder.INSTANCE; }Note: if you need
parseArgsbefore use, you could keep the current approach but addsynchronized, or switch to eager init.
66-72: No validation on parsed port values (applies to both--portand--replicaofport).
Integer.parseIntaccepts any integer, including negatives and values above 65535. A value like99999or-1would be silently accepted and cause a bind failure later with a less obvious error. Same concern applies to env var and properties parsing ingetPort().Suggested helper
private static boolean isValidPort(int port) { return port > 0 && port <= 65535; }Then guard after each successful parse:
cliPort = Integer.parseInt(args[++i]); + if (!isValidPort(cliPort)) { + System.err.println("[RedisConfig] Port out of range: " + cliPort); + cliPort = null; + }
127-179: Consider extracting a helper to reduce repetition across integer getters.All four integer getters (
getPort,getBossThreads,getWorkerThreads,getCleanupIntervalMs) repeat the same env-var → property → default pattern with identical try-catch scaffolding. A small private helper would consolidate this.Example helper
private int getIntConfig(String envKey, String propKey, int defaultValue) { String envVal = System.getenv(envKey); if (envVal != null) { try { return Integer.parseInt(envVal); } catch (NumberFormatException e) { System.err.println("[RedisConfig] Warning: Invalid " + envKey + " value '" + envVal + "', using default"); } } String propVal = properties.getProperty(propKey, String.valueOf(defaultValue)); try { return Integer.parseInt(propVal); } catch (NumberFormatException e) { System.err.println("[RedisConfig] Warning: Invalid " + propKey + " property value '" + propVal + "', using default"); return defaultValue; } }Then each getter becomes a one-liner, e.g.:
public int getBossThreads() { return getIntConfig("REDIS_BOSS_THREADS", "redis.boss.threads", DEFAULT_BOSS_THREADS); }
210-224:toString()calls all getters, which re-read env vars and re-parse on every invocation.This is fine for debugging/logging purposes, but worth being aware that
toString()has side effects (prints warnings to stderr) if config values are malformed. Not a blocker — just a heads-up.test_replication.sh (3)
162-162:localmasks the exit code of command substitutions (SC2155) — repeated throughout the file.
local value=$(send_command ...)always sets$?to 0 regardless of whethersend_commandsucceeded. If the server is unreachable, you silently get an empty string and a misleading "Expected X, got: " error rather than a clear connection failure.This affects lines 162, 176, 202, 220, 246, 254, 263, 264, 270, 293, 306, 318, 343, 357.
The fix is to declare and assign separately:
- local value=$(send_command $REPLICA1_PORT GET testkey) + local value + value=$(send_command "$REPLICA1_PORT" GET testkey)Not urgent since the test still catches the problem (just with a worse error message), but worth fixing for debuggability.
388-394:--scalesilently runs the stress test too — usage text doesn't mention this.Line 388's condition runs
stress_testfor both--stressand--scale, but the usage header (line 19-20) describes them as independent options. If this is intentional (scale implies stress), a note in the usage comment would help. Otherwise, the condition should be[[ "$1" == "--stress" ]]only.
26-29: Hardcoded ports 6379–6382 may collide with a running Redis instance.Consider using higher, non-standard ports (e.g., 16379) by default or allowing override via environment variables, similar to how
STRESS_CONCURRENCYis handled.-MASTER_PORT=6379 -REPLICA1_PORT=6380 -REPLICA2_PORT=6381 -REPLICA3_PORT=6382 +MASTER_PORT=${MASTER_PORT:-16379} +REPLICA1_PORT=${REPLICA1_PORT:-16380} +REPLICA2_PORT=${REPLICA2_PORT:-16381} +REPLICA3_PORT=${REPLICA3_PORT:-16382}src/main/java/com/redis/replication/MasterConnection.java (2)
693-705:INCRreplication usesget()+put()— not atomic.
handleIncrCommandreads withdb.get(key)then writes withdb.put(key, ...). Although replicated commands likely execute on a single Netty thread, this isn't usingdb.compute()like the other handlers do. If any other path (e.g., direct client command on the same database) runs concurrently, the increment can be lost. Consider usingdb.compute()for consistency with the other handlers.Proposed fix
private void handleIncrCommand(List<String> command, RedisDatabase db) { if (command.size() < 2) return; String key = command.get(1); - String current = db.get(key); - try { - long val = (current == null || current.isEmpty()) ? 0 : Long.parseLong(current); - db.put(key, String.valueOf(val + 1)); - } catch (NumberFormatException e) { - System.err.println("[Replication] INCR on non-numeric value for key: " + key); - } + db.compute(key, existing -> { + String current = (existing != null && existing.getType() == RedisValue.Type.STRING) + ? existing.asString() : null; + try { + long val = (current == null || current.isEmpty()) ? 0 : Long.parseLong(current); + return RedisValue.string(String.valueOf(val + 1)); + } catch (NumberFormatException e) { + System.err.println("[Replication] INCR on non-numeric value for key: " + key); + return existing; + } + }); }
997-999:addBytesProcessedis public but never called externally.This method allows external callers to manipulate
bytesProcessed, but the byte tracking is now done internally indecode()(Line 299). If this is unused, consider removing it to reduce the public API surface. If it's needed for testing, consider package-private visibility.src/main/java/com/redis/commands/list/BLPopCommand.java (1)
310-312: Use a proper logger instead ofSystem.err.println.Direct stderr output is hard to filter, lacks log levels, and won't integrate with any logging framework the project may adopt. Consider using
java.util.logging.Loggeror the project's existing logging approach.
There was a problem hiding this comment.
Actionable comments posted: 5
🤖 Fix all issues with AI agents
In `@src/main/java/com/redis/replication/MasterConnection.java`:
- Around line 686-734: In MasterConnection.handleSetCommand, the
NumberFormatException catch blocks for PXAT/PX/EX/EXAT currently call
db.put(key, value) which silently creates a non-expiring key and can cause
master-replica divergence; update each catch to log a warning (include the
command, offending TTL token and the exception) and do NOT call db.put — instead
either skip storing the key or call db.remove(key) to avoid creating a permanent
key; modify the catch handlers in handleSetCommand (replace the calls to db.put
in those NumberFormatException catches) to use the class logger to emit the
warning and then return/skip storage so replicas don't diverge.
- Around line 956-981: handleZAddCommand currently builds the sorted set using a
LinkedHashMap and returns RedisValue.sortedSet(zset), which loses score ordering
when wrapped by SortedSetValue; replace the backing structure with one that
preserves score ordering (e.g., use a TreeMap<Double, LinkedHashSet<String>> or
a custom SortedSetValue-internal structure) so members can be iterated by score
for ZRANGE/ZRANGEBYSCORE; update handleZAddCommand to insert into that ordered
structure (or construct a score-keyed map and pass it into RedisValue.sortedSet)
and adjust SortedSetValue/RedisValue.sortedSet to accept and maintain the
ordered representation.
In `@src/main/java/com/redis/replication/ReplicationManager.java`:
- Around line 544-560: The masterReplOffset is being incremented inside
requestAckFromReplicas for each streaming replica, inflating the offset; move
the single increment of masterReplOffset.addAndGet(cmdLength) so it runs once
per request (e.g., compute cmdLength, detect if there is at least one
ReplicaConnection in state ReplicaConnection.ReplicaState.STREAMING and then
call masterReplOffset.addAndGet(cmdLength) once before the loop), and keep the
loop only responsible for writing the cmd to each replica channel
(replica.getChannel().writeAndFlush(...)); reference requestAckFromReplicas,
masterReplOffset, ReplicaConnection, getState and STREAMING when making the
change.
In `@test_replication.sh`:
- Around line 332-342: The verification currently only logs "Verified $verified
sample keys on replica" and never fails; update the end of the verification in
the stress_test function (the block using verified, count, REPLICA1_PORT and
send_command for keys "stress$i") to compute the expected number of samples (the
number of iterations from for i in $(seq 1 100 $count)), compare verified
against that expected count or a configurable threshold, and if verified is
below the threshold call the failure path (e.g. log_error or log_failure) and
return/exit non‑zero (return 1 or exit 1); otherwise keep the log_success path.
Ensure you reference the same variables (verified, count, REPLICA1_PORT) and use
send_command as before.
- Around line 242-246: In test_info_replication, the block that checks for
"connected_slaves" currently only logs success via log_success without handling
the negative case; update the if in test_info_replication that matches "$info"
against "connected_slaves" to add an else branch that calls log_error (or
similar failure logging), prints the relevant "$info" snippet for debugging, and
exits the test with a non-zero status (return 1) so the test fails when no
connected replicas are reported; keep use of existing helpers (log_success,
log_error) and ensure the failure path is reached when the condition is false.
🧹 Nitpick comments (6)
src/main/java/com/redis/commands/list/BLPopCommand.java (1)
325-327: Avoid logging toSystem.errin production server code.Use the project's logging framework (e.g., SLF4J/Logback) instead of
System.err.printlnfor structured, level-controlled logging. This applies to error paths that could fire under normal operation (malformed responses).test_replication.sh (3)
26-26: DefaultMASTER_PORT=6379may collide with a production Redis instance.On dev machines or CI runners where Redis is already running on port 6379, this script will either fail to bind or send commands to the wrong server. Consider using a non-standard default (e.g., 16379) or reading from an environment variable.
-MASTER_PORT=6379 -REPLICA1_PORT=6380 -REPLICA2_PORT=6381 -REPLICA3_PORT=6382 +MASTER_PORT=${MASTER_PORT:-16379} +REPLICA1_PORT=${REPLICA1_PORT:-16380} +REPLICA2_PORT=${REPLICA2_PORT:-16381} +REPLICA3_PORT=${REPLICA3_PORT:-16382}
175-175: SC2155:local+ command substitution on the same line masks the command's exit code.This pattern appears ~14 times throughout the script (lines 175, 189, 215, 233, 259, 267, 276, 277, 283, 306, 319, 335, 360, 377). Since you check string content rather than exit codes, this is low-risk, but the idiomatic fix is:
local value value=$(send_command ...)This keeps the exit code visible if you ever switch to
set -e-style error detection for these calls.
407-414: Flag parsing only checks$1— combining flags like./test_replication.sh --stress --scalewon't work.
--scalein position$2would be ignored. Consider iterating over all arguments or using a loop-based flag parser.♻️ Suggested approach
+ local run_stress=false run_scale=false + for arg in "$@"; do + case "$arg" in + --stress) run_stress=true ;; + --scale) run_scale=true; run_stress=true ;; + esac + done + # Optional tests based on flags - if [[ "$1" == "--stress" ]] || [[ "$1" == "--scale" ]]; then + if [ "$run_stress" = true ]; then stress_test || failed=$((failed + 1)) fi - if [[ "$1" == "--scale" ]]; then + if [ "$run_scale" = true ]; then scale_test || failed=$((failed + 1)) fisrc/main/java/com/redis/replication/MasterConnection.java (1)
985-1014:cleanupRdbStateguards onloadingRdb— consider also checkingrdbBufferindependently.If
loadingRdbis somehow cleared beforecleanupRdbStateis called (e.g., by a concurrent state change), therdbBufferwould leak. SincerdbBufferis the actual resource, checking it directly is safer.Suggested change
private void cleanupRdbState() { - if (loadingRdb) { - loadingRdb = false; - expectedRdbSize = -1; - if (rdbBuffer != null) { - rdbBuffer.release(); - rdbBuffer = null; - } + loadingRdb = false; + expectedRdbSize = -1; + if (rdbBuffer != null && rdbBuffer.refCnt() > 0) { + rdbBuffer.release(); + rdbBuffer = null; }src/main/java/com/redis/replication/ReplicationManager.java (1)
312-320:getConnectedReplicaCountiterates the full map; consider caching if called frequently.This is called in the
waitForReplicaspolling loop. For a small number of replicas this is fine, but if the replica count grows, the repeated iteration adds overhead to the tight wait loop. A maintainedAtomicIntegerupdated on state transitions would be O(1).
| private void handleSetCommand(List<String> command, RedisDatabase db) { | ||
| if (command.size() < 3) return; | ||
| String key = command.get(1); | ||
| String value = command.get(2); | ||
|
|
||
| // Handle canonicalized PXAT form (absolute millisecond timestamp) | ||
| if (command.size() >= 5 && "PXAT".equalsIgnoreCase(command.get(3))) { | ||
| try { | ||
| long absTimeMs = Long.parseLong(command.get(4)); | ||
| long ttlMs = absTimeMs - System.currentTimeMillis(); | ||
| if (ttlMs > 0) { | ||
| db.put(key, value, ttlMs); | ||
| } else { | ||
| // Already expired, don't store | ||
| db.remove(key); | ||
| } | ||
| } catch (NumberFormatException e) { | ||
| db.put(key, value); | ||
| } | ||
| } else if (command.size() >= 5 && "PX".equalsIgnoreCase(command.get(3))) { | ||
| try { | ||
| long px = Long.parseLong(command.get(4)); | ||
| db.put(key, value, px); | ||
| } catch (NumberFormatException e) { | ||
| db.put(key, value); | ||
| } | ||
| } else if (command.size() >= 5 && "EX".equalsIgnoreCase(command.get(3))) { | ||
| try { | ||
| long ex = Long.parseLong(command.get(4)) * 1000; | ||
| db.put(key, value, ex); | ||
| } catch (NumberFormatException e) { | ||
| db.put(key, value); | ||
| } | ||
| } else if (command.size() >= 5 && "EXAT".equalsIgnoreCase(command.get(3))) { | ||
| try { | ||
| long absTimeSec = Long.parseLong(command.get(4)); | ||
| long ttlMs = (absTimeSec * 1000) - System.currentTimeMillis(); | ||
| if (ttlMs > 0) { | ||
| db.put(key, value, ttlMs); | ||
| } else { | ||
| db.remove(key); | ||
| } | ||
| } catch (NumberFormatException e) { | ||
| db.put(key, value); | ||
| } | ||
| } else { | ||
| db.put(key, value); | ||
| } | ||
| } |
There was a problem hiding this comment.
Silent fallback to plain SET on malformed TTL silently causes master-replica data divergence.
When NumberFormatException is caught for PXAT/PX/EX/EXAT values (lines 702, 709, 716, 728), the key is stored without expiry. Since the master validated the command before propagation, a parse failure here likely indicates data corruption on the wire. Silently dropping the expiry creates a key that never expires on the replica but will expire on the master.
At minimum, log a warning so the divergence is observable:
Proposed fix (example for PXAT)
} catch (NumberFormatException e) {
- db.put(key, value);
+ System.err.println("[Replication] Malformed PXAT value for key '" + key +
+ "': " + command.get(4) + " - storing without expiry");
+ db.put(key, value);
}🤖 Prompt for AI Agents
In `@src/main/java/com/redis/replication/MasterConnection.java` around lines 686 -
734, In MasterConnection.handleSetCommand, the NumberFormatException catch
blocks for PXAT/PX/EX/EXAT currently call db.put(key, value) which silently
creates a non-expiring key and can cause master-replica divergence; update each
catch to log a warning (include the command, offending TTL token and the
exception) and do NOT call db.put — instead either skip storing the key or call
db.remove(key) to avoid creating a permanent key; modify the catch handlers in
handleSetCommand (replace the calls to db.put in those NumberFormatException
catches) to use the class logger to emit the warning and then return/skip
storage so replicas don't diverge.
| private void handleZAddCommand(List<String> command, RedisDatabase db) { | ||
| // ZADD key score member [score member ...] | ||
| if (command.size() < 4) return; | ||
| String key = command.get(1); | ||
|
|
||
| db.compute(key, existing -> { | ||
| java.util.Map<String, Double> zset; | ||
| if (existing == null) { | ||
| zset = new java.util.LinkedHashMap<>(); | ||
| } else if (existing.getType() != RedisValue.Type.SORTED_SET) { | ||
| return existing; | ||
| } else { | ||
| @SuppressWarnings("unchecked") | ||
| java.util.Map<String, Double> existingZset = (java.util.Map<String, Double>) existing.getData(); | ||
| zset = new java.util.LinkedHashMap<>(existingZset); | ||
| } | ||
| for (int i = 2; i + 1 < command.size(); i += 2) { | ||
| try { | ||
| double score = Double.parseDouble(command.get(i)); | ||
| String member = command.get(i + 1); | ||
| zset.put(member, score); | ||
| } catch (NumberFormatException ignored) {} | ||
| } | ||
| return RedisValue.sortedSet(zset); | ||
| }); | ||
| } |
There was a problem hiding this comment.
🧩 Analysis chain
🏁 Script executed:
#!/bin/bash
# Check what data structure RedisValue.sortedSet() expects and how sorted sets are used
ast-grep --pattern 'sortedSet($$$)' --type=javaRepository: unikdahal/redis-java
Length of output: 243
🏁 Script executed:
#!/bin/bash
# Find the RedisValue class and sortedSet method implementation
rg -n 'sortedSet|SORTED_SET' --type=java -A 3 -B 1 src/main/java/com/redis/Repository: unikdahal/redis-java
Length of output: 5117
🏁 Script executed:
#!/bin/bash
# Look for ZRANGE, ZRANGEBYSCORE and how they access sorted set data
rg -n 'ZRANGE|ZRANGEBYSCORE|handleZRange|handleZRangeByScore' --type=java -A 5 -B 1 src/main/java/com/redis/replication/Repository: unikdahal/redis-java
Length of output: 46
🏁 Script executed:
#!/bin/bash
# Find command handlers for sorted set operations
rg -n 'handle.*[Zz]' --type=java -i src/main/java/com/redis/replication/ | grep -i rangeRepository: unikdahal/redis-java
Length of output: 46
🏁 Script executed:
#!/bin/bash
# Check SnapshotProducer serialization of sorted sets
sed -n '290,310p' src/main/java/com/redis/replication/SnapshotProducer.javaRepository: unikdahal/redis-java
Length of output: 1460
🏁 Script executed:
#!/bin/bash
# Search for any code that reads/iterates sorted set data
rg -n 'asSortedSet|getData.*SORTED_SET' --type=java -B2 -A5 src/main/java/com/redis/Repository: unikdahal/redis-java
Length of output: 1576
🏁 Script executed:
#!/bin/bash
# Find all command handlers to see if ZRANGE/ZRANGEBYSCORE exist
rg -n 'case.*:' src/main/java/com/redis/replication/MasterConnection.java | grep -i 'z\|range'Repository: unikdahal/redis-java
Length of output: 46
🏁 Script executed:
#!/bin/bash
# List all handle*Command methods in MasterConnection
rg -n 'private void handle' src/main/java/com/redis/replication/MasterConnection.java | head -20Repository: unikdahal/redis-java
Length of output: 1673
🏁 Script executed:
#!/bin/bash
# Check if there's a getRange or range-related command for sorted sets
rg -n 'ZRANGE|zrange|getRange' --type=java src/main/java/com/redis/Repository: unikdahal/redis-java
Length of output: 185
🏁 Script executed:
#!/bin/bash
# Find ZRANGE/ZRANGEBYSCORE implementation in RedisCommandHandler
rg -n 'ZRANGE|ZRANGEBYSCORE' src/main/java/com/redis/server/RedisCommandHandler.java -B2 -A10Repository: unikdahal/redis-java
Length of output: 587
🏁 Script executed:
#!/bin/bash
# Check if there's sorting logic for sorted set reads
rg -n 'sorted.*stream|sort.*score|TreeMap|compareTo' --type=java src/main/java/com/redis/ -iRepository: unikdahal/redis-java
Length of output: 900
🏁 Script executed:
#!/bin/bash
# Look at the complete RedisCommandHandler to understand how ZRANGE works
wc -l src/main/java/com/redis/server/RedisCommandHandler.javaRepository: unikdahal/redis-java
Length of output: 123
🏁 Script executed:
#!/bin/bash
# Search for ZRANGE case or method implementation
rg -n 'case "ZRANGE"|case "ZRANGEBYSCORE"' src/main/java/com/redis/server/RedisCommandHandler.java -A15Repository: unikdahal/redis-java
Length of output: 46
🏁 Script executed:
#!/bin/bash
# If not in switch case, search for zrange method (lowercase)
rg -n 'zrange|zrangebyscore' --type=java -i src/main/java/com/redis/server/Repository: unikdahal/redis-java
Length of output: 185
🏁 Script executed:
#!/bin/bash
# Look at the executeCommand method and how commands are dispatched
sed -n '1,150p' src/main/java/com/redis/server/RedisCommandHandler.java | grep -n 'case\|switch\|ZRANGE\|execute' | head -30Repository: unikdahal/redis-java
Length of output: 329
🏁 Script executed:
#!/bin/bash
# Find any implementation of ZRANGE/ZRANGEBYSCORE in all Java files
fd --type f "\.java$" src/main/java/com/redis/ | xargs rg -l 'ZRANGE|zrange' -iRepository: unikdahal/redis-java
Length of output: 119
🏁 Script executed:
#!/bin/bash
# Check the context of MasterConnection - what is its purpose?
head -50 src/main/java/com/redis/replication/MasterConnection.java | grep -i 'class\|purpose\|implements\|extends'Repository: unikdahal/redis-java
Length of output: 121
🏁 Script executed:
#!/bin/bash
# Check if there are client command handlers separate from MasterConnection
fd --type f "\.java$" src/main/java/com/redis/commands/ 2>/dev/null | head -10Repository: unikdahal/redis-java
Length of output: 631
🏁 Script executed:
#!/bin/bash
# Find sorted set command implementations
fd --type f "\.java$" src/main/java/com/redis/commands/ | xargs rg -l 'sorted|zset|zrange' -iRepository: unikdahal/redis-java
Length of output: 180
🏁 Script executed:
#!/bin/bash
# List all command files in commands directory
find src/main/java/com/redis/commands/ -name "*Z*.java" -o -name "*Sorted*.java"Repository: unikdahal/redis-java
Length of output: 46
🏁 Script executed:
#!/bin/bash
# Check what's in the commands directory
ls -la src/main/java/com/redis/commands/*/Repository: unikdahal/redis-java
Length of output: 2750
🏁 Script executed:
#!/bin/bash
# Verify: does SortedSetValue actually convert LinkedHashMap to ConcurrentHashMap?
sed -n '297,310p' src/main/java/com/redis/storage/RedisValue.javaRepository: unikdahal/redis-java
Length of output: 436
🏁 Script executed:
#!/bin/bash
# Check if there are any uses of sorted sets outside of handleZAddCommand and snapshot
rg -n 'SortedSetValue|asSortedSet' src/main/java/com/redis/ -B1 -A3Repository: unikdahal/redis-java
Length of output: 2631
🏁 Script executed:
#!/bin/bash
# Verify the structure: is there a sorted/ commands directory that might be missing?
find src/main/java/com/redis/commands -type dRepository: unikdahal/redis-java
Length of output: 345
Use a data structure that supports score-based ordering for sorted sets.
handleZAddCommand stores members in a LinkedHashMap, which is then converted to an unordered ConcurrentHashMap in SortedSetValue's constructor (line 299 of RedisValue.java). This loses all ordering. When operations like ZRANGE and ZRANGEBYSCORE are implemented, they will need to access members in score order, which an unordered map cannot provide. Consider using a TreeMap keyed by score or storing sorted sets with an internal structure that maintains score-based ordering.
🤖 Prompt for AI Agents
In `@src/main/java/com/redis/replication/MasterConnection.java` around lines 956 -
981, handleZAddCommand currently builds the sorted set using a LinkedHashMap and
returns RedisValue.sortedSet(zset), which loses score ordering when wrapped by
SortedSetValue; replace the backing structure with one that preserves score
ordering (e.g., use a TreeMap<Double, LinkedHashSet<String>> or a custom
SortedSetValue-internal structure) so members can be iterated by score for
ZRANGE/ZRANGEBYSCORE; update handleZAddCommand to insert into that ordered
structure (or construct a score-keyed map and pass it into RedisValue.sortedSet)
and adjust SortedSetValue/RedisValue.sortedSet to accept and maintain the
ordered representation.
| 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); | ||
|
|
||
| // Update master offset to account for GETACK command bytes | ||
| // This ensures offset consistency with replicas | ||
| int cmdLength = cmd.length; | ||
|
|
||
| for (ReplicaConnection replica : replicas.values()) { | ||
| if (replica.getState() == ReplicaConnection.ReplicaState.STREAMING) { | ||
| // Increment offset before sending to maintain consistency | ||
| masterReplOffset.addAndGet(cmdLength); | ||
| replica.getChannel().writeAndFlush( | ||
| io.netty.buffer.Unpooled.wrappedBuffer(cmd)); | ||
| } | ||
| } | ||
| } |
There was a problem hiding this comment.
masterReplOffset is incremented once per replica instead of once total, inflating the master offset.
Line 555 (masterReplOffset.addAndGet(cmdLength)) is inside the for loop over replicas. If 3 replicas are streaming, the offset grows by 3 × 37 bytes instead of 37. Replicas each only process the command once, so their reported ACK offset will permanently lag behind the master's inflated offset. This breaks WAIT — replicas can never acknowledge up to the master's offset.
Proposed fix — increment once before the loop
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);
- // Update master offset to account for GETACK command bytes
- // This ensures offset consistency with replicas
- int cmdLength = cmd.length;
+ // Increment offset once (shared replication stream position)
+ masterReplOffset.addAndGet(cmd.length);
for (ReplicaConnection replica : replicas.values()) {
if (replica.getState() == ReplicaConnection.ReplicaState.STREAMING) {
- // Increment offset before sending to maintain consistency
- masterReplOffset.addAndGet(cmdLength);
replica.getChannel().writeAndFlush(
io.netty.buffer.Unpooled.wrappedBuffer(cmd));
}
}
}🤖 Prompt for AI Agents
In `@src/main/java/com/redis/replication/ReplicationManager.java` around lines 544
- 560, The masterReplOffset is being incremented inside requestAckFromReplicas
for each streaming replica, inflating the offset; move the single increment of
masterReplOffset.addAndGet(cmdLength) so it runs once per request (e.g., compute
cmdLength, detect if there is at least one ReplicaConnection in state
ReplicaConnection.ReplicaState.STREAMING and then call
masterReplOffset.addAndGet(cmdLength) once before the loop), and keep the loop
only responsible for writing the cmd to each replica channel
(replica.getChannel().writeAndFlush(...)); reference requestAckFromReplicas,
masterReplOffset, ReplicaConnection, getState and STREAMING when making the
change.
| if [[ "$info" == *"connected_slaves:"* ]]; then | ||
| log_success "Master shows connected_slaves info" | ||
| echo "$info" | grep -E "(role|connected_slaves|master_repl)" | head -10 | ||
| fi | ||
| } |
There was a problem hiding this comment.
test_info_replication silently passes when connected_slaves is absent.
The connected_slaves check (lines 242-245) only logs success but has no else branch returning 1. If the master reports role:master but shows no connected replicas, the test still passes — which defeats the purpose of verifying replication info.
🐛 Proposed fix
if [[ "$info" == *"connected_slaves:"* ]]; then
log_success "Master shows connected_slaves info"
echo "$info" | grep -E "(role|connected_slaves|master_repl)" | head -10
+ else
+ log_error "Master does not show connected_slaves info"
+ return 1
fi🤖 Prompt for AI Agents
In `@test_replication.sh` around lines 242 - 246, In test_info_replication, the
block that checks for "connected_slaves" currently only logs success via
log_success without handling the negative case; update the if in
test_info_replication that matches "$info" against "connected_slaves" to add an
else branch that calls log_error (or similar failure logging), prints the
relevant "$info" snippet for debugging, and exits the test with a non-zero
status (return 1) so the test fails when no connected replicas are reported;
keep use of existing helpers (log_success, log_error) and ensure the failure
path is reached when the condition is false.
| log_info "Verifying replication..." | ||
| local verified=0 | ||
| for i in $(seq 1 100 $count); do | ||
| local value=$(send_command $REPLICA1_PORT GET "stress$i") | ||
| if [[ "$value" == *"v$i"* ]]; then | ||
| verified=$((verified + 1)) | ||
| fi | ||
| done | ||
|
|
||
| log_success "Verified $verified sample keys on replica" | ||
| } |
There was a problem hiding this comment.
stress_test verification never fails — it always reports success regardless of how many keys replicated.
The sample verification loop counts verified keys but never compares against a threshold or returns 1 on insufficient replication. A test that can't fail provides no signal.
🐛 Proposed fix
- log_success "Verified $verified sample keys on replica"
+ local expected=$(( (count - 1) / 100 + 1 )) # number of samples
+ if [ "$verified" -lt "$expected" ]; then
+ log_error "Only $verified/$expected sampled keys verified on replica"
+ return 1
+ fi
+ log_success "Verified $verified/$expected sample keys on replica"
}🧰 Tools
🪛 Shellcheck (0.11.0)
[warning] 335-335: Declare and assign separately to avoid masking return values.
(SC2155)
🤖 Prompt for AI Agents
In `@test_replication.sh` around lines 332 - 342, The verification currently only
logs "Verified $verified sample keys on replica" and never fails; update the end
of the verification in the stress_test function (the block using verified,
count, REPLICA1_PORT and send_command for keys "stress$i") to compute the
expected number of samples (the number of iterations from for i in $(seq 1 100
$count)), compare verified against that expected count or a configurable
threshold, and if verified is below the threshold call the failure path (e.g.
log_error or log_failure) and return/exit non‑zero (return 1 or exit 1);
otherwise keep the log_success path. Ensure you reference the same variables
(verified, count, REPLICA1_PORT) and use send_command as before.
1 participant
Summary by CodeRabbit
New Features
Documentation
Tests