Streaming Compression support for Replication - #3853
sarthakaggarwal97 merged 1 commit into
Conversation
|
Note Reviews pausedIt looks like this branch is under active development. To avoid overwhelming you with review comments due to an influx of new commits, CodeRabbit has automatically paused this review. You can configure this behavior by changing the Use the following commands to manage reviews:
Use the checkboxes below for quick actions:
📝 WalkthroughWalkthroughAdds LZ4-based streaming compression support, wires it into RDB and replication transport paths, updates config and build integration, and adds tests plus a CI job for the new compression scenarios. ChangesStreaming compression and replication transport
Estimated code review effort🎯 5 (Critical) | ⏱️ ~120 minutes Suggested reviewers
🚥 Pre-merge checks | ✅ 4 | ❌ 1❌ Failed checks (1 warning)
✅ Passed checks (4 passed)
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: 12
🧹 Nitpick comments (7)
.github/workflows/ci.yml (1)
212-220: ⚡ Quick winConsider setting
persist-credentials: falsefor supply-chain hardening.The checkout actions at lines 213 and 220 do not explicitly set
persist-credentials: false. While not a functional issue, setting this option prevents the action from configuring Git credentials that could leak in logs or be misused by malicious code.🔒 Proposed fix to add persist-credentials: false
- name: Install libbacktrace uses: actions/checkout@de0fac2e4500dabe0009e67214ff5f5447ce83dd # v6.0.2 with: repository: ianlancetaylor/libbacktrace ref: b9e40069c0b47a722286b94eb5231f7f05c08713 path: libbacktrace + persist-credentials: false - run: cd libbacktrace && ./configure && make && sudo make install - name: Checkout Valkey uses: actions/checkout@de0fac2e4500dabe0009e67214ff5f5447ce83dd # v6.0.2 + with: + persist-credentials: false🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In @.github/workflows/ci.yml around lines 212 - 220, Update the two actions/checkout steps (the "Install libbacktrace" checkout using uses: actions/checkout@de0fac2e4500d... and the "Checkout Valkey" checkout with the same uses) to include persist-credentials: false under their with: blocks to prevent Git credentials from being persisted; ensure you add the persist-credentials: false key alongside the existing repository/ref/path keys and keep YAML indentation consistent.src/compression.h (1)
47-55: ⚡ Quick winAdd function-level docs for the remaining public declarations.
compressionAlgoSupportsStreaming,compressionAlgoName, and the init/destroy APIs are currently undocumented. Please add brief contract comments (return semantics, ownership/lifecycle expectations) for these declarations to keep this public header fully self-describing.As per coding guidelines: "Document why code exists, not just what it does; document all functions in C code".
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@src/compression.h` around lines 47 - 55, Add brief function-level comments for compressionAlgoSupportsStreaming, compressionAlgoName, streamCompressorInit, streamCompressorDestroy, streamDecompressorInit, and streamDecompressorDestroy that describe each function's contract: the meaning of return values (e.g., true/false or 0/error codes), who owns any returned pointers or resources, lifecycle expectations (caller allocates/initializes struct before/after call, who must call Destroy), and any error conditions; place these comments immediately above each declaration in the header so the public API is self-describing and documents why the functions exist as well as how callers must use them.src/rdb.c (1)
3151-3192: ⚡ Quick winDocument the new input-stream helper lifecycle.
rdbInputStreamInit(),rdbInputStreamDestroy(), andrdbInputStreamValidateEnd()add a non-trivial wrapper lifecycle, but onlyrdbInputStreamPrepare()is documented. Please add short contract comments so callers know the required call order and ownership rules.As per coding guidelines, "Document all functions in C code".
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@src/rdb.c` around lines 3151 - 3192, Add short contract comments for rdbInputStreamInit, rdbInputStreamPrepare, rdbInputStreamValidateEnd, and rdbInputStreamDestroy describing the required call order (Init() to set up the struct, optional Prepare() to wrap raw_rio into a decompressor which may replace rdb_rio and set RIO_FLAG_SKIP_RDB_CHECKSUM, ValidateEnd() to be called after reads to verify the decompressor stream, and Destroy() to free the decompressor), the ownership rules (input does not take ownership of the provided raw_rio; Prepare() initializes internal decompressor state that must be cleaned up by Destroy()), and return/behavior expectations (Prepare() returns DECOMPRESS_RIO_INIT_* codes, Destroy() is idempotent, ValidateEnd() returns C_OK/C_ERR only when initialized). Include these comments immediately above the definitions of rdbInputStreamInit, rdbInputStreamPrepare, rdbInputStreamValidateEnd, and rdbInputStreamDestroy and reference the struct fields input->raw_rio, input->rdb_rio, input->decompressor, and input->initialized.src/rdb.h (1)
190-197: ⚡ Quick winAdd declaration-level contract docs for
rdbInputStreamlifecycle API.Please document ownership and required call order (
Init -> Prepare -> ValidateEnd -> Destroy) in the header so callers can’t misuse the stream wrapper.♻️ Suggested header comment shape
+/* Wraps a raw input rio and exposes a logical RDB byte stream. + * Lifecycle: Init -> Prepare -> (use input.rdb_rio) -> ValidateEnd -> Destroy. + * `raw_rio` ownership remains with caller. + */ typedef struct { rio *raw_rio; rio *rdb_rio; decompressRio decompressor; streamReaderInfo stream_info; bool initialized; } rdbInputStream; +/* Initialize wrapper state around `raw_rio` (no stream inspection yet). */ void rdbInputStreamInit(rdbInputStream *input, rio *raw_rio); +/* Prepare input stream; may attach decompression rio depending on envelope. */ decompressRioInitResult rdbInputStreamPrepare(rdbInputStream *input); +/* Validate stream terminates cleanly at logical end. */ int rdbInputStreamValidateEnd(rdbInputStream *input); +/* Release wrapper/decompression resources. */ void rdbInputStreamDestroy(rdbInputStream *input);As per coding guidelines "Document why code exists, not just what it does; document all functions in C code".
Also applies to: 234-238
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@src/rdb.h` around lines 190 - 197, Add a clear declaration-level lifecycle contract comment for the rdbInputStream type describing ownership and required call order: state that callers own the struct, which fields are initialized by the creator vs by Init/Prepare, and that the API must be used in order Init -> Prepare -> ValidateEnd -> Destroy (with Destroy cleaning up raw_rio/rdb_rio/decompressor if initialized). Reference the rdbInputStream type and the lifecycle functions (Init, Prepare, ValidateEnd, Destroy) and document when initialized boolean is set/cleared, which function takes/returns ownership of rio pointers, and that calling functions out of order is undefined to prevent misuse.tests/integration/rdb-compression.tcl (1)
273-275: ⚡ Quick winStrengthen corruption test assertions to validate the failure reason.
These checks only match
*Error*, which is too broad. Prefer asserting specific error/log patterns per scenario (like the existingverify_log_messageusage) so false positives don’t mask regressions.As per coding guidelines "Use clear assertions with meaningful error messages in tests".
Also applies to: 294-296, 350-352, 368-369
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@tests/integration/rdb-compression.tcl` around lines 273 - 275, The test uses broad assertions like the catch/err + assert_match "*Error*" pattern (e.g., the catch {r debug reload nosave} err followed by assert_match "*Error*" $err) which is too generic; replace those with precise checks that validate the expected failure reason—either call the existing verify_log_message helper with the expected error/log substring or change assert_match to match a specific error pattern (include the expected text) for each scenario (also update the similar occurrences at the other reported spots). Ensure you update the specific statements (the catch/err blocks and their assert_match lines) so they assert the exact error or log message that the test intends to trigger rather than the generic "*Error*".src/replication.c (1)
90-219: Please pull in@core-teamfor thisreplication.cchange set.This patch changes replication protocol negotiation, state transitions, and IO-thread interaction in a file the repo treats as architecture-review-required.
As per coding guidelines, "Request
@core-teamarchitectural review for changes to cluster*.c, replication.c, rdb.c, or aof.c".🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@src/replication.c` around lines 90 - 219, This change touches replication.c (functions like markCompressedReplicasForDisconnect, replBalanceAffinity, replicaInitCompressionOnPsync, replInitDecompression, replDestroyDecompression) which per project policy requires an architecture-level review; please add `@core-team` to the PR/reviewers and include a short justification referencing the altered replication protocol negotiation, state transitions, and IO-thread interaction so the core team can perform the requested architectural review before merging.src/server.h (1)
3048-3055: ⚡ Quick winAdd brief header docs for the new replication-compression API.
These are new exported entry points, but the lifecycle/ownership split is not obvious from the names alone. A short comment block here describing per-replica vs singleton state, expected call order, and return semantics would make this interface much safer to consume.
As per coding guidelines, "Document all functions in C code" and "Use comments for non-obvious behavior and rationale, not for restating code".📝 Example
+/* Primary-side per-replica compression lifecycle. Returns C_OK/C_ERR. */ void markCompressedReplicasForDisconnect(void); int replInitCompression(client *c, compressionAlgo algo, int level); void replDestroyCompression(client *c); void replBalanceAffinity(void); + +/* Replica-side compressed stream decode path. `new_data_start` is the first + * newly appended byte in `c->querybuf`. Returns C_OK/C_ERR. */ int replDecompressQueryBuf(client *c, size_t new_data_start); + +/* Replica-side singleton decompressor lifecycle. */ int replInitDecompression(void); void replDestroyDecompression(void); void replRefreshDecompression(void);🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@src/server.h` around lines 3048 - 3055, Add brief header documentation above the new replication-compression API declaring which functions operate on per-replica state vs global/singleton state, the expected call order/lifecycle (e.g. replInitCompression/replDestroyCompression per-client, replInitDecompression/replDestroyDecompression once for the process, when to call replRefreshDecompression and markCompressedReplicasForDisconnect), and the return semantics (which functions return 0 on success / non-zero on error and whether callers own/must free any returned resources). Reference the exact symbols in the docs: markCompressedReplicasForDisconnect, replInitCompression(client *c, compressionAlgo algo, int level), replDestroyCompression(client *c), replBalanceAffinity, replDecompressQueryBuf(client *c, size_t new_data_start), replInitDecompression, replDestroyDecompression, and replRefreshDecompression so callers can quickly see ownership and ordering.
🤖 Prompt for all review comments with AI agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
Inline comments:
In `@deps/lz4/lz4frame.c`:
- Around line 1619-1620: The code computes srcStart = (const BYTE*)srcBuffer and
srcEnd = srcStart + *srcSizePtr even when srcBuffer can be NULL (seen when
calling LZ4F_decompress(dctx, NULL, &o, NULL, &i, NULL)), which causes undefined
behavior; instead of patching the vendored file inline, update/re-vendor to an
upstream LZ4 release that contains the fix, and while doing so verify that the
logic around LZ4F_decompress, srcBuffer, srcSizePtr, srcStart and srcEnd
guards/skips the pointer arithmetic when srcBuffer == NULL or *srcSizePtr == 0
(i.e., ensure the code checks for NULL/zero before computing srcEnd or uses a
safe branch), then run the decompression unit tests that exercise the NULL input
path to confirm the upstream fix resolves the UB.
In `@src/aof.c`:
- Around line 1014-1022: The code currently treats a short read (0 < nread <
sizeof(header)) as success by returning 0, which lets a truncated RDB header be
considered uncompressed; change this so any partial read is treated as an error:
after reading into header from fd (variables nread, header, sizeof(header), fd),
if nread == -1 restore read_errno and return -1 (as already done), and if nread
< (ssize_t)sizeof(header) also set errno to EIO (or restore read_errno if
appropriate) and return -1 so the caller will treat the probe as a failure and
trigger the fallback path instead of accepting a truncated header.
In `@src/compression_stream.c`:
- Around line 818-820: The cap check can overflow because effective_len + len
may wrap; update the check in the push/feed logic (where effective_len is
computed from sdslen(t->feed_queue) - t->feed_head) to first guard against
addition overflow and oversized single inputs by verifying either len >
t->feed_cap or that effective_len is greater than t->feed_cap - len (or
equivalently SIZE_MAX - effective_len < len) before calling
streamReaderSetError(t, STREAM_READER_ERROR_IO); this ensures you never perform
wrapping addition and correctly detect when the new data would exceed
t->feed_cap.
In `@src/config.c`:
- Around line 2590-2596: The updateReplCompression apply-hook is performing an
irreversible action (markCompressedReplicasForDisconnect) during CONFIG SET,
which can be rolled back by configSetCommand; change updateReplCompression to
avoid side effects: instead of calling markCompressedReplicasForDisconnect
directly, record the intent (e.g., set a new flag or push affected replica IDs
into a pending list) and return success; then invoke
markCompressedReplicasForDisconnect only from the post-commit path in
configSetCommand (or a dedicated commit callback) after all option apply-hooks
succeed, using the new flag/pending list; keep the function signature of
updateReplCompression and use symbols updateReplCompression,
markCompressedReplicasForDisconnect, and configSetCommand to locate and wire the
deferred action.
In `@src/networking.c`:
- Around line 4598-4606: c->querybuf is being mutated with sdscatlen() which may
realloc the SDS and corrupt the thread-local shared pointer (thread_shared_qb)
if the primary is still using it; before appending
server.repl_stream_decode_buf, check if c->querybuf == thread_shared_qb and if
so call sdsnewlen()/sdsdup to make a private copy (or otherwise allocate a fresh
SDS) and assign it to c->querybuf, then perform the sdscatlen() append and
update c->querybuf_peak; refer to c->querybuf, server.repl_stream_decode_buf,
thread_shared_qb and the decompression path to implement this guard.
In `@src/rdb.c`:
- Around line 3695-3698: The conditional is checking the wrong streaming flag:
replace the RIO_FLAG_STREAMING_COMPRESSION test with
RIO_FLAG_STREAMING_DECOMPRESSION so load-side streaming-decompression hits the
special notice; update the branch that currently reads "(rdb->flags &
RIO_FLAG_STREAMING_COMPRESSION) && (rdb->flags & RIO_FLAG_SKIP_RDB_CHECKSUM)" to
use RIO_FLAG_STREAMING_DECOMPRESSION, keeping the RIO_FLAG_SKIP_RDB_CHECKSUM
check and the serverLog(LL_NOTICE, ...) calls unchanged (refer to rdb->flags,
RIO_FLAG_STREAMING_DECOMPRESSION, RIO_FLAG_SKIP_RDB_CHECKSUM, and serverLog).
In `@src/replication.c`:
- Around line 4042-4049: The code only advertises compression when
use_diskless_load is true, making server.repl_compression a no-op for disk-based
syncs; change the branch to advertise REPLICA_CAPA_COMPRESSION_STR whenever
server.repl_compression is enabled (i.e., check server.repl_compression instead
of use_diskless_load) so the capability is sent for all full-sync/PSYNC flows;
update the block that sets argv/ lens/ argc (the one referencing
use_diskless_load, REPLICA_CAPA_COMPRESSION_STR and server.repl_compression) to
add the capability whenever server.repl_compression is true and leave any
diskless-specific logic unchanged.
- Around line 216-219: replRefreshDecompression currently drops failures from
replInitDecompression; change replRefreshDecompression to propagate errors by
returning an int (return C_ERR on failure, C_OK on success), call
replDestroyDecompression() then check the return value of
replInitDecompression() and return C_ERR if it fails, and update all new call
sites (those that run handshake/full-sync immediately after calling
replRefreshDecompression) to check its return value and abort the sync session
on C_ERR; keep references to replInitDecompression, replDestroyDecompression,
replRefreshDecompression and use the C_ERR/C_OK constants so callers can act
accordingly.
In `@src/rio.c`:
- Around line 506-509: In rioReadPartial, before calling r->read_some, ensure
the computed bytes_to_read (the min of len and r->max_processing_chunk) is not
greater than SSIZE_MAX; if it is, reject the request and return an appropriate
error instead of invoking r->read_some to avoid backend-dependent truncation/
sign issues. Update the logic around bytes_to_read (used in the existing
computation) to perform this SSIZE_MAX bound check and return an error (and set
errno consistently, e.g. EINVAL/ERANGE) when exceeded; also include the required
header for SSIZE_MAX if not present.
In `@src/server.c`:
- Around line 6654-6665: The compression_ratio calculation is inverted: change
the expression that now computes (uncompressed / compressed) to compute
(compressed / uncompressed) instead, guarding against division by zero. In the
printf-style block that builds the stats string (reference symbols:
"compression_ratio=%.2f", compressionAlgoName(REPL_COMPRESSION_ALGO),
replica->repl_data->repl_compressed_bytes_total,
replica->repl_data->repl_uncompressed_bytes_total), replace the ternary that
returns (double)repl_uncompressed_bytes_total /
(double)repl_compressed_bytes_total with one that returns
repl_uncompressed_bytes_total == 0 ? 0.0 :
(double)replica->repl_data->repl_compressed_bytes_total /
(double)replica->repl_data->repl_uncompressed_bytes_total so compression_ratio
reflects compressed/uncompressed safely.
- Around line 6650-6669: The INFO path reads replication compression counters
(repl_compressed_bytes_total, repl_uncompressed_bytes_total,
repl_compression_errors, repl_compression_cpu_usec,
repl_compression_pending_drains, repl_compression_thread_switches) while IO
threads update them in writeToReplicaCompressed/postWriteToReplica, creating a
data race; fix by making these counters atomic (e.g., _Atomic size_t or
atomic_ullong) or by protecting updates/reads with the replica lock and then use
atomic_load (or the lock) when building the INFO string in src/server.c; update
all increment/+= sites in writeToReplicaCompressed/postWriteToReplica to use
atomic_fetch_add (or acquire the same lock) and replace direct reads in the INFO
formatting block with atomic loads (or locked reads) to ensure race-free access.
In `@src/server.h`:
- Around line 1281-1288: The lifetime replication counters use size_t/long long
and will wrap on 32-bit builds; change repl_compressed_bytes_total,
repl_uncompressed_bytes_total, repl_compression_errors,
repl_compression_pending_drains and repl_compression_thread_switches from size_t
to uint64_t and change repl_compression_cpu_usec from long long to uint64_t
(leave last_processed_tid and affinity_tid as int), ensure the header includes
<stdint.h> and update any INFO/printf format specifiers or casts to use
PRIu64/uint64_t where these fields are printed; apply the same replacements to
the other identical field group mentioned in the comment.
---
Nitpick comments:
In @.github/workflows/ci.yml:
- Around line 212-220: Update the two actions/checkout steps (the "Install
libbacktrace" checkout using uses: actions/checkout@de0fac2e4500d... and the
"Checkout Valkey" checkout with the same uses) to include persist-credentials:
false under their with: blocks to prevent Git credentials from being persisted;
ensure you add the persist-credentials: false key alongside the existing
repository/ref/path keys and keep YAML indentation consistent.
In `@src/compression.h`:
- Around line 47-55: Add brief function-level comments for
compressionAlgoSupportsStreaming, compressionAlgoName, streamCompressorInit,
streamCompressorDestroy, streamDecompressorInit, and streamDecompressorDestroy
that describe each function's contract: the meaning of return values (e.g.,
true/false or 0/error codes), who owns any returned pointers or resources,
lifecycle expectations (caller allocates/initializes struct before/after call,
who must call Destroy), and any error conditions; place these comments
immediately above each declaration in the header so the public API is
self-describing and documents why the functions exist as well as how callers
must use them.
In `@src/rdb.c`:
- Around line 3151-3192: Add short contract comments for rdbInputStreamInit,
rdbInputStreamPrepare, rdbInputStreamValidateEnd, and rdbInputStreamDestroy
describing the required call order (Init() to set up the struct, optional
Prepare() to wrap raw_rio into a decompressor which may replace rdb_rio and set
RIO_FLAG_SKIP_RDB_CHECKSUM, ValidateEnd() to be called after reads to verify the
decompressor stream, and Destroy() to free the decompressor), the ownership
rules (input does not take ownership of the provided raw_rio; Prepare()
initializes internal decompressor state that must be cleaned up by Destroy()),
and return/behavior expectations (Prepare() returns DECOMPRESS_RIO_INIT_* codes,
Destroy() is idempotent, ValidateEnd() returns C_OK/C_ERR only when
initialized). Include these comments immediately above the definitions of
rdbInputStreamInit, rdbInputStreamPrepare, rdbInputStreamValidateEnd, and
rdbInputStreamDestroy and reference the struct fields input->raw_rio,
input->rdb_rio, input->decompressor, and input->initialized.
In `@src/rdb.h`:
- Around line 190-197: Add a clear declaration-level lifecycle contract comment
for the rdbInputStream type describing ownership and required call order: state
that callers own the struct, which fields are initialized by the creator vs by
Init/Prepare, and that the API must be used in order Init -> Prepare ->
ValidateEnd -> Destroy (with Destroy cleaning up raw_rio/rdb_rio/decompressor if
initialized). Reference the rdbInputStream type and the lifecycle functions
(Init, Prepare, ValidateEnd, Destroy) and document when initialized boolean is
set/cleared, which function takes/returns ownership of rio pointers, and that
calling functions out of order is undefined to prevent misuse.
In `@src/replication.c`:
- Around line 90-219: This change touches replication.c (functions like
markCompressedReplicasForDisconnect, replBalanceAffinity,
replicaInitCompressionOnPsync, replInitDecompression, replDestroyDecompression)
which per project policy requires an architecture-level review; please add
`@core-team` to the PR/reviewers and include a short justification referencing the
altered replication protocol negotiation, state transitions, and IO-thread
interaction so the core team can perform the requested architectural review
before merging.
In `@src/server.h`:
- Around line 3048-3055: Add brief header documentation above the new
replication-compression API declaring which functions operate on per-replica
state vs global/singleton state, the expected call order/lifecycle (e.g.
replInitCompression/replDestroyCompression per-client,
replInitDecompression/replDestroyDecompression once for the process, when to
call replRefreshDecompression and markCompressedReplicasForDisconnect), and the
return semantics (which functions return 0 on success / non-zero on error and
whether callers own/must free any returned resources). Reference the exact
symbols in the docs: markCompressedReplicasForDisconnect,
replInitCompression(client *c, compressionAlgo algo, int level),
replDestroyCompression(client *c), replBalanceAffinity,
replDecompressQueryBuf(client *c, size_t new_data_start), replInitDecompression,
replDestroyDecompression, and replRefreshDecompression so callers can quickly
see ownership and ordering.
In `@tests/integration/rdb-compression.tcl`:
- Around line 273-275: The test uses broad assertions like the catch/err +
assert_match "*Error*" pattern (e.g., the catch {r debug reload nosave} err
followed by assert_match "*Error*" $err) which is too generic; replace those
with precise checks that validate the expected failure reason—either call the
existing verify_log_message helper with the expected error/log substring or
change assert_match to match a specific error pattern (include the expected
text) for each scenario (also update the similar occurrences at the other
reported spots). Ensure you update the specific statements (the catch/err blocks
and their assert_match lines) so they assert the exact error or log message that
the test intends to trigger rather than the generic "*Error*".
🪄 Autofix (Beta)
Fix all unresolved CodeRabbit comments on this PR:
- Push a commit to this branch (recommended)
- Create a new PR with the fixes
ℹ️ Review info
⚙️ Run configuration
Configuration used: Repository UI
Review profile: CHILL
Plan: Pro Plus
Run ID: 4465ea09-be4a-45a9-9f78-c6764e703003
📒 Files selected for processing (47)
.github/workflows/ci.ymlcmake/Modules/SourceFiles.cmakecmake/Modules/ValkeySetup.cmakedeps/CMakeLists.txtdeps/Makefiledeps/lz4/CMakeLists.txtdeps/lz4/LICENSEdeps/lz4/Makefiledeps/lz4/lz4.cdeps/lz4/lz4.hdeps/lz4/lz4frame.cdeps/lz4/lz4frame.hdeps/lz4/lz4hc.cdeps/lz4/lz4hc.hdeps/lz4/xxhash.cdeps/lz4/xxhash.hsrc/CMakeLists.txtsrc/Makefilesrc/aof.csrc/compression.csrc/compression.hsrc/compression_lz4.csrc/compression_lz4.hsrc/compression_rio.csrc/compression_rio.hsrc/compression_stream.csrc/compression_stream.hsrc/config.csrc/io_threads.csrc/networking.csrc/rdb.csrc/rdb.hsrc/replication.csrc/rio.csrc/rio.hsrc/server.csrc/server.hsrc/unit/CMakeLists.txtsrc/unit/Makefilesrc/unit/test_compression.cppsrc/unit/test_repl_compression.cppsrc/valkey-check-rdb.ctests/integration/rdb-compression.tcltests/integration/repl-compression.tcltests/integration/replication-aof-sync.tcltests/integration/valkey-check-rdb.tclvalkey.conf
| const BYTE* const srcStart = (const BYTE*)srcBuffer; | ||
| const BYTE* const srcEnd = srcStart + *srcSizePtr; |
There was a problem hiding this comment.
❓ Verification inconclusive
🏁 Script executed:
#!/bin/bash
# Verify the reachable NULL/0 probe path and the null-pointer arithmetic.
rg -n -C2 'LZ4F_decompress\(dctx, NULL, &o, NULL, &i, NULL\)|srcEnd = srcStart \+ \*srcSizePtr' deps/lz4/lz4frame.cRepository: valkey-io/valkey
Length of output: 590
Prevent UB from srcEnd = srcStart + *srcSizePtr when srcBuffer is NULL in deps/lz4/lz4frame.c
LZ4F_decompress(dctx, NULL, &o, NULL, &i, NULL) is called at ~line 1496, and the file later computes srcStart = (const BYTE*)srcBuffer and srcEnd = srcStart + *srcSizePtr at lines 1619–1620. If that block can run with srcBuffer == NULL, pointer arithmetic on a NULL-derived pointer is undefined in C; confirm the probe path’s i/*srcSizePtr value (and that the code guards/skips this arithmetic for the NULL/0 case) and fix via re-vendoring a corrected upstream LZ4 release rather than patching the vendor file in place.
🧰 Tools
🪛 Cppcheck (2.20.0)
[error] 1620-1620: Pointer addition with NULL pointer.
(nullPointerArithmetic)
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@deps/lz4/lz4frame.c` around lines 1619 - 1620, The code computes srcStart =
(const BYTE*)srcBuffer and srcEnd = srcStart + *srcSizePtr even when srcBuffer
can be NULL (seen when calling LZ4F_decompress(dctx, NULL, &o, NULL, &i, NULL)),
which causes undefined behavior; instead of patching the vendored file inline,
update/re-vendor to an upstream LZ4 release that contains the fix, and while
doing so verify that the logic around LZ4F_decompress, srcBuffer, srcSizePtr,
srcStart and srcEnd guards/skips the pointer arithmetic when srcBuffer == NULL
or *srcSizePtr == 0 (i.e., ensure the code checks for NULL/zero before computing
srcEnd or uses a safe branch), then run the decompression unit tests that
exercise the NULL input path to confirm the upstream fix resolves the UB.
| size_t bytes_to_read = | ||
| (r->max_processing_chunk && r->max_processing_chunk < len) ? r->max_processing_chunk : len; | ||
| ssize_t got = r->read_some(r, buf, bytes_to_read); | ||
|
|
There was a problem hiding this comment.
Add SSIZE_MAX bound check before read_some.
rioReadPartial() should reject requests above SSIZE_MAX (Line 506 onward) before calling backend read_some, to avoid backend-dependent truncation/sign behavior on large reads.
Proposed fix
size_t bytes_to_read =
(r->max_processing_chunk && r->max_processing_chunk < len) ? r->max_processing_chunk : len;
+ if (bytes_to_read > (size_t)SSIZE_MAX) {
+ r->flags |= RIO_FLAG_READ_ERROR;
+ return -1;
+ }
ssize_t got = r->read_some(r, buf, bytes_to_read);📝 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.
| size_t bytes_to_read = | |
| (r->max_processing_chunk && r->max_processing_chunk < len) ? r->max_processing_chunk : len; | |
| ssize_t got = r->read_some(r, buf, bytes_to_read); | |
| size_t bytes_to_read = | |
| (r->max_processing_chunk && r->max_processing_chunk < len) ? r->max_processing_chunk : len; | |
| if (bytes_to_read > (size_t)SSIZE_MAX) { | |
| r->flags |= RIO_FLAG_READ_ERROR; | |
| return -1; | |
| } | |
| ssize_t got = r->read_some(r, buf, bytes_to_read); |
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@src/rio.c` around lines 506 - 509, In rioReadPartial, before calling
r->read_some, ensure the computed bytes_to_read (the min of len and
r->max_processing_chunk) is not greater than SSIZE_MAX; if it is, reject the
request and return an appropriate error instead of invoking r->read_some to
avoid backend-dependent truncation/ sign issues. Update the logic around
bytes_to_read (used in the existing computation) to perform this SSIZE_MAX bound
check and return an error (and set errno consistently, e.g. EINVAL/ERANGE) when
exceeded; also include the required header for SSIZE_MAX if not present.
|
@CodeRabbit review the replication compression implementation only |
|
Tip For best results, initiate chat on the files or code changes. 🧩 Analysis chain🏁 Script executed: #!/bin/bash
cat src/replication.c | grep -n "compress\|Compress\|COMPRESS\|lz4\|LZ4\|decompres\|Decompres\|CAPA_COMPRESS\|replBalance\|markCompressed\|replInitDecomp\|replDestroyDecomp\|replRefreshDecomp\|replInitComp\|replDestroyComp" | head -200Length of output: 3685 🏁 Script executed: #!/bin/bash
cat src/networking.c | grep -n "compress\|Compress\|COMPRESS\|lz4\|LZ4\|decompres\|Decompres\|replDecomp\|replInitComp\|replDestroyComp\|compressed_buf\|writeToReplicaComp\|postWriteToReplica" | head -200Length of output: 9715 🏁 Script executed: #!/bin/bash
# Get broader context of replication.c compression functions
rg -n "markCompressedReplicasForDisconnect|replBalanceAffinity|replInitDecompression|replDestroyDecompression|replRefreshDecompression|REPLICA_CAPA_COMPRESSION|repl_compression|repl_stream_decoder" src/replication.c | head -150Length of output: 1831 🏁 Script executed: #!/bin/bash
# Read the compression-related sections from replication.c
awk '/markCompressedReplicasForDisconnect|replBalanceAffinity|replInitDecompression|replDestroyDecompression|replRefreshDecompression/{found=1; start=NR} found && NR>=start && NR<=start+60{print NR": "$0} NR>start+60{found=0}' src/replication.c | head -400Length of output: 19200 🏁 Script executed: #!/bin/bash
# Read networking.c compression functions
awk '/replInitCompression|replDestroyCompression|writeToReplicaCompressed|replDecompressQueryBuf|postWriteToReplica/{found=1; start=NR} found && NR>=start && NR<=start+80{print NR": "$0} NR>start+80{found=0}' src/networking.c | head -500Length of output: 22176 🏁 Script executed: #!/bin/bash
cat src/compression_stream.c | head -150Length of output: 4581 🏁 Script executed: #!/bin/bash
cat src/compression_lz4.cLength of output: 4981 🏁 Script executed: #!/bin/bash
cat src/io_threads.c | grep -n "compress\|Compress\|replBalance\|affinity\|repl_stream_decoder\|repl_compression\|COMPRESSION" | head -100Length of output: 747 🏁 Script executed: #!/bin/bash
# Read server.h for the ClientReplicationData and valkeyServer compression fields
rg -n "repl_compressor|compressed_buf|repl_stream_decoder|repl_compression|REPLICA_CAPA_COMPRESS|compression_error|ClientReplicationData|repl_decompression|repl_apply" src/server.h | head -100Length of output: 2393
Replication Compression ReviewArchitecture SummaryThe design is sound: one long-lived LZ4 frame per replica ( Issues Found1.
|
|
Actionable comments posted: 0 |
sarthakaggarwal97
left a comment
sarthakaggarwal97
left a comment
There was a problem hiding this comment.
Thanks @roshkhatri for this PR. I took an initial look at this change. The idea looks sound to me, and I have posted some comments related to API. I do have another suggestion.
At this point, based on the numbers, thread-affinity doesn't look too helpful. If we conclude that this is helpful. It would be a good follow up PR to this change. I think you also mentioned this offline.
There was a problem hiding this comment.
Actionable comments posted: 6
Caution
Some comments are outside the diff and can’t be posted inline due to platform limitations.
⚠️ Outside diff range comments (1)
src/networking.c (1)
2418-2422:⚠️ Potential issue | 🟡 Minor | ⚡ Quick winLog the negotiated codec, not the build-time default.
compressionAlgoName(REPL_COMPRESSION_ALGO)hard-codes the error message to the default codec, so once another replica codec is negotiated this disconnect log will report the wrong algorithm. Read the algo from the replica's active compressor (or persist the negotiated algo inrepl_data) so diagnostics stay accurate.🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@src/networking.c` around lines 2418 - 2422, The log incorrectly uses the build-time default via compressionAlgoName(REPL_COMPRESSION_ALGO); instead, read the negotiated codec from the replica's active compressor (or from a stored negotiated field in repl_data) so the message reflects the real codec. Update the serverLog call in networking.c to obtain the algo from c->repl_data->repl_compressor (or repl_data->negotiated_compression_algo if you prefer to persist it) and pass compressionAlgoName(...) that value, guarding for a NULL repl_compressor to fall back to "none"/0 as currently done.
♻️ Duplicate comments (2)
src/replication.c (2)
4011-4018:⚠️ Potential issue | 🟠 Major | 🏗️ Heavy liftDon't derive session compression from the live config.
server.repl_compressiononly says what this node wants right now. It does not prove that the current primary negotiated compression for this session, and it can change afterREPLCONF capawas sent. Using it later to decide whether to initialize the decoder can make a replica try to decode plaintext from an older/disabled primary, or skip decoding a stream that was negotiated beforeCONFIG SET replcompression. Please latch the negotiated mode on the replication session/client and use that instead of the mutable global.Also applies to: 2483-2485, 3859-3861, 4388-4390
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@src/replication.c` around lines 4011 - 4018, The code is incorrectly reading the mutable global server.repl_compression when deciding whether a session uses compression; instead latch and consult the per-replication-session field (e.g., add or use a client/session member like client->repl_compression or replClient->negotiated_repl_compression) that is set when the REPLCONF/negotiation completes, and replace all uses of server.repl_compression in the listed spots (the REPLCONF capa construction and the later decoder initialization checks referenced) with that session-level field so the decision reflects the negotiated mode for that specific replication client.
2483-2485:⚠️ Potential issue | 🟠 Major | ⚡ Quick winAbort the sync when decompressor setup fails.
These branches log
replRefreshDecompression() == C_ERRand keep going. At that point the link is already transitioning into incremental replication, so the next compressed bytes reach the parser undecoded. This needs to tear the session down immediately instead of leaving the connection up.Also applies to: 3859-3861, 4388-4390
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@src/replication.c` around lines 2483 - 2485, When server.repl_compression is set and replRefreshDecompression() returns C_ERR, don't just log; immediately tear down the replication session using the failure/cleanup path used for other replication init errors (i.e., invoke the same replication abort/connection teardown routine used elsewhere), so the incremental sync is aborted and the client connection is closed instead of letting compressed bytes reach the parser; update the branch around replRefreshDecompression(), the similar branches at the other locations (lines referenced as 3859-3861 and 4388-4390), to call that cleanup routine right after serverLog when decompression setup fails.
🧹 Nitpick comments (4)
src/server.h (2)
462-469: ⚡ Quick winRename these macros to make “default” explicit.
This block defines hardcoded defaults, but
REPL_COMPRESSION_ALGO/REPL_COMPRESSION_LEVELread like the active negotiated codec. That makes it too easy for runtime code to use the compile-time default where it should read per-replica state.♻️ Suggested rename
-#define REPL_COMPRESSION_ALGO ALGO_LZ4 -#define REPL_COMPRESSION_LEVEL REPL_COMPRESSION_DEFAULT_LEVEL_LZ4 +#define REPL_COMPRESSION_DEFAULT_ALGO ALGO_LZ4 +#define REPL_COMPRESSION_DEFAULT_LEVEL REPL_COMPRESSION_DEFAULT_LEVEL_LZ4🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@src/server.h` around lines 462 - 469, Rename the compile-time macros that represent defaults so they cannot be mistaken for runtime/negotiated values: change REPL_COMPRESSION_ALGO -> REPL_COMPRESSION_DEFAULT_ALGO and REPL_COMPRESSION_LEVEL -> REPL_COMPRESSION_DEFAULT_LEVEL (leave REPL_COMPRESSION_DEFAULT_LEVEL_LZ4 and any per-algo default macros as-is), update the comment accordingly, and then update all uses across the codebase that currently read REPL_COMPRESSION_ALGO or REPL_COMPRESSION_LEVEL to instead read the per-replica runtime fields (e.g., the replication state/codec fields) so compile-time defaults are only referenced via the new *_DEFAULT_* macros.
3037-3043: ⚡ Quick winAdd short contract comments for the new replication-compression APIs.
These are non-obvious cross-module entry points, but the header does not say when each call is valid, what owns the singleton decompressor, or what the
intreturn values mean. A one-line contract above each prototype would make misuse much less likely.💡 Suggested header comments
void disconnectReplicas(void); +/* Reconcile replication-compression state after config or capability changes. */ void reconcileReplicaCompression(void); +/* Initialize or replace the per-replica compressor. Returns C_OK/C_ERR. */ int replInitCompression(client *c, compressionAlgo algo, int level); +/* Release any per-replica compressor state owned by `c`. */ void replDestroyCompression(client *c); +/* Decompress newly appended replication bytes in `c->querybuf`. Returns C_OK/C_ERR. */ int replDecompressQueryBuf(client *c, size_t new_data_start); +/* Initialize the replica-side singleton decompressor. Returns C_OK/C_ERR. */ int replInitDecompression(void); +/* Destroy the replica-side singleton decompressor, if initialized. */ void replDestroyDecompression(void); +/* Recreate replica-side decompression state after refresh events. Returns C_OK/C_ERR. */ int replRefreshDecompression(void);As per coding guidelines: "Use comments for non-obvious behavior and rationale, not for restating code" and "Document all functions in C code".
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@src/server.h` around lines 3037 - 3043, Add one-line contract comments above each declared function (reconcileReplicaCompression, replInitCompression, replDestroyCompression, replDecompressQueryBuf, replInitDecompression, replDestroyDecompression, replRefreshDecompression) describing when the call is valid (e.g., master/replica state or lifecycle ordering), who owns/manages the singleton decompressor (which calls allocate/free it), and what the int return values mean (explicitly list return codes such as 0 for success and non-zero/-1 for error). Keep each comment short and specific (one sentence) and place it immediately above the corresponding prototype so cross-module callers can see preconditions, ownership, and result semantics.Source: Coding guidelines
src/rdb.c (1)
3137-3181: ⚡ Quick winDocument the new RDB stream helper functions.
Only
rdbInputStreamPrepare()is documented right now. Please add brief function comments for the other new helpers as well, especially since this path mixes passthrough and decompression state.As per coding guidelines, "Document why code exists, not just what it does; document all functions in C code".
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@src/rdb.c` around lines 3137 - 3181, Add brief doc-comments for rdbInputStreamInit, rdbInputStreamFree, rdbInputStreamValidateEnd, and rdbRioHasCorruptCompressedInput explaining why each helper exists and how they behave in the passthrough vs decompression path: for rdbInputStreamInit describe it initializes the struct to use a raw rio by default and sets raw_rio/rdb_rio; for rdbInputStreamFree explain it cleans up the decompressor only when initialized and resets rdb_rio to raw_rio; for rdbInputStreamValidateEnd note it asserts initialized and returns C_OK/C_ERR based on decompressRioValidateEnd; and for rdbRioHasCorruptCompressedInput document that it detects corrupt compressed streams only when RIO_FLAG_STREAMING_DECOMPRESSION is set and returns true if decompressRioGetError reports STREAM_READER_ERROR_CORRUPT. Include mention of the initialized flag, the RIO_FLAG_SKIP_RDB_CHECKSUM/RIO_FLAG_STREAMING_DECOMPRESSION flags where relevant, and expected return/side-effect semantics.Source: Coding guidelines
tests/unit/cluster/cluster-shards.tcl (1)
288-288: ⚡ Quick winAssert
CLUSTER SAVECONFIGsuccess explicitly in setup steps.Line 288 and Lines 305-307 should assert
OKso setup failures surface immediately instead of causing later, indirect restart/shard-id failures.Suggested change
- R $i cluster saveconfig + assert_equal {OK} [R $i cluster saveconfig] @@ - for {set i 0} {$i < 8} {incr i} { - R $i cluster saveconfig - } + for {set i 0} {$i < 8} {incr i} { + assert_equal {OK} [R $i cluster saveconfig] + }As per coding guidelines, "Use clear assertions with meaningful error messages in tests."
Also applies to: 305-307
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@tests/unit/cluster/cluster-shards.tcl` at line 288, The setup invocation "R $i cluster saveconfig" currently runs without an explicit assertion; change the test to capture the response and assert it equals "OK" immediately after each "R $i cluster saveconfig" call (including the occurrences around the shard setup), using the test harness's assertion helper and a clear message like "CLUSTER SAVECONFIG failed during setup" so failures surface immediately instead of causing downstream errors.Source: Coding guidelines
🤖 Prompt for all review comments with AI agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
Inline comments:
In `@src/cluster.c`:
- Around line 776-786: Add an inline comment above the isValidAuxChar function
requesting an architectural review by the core team (mentioning `@core-team`)
because this change to cluster.c falls under the Valkey Core Engine “Critical
Escalation” trigger; reference the function name isValidAuxChar and the
invalid_charset symbol in the comment so reviewers can quickly locate the change
and audit the validation logic.
In `@src/compression_lz4.c`:
- Around line 96-113: The checks that return -1 when offset >= output_capacity
in compressionLz4CompressFeed are incorrect as labeled "retriable" because
streamWriterEnsureOutBuf/streamCompressorOutputBound should guarantee sufficient
space and LZ4F_compressBegin may already have written frame prefix
(compressor->stream_started is set), making a silent -1 unsafe; fix by doing a
preflight sizing check using streamCompressorOutputBound (or calculating
required 'need') before calling LZ4F_compressBegin so no bytes are emitted if
capacity is insufficient, or if you keep the current emit-before-check flow then
change the behavior/comments to a non-retriable fatal error (return an explicit
fatal error code and update the comment) for the offset >= output_capacity
branches (refer to compressionLz4CompressFeed, streamWriterEnsureOutBuf,
streamCompressorOutputBound, LZ4F_compressBegin, and
compressor->stream_started).
In `@src/compression_stream.c`:
- Around line 376-416: streamReaderProbe currently returns 0 both when the probe
is actually ready and when it needs more input (STREAM_READER_READ_WOULD_BLOCK),
causing callers like streamReaderGetInfo to treat an incomplete probe as valid;
change the function to return distinct values (e.g. 0 for success/ready, a new
negative or positive code like STREAM_READER_NEED_MORE_INPUT) when probe is
incomplete so callers can detect "need more input" vs "probe completed". Update
streamReaderProbe to check reader->probe.ready and return 0 only when ready,
return STREAM_READER_NEED_MORE_INPUT immediately when read_cb returns
STREAM_READER_READ_WOULD_BLOCK or when streamReaderProbeFeed returns
VKCS_PROBE_NEED_INPUT, and update callers (notably streamReaderGetInfo) to
handle the new code instead of assuming 0 means ready; use the existing helpers
streamReaderProbeBytesNeeded, streamReaderProbeFeed and the
STREAM_READER_READ_WOULD_BLOCK symbol to locate where to change control flow.
- Around line 626-659: streamReaderValidateEnd can spin when streamReaderRead
returns 0 for a compressed push-mode stream while more bytes may still arrive;
modify the loop that calls streamReaderRead so that if nread == 0 and the reader
is operating in push mode and EOF/FeedEnd hasn't been signaled (i.e.,
streamReaderFeedEnd() has not been called / reader->feed_end is false), the
function returns early (return -1) without marking STREAM_READER_ERROR_CORRUPT
so it doesn't loop forever; keep all other checks on
reader->decompressor.frame_done, streamReaderDecompressedBufAvail,
reader->compressed_buf_len and the final read_cb probe as-is.
In `@src/networking.c`:
- Line 4512: The code currently returns C_OK when server.repl_decompressor is
NULL which lets compressed streams fall back to RESP parsing; instead detect
missing decompressor and treat it as a fatal replication error: log an
explanatory error referencing server.repl_decompressor, return C_ERR (or invoke
the existing replication link-failure routine such as the replication
abort/close helper used elsewhere) so the caller will tear down the link rather
than continue parsing compressed data. Ensure you modify the branch guarding
server.repl_decompressor to fail the link path used by the surrounding
replication code.
In `@src/rdb.c`:
- Around line 3132-3134: In rdbLoadProgressCallback(), the current increment of
server.stat_net_repl_input_bytes using len overcounts when r is a decompressRio;
instead detect the conn-backed/transport case (rioIsConnBacked(r) /
decompressRio adapter) and add the transported byte delta from rioTell(r)
(current rioTell minus a stored previous tell or compute delta from the
raw/inner rio) rather than len; update the logic that increments
server.stat_net_repl_input_bytes to use the rioTell-based delta for conn-backed
rios so the metric reflects on-wire bytes while keeping len for non-conn-backed
reads.
---
Outside diff comments:
In `@src/networking.c`:
- Around line 2418-2422: The log incorrectly uses the build-time default via
compressionAlgoName(REPL_COMPRESSION_ALGO); instead, read the negotiated codec
from the replica's active compressor (or from a stored negotiated field in
repl_data) so the message reflects the real codec. Update the serverLog call in
networking.c to obtain the algo from c->repl_data->repl_compressor (or
repl_data->negotiated_compression_algo if you prefer to persist it) and pass
compressionAlgoName(...) that value, guarding for a NULL repl_compressor to fall
back to "none"/0 as currently done.
---
Duplicate comments:
In `@src/replication.c`:
- Around line 4011-4018: The code is incorrectly reading the mutable global
server.repl_compression when deciding whether a session uses compression;
instead latch and consult the per-replication-session field (e.g., add or use a
client/session member like client->repl_compression or
replClient->negotiated_repl_compression) that is set when the
REPLCONF/negotiation completes, and replace all uses of server.repl_compression
in the listed spots (the REPLCONF capa construction and the later decoder
initialization checks referenced) with that session-level field so the decision
reflects the negotiated mode for that specific replication client.
- Around line 2483-2485: When server.repl_compression is set and
replRefreshDecompression() returns C_ERR, don't just log; immediately tear down
the replication session using the failure/cleanup path used for other
replication init errors (i.e., invoke the same replication abort/connection
teardown routine used elsewhere), so the incremental sync is aborted and the
client connection is closed instead of letting compressed bytes reach the
parser; update the branch around replRefreshDecompression(), the similar
branches at the other locations (lines referenced as 3859-3861 and 4388-4390),
to call that cleanup routine right after serverLog when decompression setup
fails.
---
Nitpick comments:
In `@src/rdb.c`:
- Around line 3137-3181: Add brief doc-comments for rdbInputStreamInit,
rdbInputStreamFree, rdbInputStreamValidateEnd, and
rdbRioHasCorruptCompressedInput explaining why each helper exists and how they
behave in the passthrough vs decompression path: for rdbInputStreamInit describe
it initializes the struct to use a raw rio by default and sets raw_rio/rdb_rio;
for rdbInputStreamFree explain it cleans up the decompressor only when
initialized and resets rdb_rio to raw_rio; for rdbInputStreamValidateEnd note it
asserts initialized and returns C_OK/C_ERR based on decompressRioValidateEnd;
and for rdbRioHasCorruptCompressedInput document that it detects corrupt
compressed streams only when RIO_FLAG_STREAMING_DECOMPRESSION is set and returns
true if decompressRioGetError reports STREAM_READER_ERROR_CORRUPT. Include
mention of the initialized flag, the
RIO_FLAG_SKIP_RDB_CHECKSUM/RIO_FLAG_STREAMING_DECOMPRESSION flags where
relevant, and expected return/side-effect semantics.
In `@src/server.h`:
- Around line 462-469: Rename the compile-time macros that represent defaults so
they cannot be mistaken for runtime/negotiated values: change
REPL_COMPRESSION_ALGO -> REPL_COMPRESSION_DEFAULT_ALGO and
REPL_COMPRESSION_LEVEL -> REPL_COMPRESSION_DEFAULT_LEVEL (leave
REPL_COMPRESSION_DEFAULT_LEVEL_LZ4 and any per-algo default macros as-is),
update the comment accordingly, and then update all uses across the codebase
that currently read REPL_COMPRESSION_ALGO or REPL_COMPRESSION_LEVEL to instead
read the per-replica runtime fields (e.g., the replication state/codec fields)
so compile-time defaults are only referenced via the new *_DEFAULT_* macros.
- Around line 3037-3043: Add one-line contract comments above each declared
function (reconcileReplicaCompression, replInitCompression,
replDestroyCompression, replDecompressQueryBuf, replInitDecompression,
replDestroyDecompression, replRefreshDecompression) describing when the call is
valid (e.g., master/replica state or lifecycle ordering), who owns/manages the
singleton decompressor (which calls allocate/free it), and what the int return
values mean (explicitly list return codes such as 0 for success and non-zero/-1
for error). Keep each comment short and specific (one sentence) and place it
immediately above the corresponding prototype so cross-module callers can see
preconditions, ownership, and result semantics.
In `@tests/unit/cluster/cluster-shards.tcl`:
- Line 288: The setup invocation "R $i cluster saveconfig" currently runs
without an explicit assertion; change the test to capture the response and
assert it equals "OK" immediately after each "R $i cluster saveconfig" call
(including the occurrences around the shard setup), using the test harness's
assertion helper and a clear message like "CLUSTER SAVECONFIG failed during
setup" so failures surface immediately instead of causing downstream errors.
🪄 Autofix (Beta)
Fix all unresolved CodeRabbit comments on this PR:
- Push a commit to this branch (recommended)
- Create a new PR with the fixes
ℹ️ Review info
⚙️ Run configuration
Configuration used: Repository UI
Review profile: CHILL
Plan: Pro Plus
Run ID: d1f82bee-2232-4166-8dd8-f625cdc6db2b
📒 Files selected for processing (63)
.github/workflows/ci.yml.github/workflows/clang-format.yml.github/workflows/trigger-build-release.ymlcmake/Modules/SourceFiles.cmakecmake/Modules/ValkeySetup.cmakesrc/CMakeLists.txtsrc/Makefilesrc/cluster.csrc/commands.defsrc/commands/cluster-flushslot.jsonsrc/compression.csrc/compression.hsrc/compression_lz4.csrc/compression_lz4.hsrc/compression_repl.csrc/compression_repl.hsrc/compression_rio.csrc/compression_rio.hsrc/compression_stream.csrc/compression_stream.hsrc/config.csrc/db.csrc/debug.csrc/io_threads.csrc/module.csrc/networking.csrc/rdb.csrc/rdb.hsrc/replication.csrc/rio.csrc/rio.hsrc/sentinel.csrc/server.csrc/server.hsrc/t_hash.csrc/unit/CMakeLists.txtsrc/unit/test_compression.cppsrc/valkey-check-rdb.ctests/integration/block-repl.tcltests/integration/dual-channel-replication.tcltests/integration/rdb-compression.tcltests/integration/repl-compression.tcltests/integration/replication-2.tcltests/integration/replication-3.tcltests/integration/replication-4.tcltests/integration/replication-aof-sync.tcltests/integration/replication-buffer.tcltests/integration/replication.tcltests/integration/skip-rdb-checksum.tcltests/modules/cluster.ctests/rdma/CMakeLists.txttests/sentinel/tests/03-runtime-reconf.tcltests/support/server.tcltests/test_helper.tcltests/unit/acl-v2.tcltests/unit/cluster/acl.tcltests/unit/cluster/cluster-aux-field-validation.tcltests/unit/cluster/cluster-shards.tcltests/unit/functions.tcltests/unit/memefficiency.tcltests/unit/moduleapi/cluster.tcltests/unit/shutdown.tclvalkey.conf
💤 Files with no reviewable changes (1)
- .github/workflows/clang-format.yml
✅ Files skipped from review due to trivial changes (4)
- tests/integration/replication-2.tcl
- src/commands/cluster-flushslot.json
- tests/test_helper.tcl
- tests/rdma/CMakeLists.txt
🚧 Files skipped from review as they are similar to previous changes (11)
- cmake/Modules/SourceFiles.cmake
- src/rdb.h
- src/compression_lz4.h
- src/compression_stream.h
- src/compression.h
- tests/integration/replication-aof-sync.tcl
- .github/workflows/ci.yml
- src/rio.c
- src/valkey-check-rdb.c
- tests/integration/rdb-compression.tcl
- src/unit/test_compression.cpp
👮 Files not reviewed due to content moderation or server errors (1)
- src/module.c
| static int isValidAuxChar(unsigned char c) { | ||
| /* Reject everything up through ',' (0x2C) inclusive: control characters | ||
| * (0x00-0x1F), space !"#$%&'()*+, (0x20-0x2C), and DEL (0x7F). */ | ||
| if (c <= ',' || c == 0x7F) return 0; | ||
|
|
||
| /* List of invalid characters */ | ||
| static const char *invalid_charset = "!#$%&()*+;<>?@[]^{|}~"; | ||
| /* Reject additional characters above 0x2C (comma) that are format-significant in | ||
| * nodes.conf or otherwise unsafe. */ | ||
| static const char *invalid_charset = ";<=>?@[]^{|}~\\"; | ||
|
|
||
| /* Return true if the character is NOT in the invalid charset */ | ||
| return strchr(invalid_charset, c) == NULL; | ||
| } |
There was a problem hiding this comment.
🧩 Analysis chain
🏁 Script executed:
#!/bin/bash
set -euo pipefail
ls -la
fd -t f DEVELOPMENT_GUIDE.md . || true
if [ -f DEVELOPMENT_GUIDE.md ]; then
rg -n "core-team|architectural review|architectural|cluster\*\.c|replication\.c|rdb\.c|aof\.c" DEVELOPMENT_GUIDE.md || true
fi
# Also confirm the function exists at the referenced lines in the current src/cluster.c
if [ -f src/cluster.c ]; then
sed -n '740,820p' src/cluster.c
fiRepository: valkey-io/valkey
Length of output: 5207
🏁 Script executed:
#!/bin/bash
set -euo pipefail
rg -n "(`@core-team`|core-team|architectural review|architectural|cluster\\*\\.c|replication\\.c|rdb\\.c|aof\\.c)" \
DEVELOPMENT_GUIDE.md GOVERNANCE.md AGENTS.md MAINTAINERS.md 2>/dev/null || true
# If anything matches, show a bit of surrounding context for the key phrases.
for f in DEVELOPMENT_GUIDE.md GOVERNANCE.md AGENTS.md; do
if rg -n "(`@core-team`|core-team|architectural review|architectural)" "$f" >/dev/null 2>&1; then
echo "---- $f ----"
rg -n -C 2 "(`@core-team`|core-team|architectural review|architectural)" "$f" | head -n 80
fi
doneRepository: valkey-io/valkey
Length of output: 42
🏁 Script executed:
#!/bin/bash
set -euo pipefail
for f in DEVELOPMENT_GUIDE.md GOVERNANCE.md AGENTS.md MAINTAINERS.md CONTRIBUTING.md; do
echo "---- $f ----"
rg -n -i "core[- ]team|architectural review|architecture review|architectural|architecture|review required|required.*review|approval required|approval" "$f" || true
done
# Also search for the file-pattern rule itself (cluster*.c etc.)
rg -n "src/\\{cluster\\*\\.c,replication\\.c,rdb\\.c,aof\\.c\\}|cluster\\*\\.c|replication\\.c|rdb\\.c|aof\\.c" \
DEVELOPMENT_GUIDE.md GOVERNANCE.md AGENTS.md MAINTAINERS.md CONTRIBUTING.md || trueRepository: valkey-io/valkey
Length of output: 799
🏁 Script executed:
#!/bin/bash
set -euo pipefail
rg -n -i "`@core-team`|core[- ]team|architectural review|architecture review|architectural" . | head -n 200
# Look specifically for cluster-path governance references (file-pattern style)
rg -n -i "cluster\\*\\.c|replication\\.c|rdb\\.c|aof\\.c|src/\\{cluster\\*\\.c" . | head -n 200Repository: valkey-io/valkey
Length of output: 42
🏁 Script executed:
#!/bin/bash
set -euo pipefail
rg -n -i "`@core-team`|core[- ]team|architectural review|architecture review" \
.github DEVELOPMENT_GUIDE.md GOVERNANCE.md AGENTS.md MAINTAINERS.md CONTRIBUTING.md 2>/dev/null || true
rg -n -i "cluster\\*\\.c|replication\\.c|rdb\\.c|aof\\.c|src/\\{cluster\\*\\.c" \
.github DEVELOPMENT_GUIDE.md GOVERNANCE.md AGENTS.md MAINTAINERS.md CONTRIBUTING.md 2>/dev/null || true
rg -n -i "architectural.*review|review.*required|approval.*required" . 2>/dev/null | head -n 200 || trueRepository: valkey-io/valkey
Length of output: 637
🏁 Script executed:
#!/bin/bash
set -euo pipefail
sed -n '1,120p' .github/instructions/core-engine.instructions.md
sed -n '35,70p' .github/instructions/core-engine.instructions.mdRepository: valkey-io/valkey
Length of output: 3285
Request @core-team architectural review for src/cluster.c change
src/cluster.c is covered by the Valkey Core Engine “Critical Escalation” trigger for changes to cluster*.c; add a comment mentioning @core-team to request architectural review.
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@src/cluster.c` around lines 776 - 786, Add an inline comment above the
isValidAuxChar function requesting an architectural review by the core team
(mentioning `@core-team`) because this change to cluster.c falls under the Valkey
Core Engine “Critical Escalation” trigger; reference the function name
isValidAuxChar and the invalid_charset symbol in the comment so reviewers can
quickly locate the change and audit the validation logic.
Source: Coding guidelines
Codecov Report❌ Patch coverage is
Additional details and impacted files@@ Coverage Diff @@
## unstable #3853 +/- ##
============================================
+ Coverage 80.41% 80.51% +0.10%
============================================
Files 191 191
Lines 99454 100030 +576
============================================
+ Hits 79975 80542 +567
- Misses 19479 19488 +9
🚀 New features to boost your workflow:
|
|
Will wait for more reviews and will address these together + base branch |
|
@coderabbitai full review of the replication compression code |
|
Tip For best results, initiate chat on the files or code changes. 🧩 Analysis chain🏁 Script executed: #!/bin/bash
echo "=== src/compression_repl.c ==="
cat src/compression_repl.c
echo ""
echo "=== src/compression_repl.h ==="
cat src/compression_repl.hLength of output: 9829 🏁 Script executed: #!/bin/bash
echo "=== src/replication.c compression sections ==="
# Get all compression-related functions and their surrounding context
rg -n "replInitDecompression|replDestroyDecompression|replRefreshDecompression|reconcileReplicaCompression|REPLICA_CAPA_COMPRESSION|repl_compression|repl_decompressor|repl_compressor|REPL_COMPRESSION" src/replication.c | head -200Length of output: 2401 🏁 Script executed: #!/bin/bash
# Read full compression sections of replication.c
awk 'NR>=1 && NR<=50' src/replication.c
echo "..."
# Find line numbers of key functions
grep -n "replInitDecompression\|replDestroyDecompression\|replRefreshDecompression\|reconcileReplicaCompression\|replInitDecompression\|REPLICA_CAPA_COMPRESSION" src/replication.c | head -50Length of output: 3622 🏁 Script executed: #!/bin/bash
# Read the networking.c compression functions in detail
grep -n "replInitCompression\|replDestroyCompression\|writeToReplicaCompressed\|replDecompressQueryBuf\|postWriteToReplica\|compressionError\|compression_error\|affinity_tid\|compressed_buf\|repl_compressor" src/networking.c | head -100Length of output: 3247 🏁 Script executed: #!/bin/bash
echo "=== src/compression_stream.c key sections ==="
# streamWriterFlush, streamWriterFinish, push mode feed
grep -n "streamWriterFlush\|streamWriterFinish\|streamReaderFeed\|streamReaderInitPush\|FLUSH_SYNC\|FLUSH_CONTINUE\|FLUSH_END\|streamWriterWrite\|frame_done\|errored" src/compression_stream.c | head -100Length of output: 1808 🏁 Script executed: #!/bin/bash
echo "=== src/compression_lz4.c full ==="
cat src/compression_lz4.cLength of output: 6002 🏁 Script executed: #!/bin/bash
# Read key sections of networking.c for compression
# Get line ranges for each function
awk '/^int replInitCompression|^void replDestroyCompression|^static.*writeToReplicaCompressed|^int replDecompressQueryBuf|^static.*postWriteToReplica/{found=1; start=NR} found && NR>=start && NR<=start+100{print NR": "$0} NR>start+100{found=0}' src/networking.c | head -600Length of output: 15873 🏁 Script executed: #!/bin/bash
# Read replication.c compression functions
awk '/^int replInitDecompression|^void replDestroyDecompression|^int replRefreshDecompression|^void reconcileReplicaCompression/{found=1; start=NR} found && NR>=start && NR<=start+80{print NR": "$0} NR>start+80{found=0}' src/replication.c | head -500Length of output: 7436 🏁 Script executed: #!/bin/bash
# Check how replRefreshDecompression is called and result handling
rg -n -A3 -B3 "replRefreshDecompression\|replInitDecompression\|replDestroyDecompression" src/replication.c | head -150Length of output: 441 🏁 Script executed: #!/bin/bash
# Check config reconciliation
rg -n -A10 "reconcileReplicaCompression\|repl_compression_reconcile" src/replication.c src/config.c | head -150Length of output: 396 🏁 Script executed: #!/bin/bash
# Check io_threads.c for affinity and compression-related code
cat src/io_threads.c | head -200Length of output: 7681 🏁 Script executed: #!/bin/bash
# Check server.h for ClientReplicationData and compression fields
rg -n -A5 -B2 "affinity_tid\|repl_compressor\|compression_error\|repl_stream_decoder\|repl_decompressor\|REPL_COMPRESSION\|REPLICA_CAPA_COMPRESS" src/server.h | head -200Length of output: 491 🏁 Script executed: #!/bin/bash
# Check compression.c dispatch layer
cat src/compression.cLength of output: 5002 🏁 Script executed: #!/bin/bash
# Check full compression_stream.c
wc -l src/compression_stream.c
# Check key areas: envelope, push mode, error handling
awk 'NR>=1 && NR<=100' src/compression_stream.c
echo "..."
grep -n "streamReaderFeed\|streamReaderFeedEnd\|feed_cap\|feed_queue\|NEEDS_INPUT\|frame_done\|errored\|FLUSH_SYNC\|FLUSH_END\|streamWriterFlush\b" src/compression_stream.c | head -80Length of output: 6041 🏁 Script executed: #!/bin/bash
echo "=== affinity_tid usage ==="
rg -n "affinity_tid" src/ | head -50
echo ""
echo "=== server.h ClientReplicationData struct ==="
awk '/typedef struct ClientReplicationData|struct ClientReplicationData/{found=1; start=NR} found && NR>=start && NR<=start+60{print NR": "$0} NR>start+60{found=0}' src/server.h
echo ""
echo "=== REPL_COMPRESSION_ALGO in server.h ==="
rg -n "REPL_COMPRESSION_ALGO\|REPL_COMPRESSION_LEVEL\|REPL_COMPRESSION_DEFAULT\|REPL_STREAM_DECODER" src/server.h | head -20Length of output: 6185 🏁 Script executed: #!/bin/bash
echo "=== dual channel decompression in replication.c ==="
awk 'NR>=3480 && NR<=3530' src/replication.c
echo "..."
awk 'NR>=2470 && NR<=2500' src/replication.c
echo ""
echo "=== reconcileReplicaCompression missing cases ==="
# check whether compression is ever enabled on a non-ONLINE replica
rg -n "replicaInitCompressionOnPsync\|REPLICA_STATE_ONLINE\|replInitCompression" src/replication.c | head -30Length of output: 4714 🏁 Script executed: #!/bin/bash
echo "=== non-atomic repl_compressed_bytes and repl_uncompressed_bytes ==="
rg -n "repl_compressed_bytes_total\|repl_uncompressed_bytes_total" src/ | head -30
echo ""
echo "=== writeToReplicaCompressed affinity section ==="
awk 'NR>=2560 && NR<=2650' src/networking.cLength of output: 4165 🏁 Script executed: #!/bin/bash
echo "=== io_threads.c private inbox / affinity path ==="
rg -n "affinity_tid\|private_inbox\|JOB_REQ_WRITE_CLIENT\|io_private_inbox" src/io_threads.c | head -40
echo ""
echo "=== trySendWriteToIOThreads or equivalent ==="
rg -n "affinity_tid\|io_last_reply_block\|io_last_bufpos" src/networking.c | head -30
rg -n "affinity_tid\|io_last_reply_block\|io_last_bufpos" src/io_threads.c | head -30Length of output: 550 🏁 Script executed: #!/bin/bash
echo "=== check what happens when repl_compressor->out_buf is empty after flush ==="
awk 'NR>=2620 && NR<=2660' src/networking.c
echo ""
echo "=== check streamWriterFlush behavior for empty input ==="
awk 'NR>=180 && NR<=310' src/compression_stream.cLength of output: 6738 🏁 Script executed: #!/bin/bash
echo "=== full writeToReplicaCompressed send path ==="
awk 'NR>=2637 && NR<=2680' src/networking.c
echo ""
echo "=== replDecompressQueryBuf dual channel path ==="
awk 'NR>=3488 && NR<=3520' src/replication.cLength of output: 3358 🏁 Script executed: #!/bin/bash
echo "=== check compression capability advertised only conditionally ==="
awk 'NR>=4000 && NR<=4030' src/replication.c
echo ""
echo "=== replication.c lines around PSYNC init compression ==="
rg -n "replicaInitCompressionOnPsync\|shouldEnableReplicaCompression" src/replication.c | head -20Length of output: 1577
Full Review: Replication Compression ImplementationArchitecture OverviewThe design is well-structured. Per-replica compression uses a Issues Requiring Attention1.
replRefreshDecompression ignoring its return value |
| Area | Assessment |
|---|---|
| Thread teardown | waitForClientIO before streamWriterDestroy — correct and critical |
| Cursor pinning | out_buf_pos tracking keeps logical offset stable during partial compressed sends |
read_reploff accounting |
replDecompressQueryBuf correctly converts transport bytes ↔ logical bytes |
| Back-pressure | 64 MB push feed cap (REPL_STREAM_DECODER_FEED_CAP) and per-batch limits (1 MB compress, 256 MB decompress) bound memory/latency |
| Frame-done detection | REPL_DECODE_FRAME_DONE path correctly logs and disconnects |
| IO-thread guard | trySendReadToIOThreads rejects read offloading when server.repl_decompressor is active |
| VKCS lazy envelope | streamWriterEnsureEnvelope correctly defers the 8-byte envelope until first write, so an unused writer leaves no stub |
| Decompressor memory bounds | REPL_COMPRESSION_RETAIN_LIMIT buffer shrink in replCompressorResetBatch and replDecompressorDecode prevents peak-allocation retention |
| Metrics | Atomic per-replica counters + server-level INFO fields are comprehensive |
| Capability gating | Replica only advertises compression capa when server.repl_compression=yes; primary gates compression on the per-replica REPLICA_CAPA_COMPRESSION bit |
Summary of Action Items
| # | Severity | Status | File |
|---|---|---|---|
#1 replRefreshDecompression return value |
Fixed | replication.c | |
#2 Hardcoded REPL_COMPRESSION_ALGO in error log |
Low | Open | networking.c:2421 |
#3 Affinity claim bug |
Not present | — | |
#4 LZ4 header/capacity bug on retry |
Medium | Open | compression_lz4.c |
#5 FLUSH_SYNC per batch overhead |
Low | Open (design tradeoff) | compression_stream.c |
#6 assert() vs. error return in compressionLz4CompressorInit |
Trivial | Open | compression_lz4.c |
|
@roshkhatri can you please rebase with PR #3531. It's a little hard for me to pick out the replication compression code to review. |
@sarthakaggarwal97, Yes I was working on it and #3531 was moving, so I was waiting for it to be stabilized, do you think you major refactors are done? |
|
Yes. It's annoying to keep rebasing, but hoping the #3531 is merged soon. |
…-io#3942, valkey-io#3897) These came in via the unstable merge in valkey-io#3853 but are unrelated to replication compression. Restore cluster_legacy.c, db.c, and failover.tcl to the rio-pr base so this review branch shows only the replication-compression delta. Signed-off-by: Sarthak Aggarwal <sarthagg@amazon.com>
sarthakaggarwal97
left a comment
sarthakaggarwal97
left a comment
There was a problem hiding this comment.
Took a look again. Sharing some more review comments.
3155962 to
ef20044
Compare
|
@CodeRabbit review the replication compression implementation only |
|
Tip For best results, initiate chat on the files or code changes. 🧩 Analysis chain🏁 Script executed: #!/bin/bash
echo "=== src/compression_repl.c ==="
cat src/compression_repl.c
echo ""
echo "=== src/compression_repl.h ==="
cat src/compression_repl.hLength of output: 14189 🏁 Script executed: #!/bin/bash
echo "=== src/compression_lz4.c ==="
cat src/compression_lz4.cLength of output: 5992 🏁 Script executed: #!/bin/bash
echo "=== replication.c: compression functions ==="
rg -n "replInitDecompression|replDestroyDecompression|replRefreshDecompression|reconcileReplicaCompression|REPLICA_CAPA_COMPRESSION|repl_compression\b|repl_decompressor|repl_compressor\b" src/replication.c | head -80
echo ""
echo "=== replication.c: function bodies ==="
awk '/^int replInitDecompression|^void replDestroyDecompression|^int replRefreshDecompression|^void reconcileReplicaCompression/{found=1; start=NR} found && NR>=start && NR<=start+80{print NR": "$0} NR>start+80{found=0}' src/replication.c | head -500Length of output: 11026 🏁 Script executed: #!/bin/bash
echo "=== networking.c: compression functions ==="
rg -n "replInitCompression|replDestroyCompression|writeToReplicaCompressed|replDecompressQueryBuf|postWriteToReplica|REPL_COMPRESSION_ALGO\b" src/networking.c | head -60
echo ""
echo "=== networking.c: key function bodies ==="
awk '/^int replInitCompression|^void replDestroyCompression|^static.*writeToReplicaCompressed|^int replDecompressQueryBuf|^static.*postWriteToReplica/{found=1; start=NR} found && NR>=start && NR<=start+100{print NR": "$0} NR>start+100{found=0}' src/networking.c | head -600Length of output: 13392 🏁 Script executed: #!/bin/bash
echo "=== server.h: ClientReplicationData and valkeyServer compression fields ==="
awk '/typedef struct ClientReplicationData|struct ClientReplicationData/{found=1; start=NR} found && NR>=start && NR<=start+70{print NR": "$0} NR>start+70{found=0}' src/server.h
echo ""
rg -n "REPL_COMPRESSION_ALGO\|REPL_COMPRESSION_LEVEL\|REPL_COMPRESSION_DEFAULT\|REPLICA_CAPA_COMPRESS\|repl_decompressor\|repl_stream_decoder" src/server.h | head -30
echo ""
echo "=== io_threads.c: replication / compression guards ==="
rg -n "repl_decompressor\|repl_compression\|repl_stream_decoder\|JOB_REQ_WRITE_CLIENT" src/io_threads.c | head -30Length of output: 6689 🏁 Script executed: #!/bin/bash
echo "=== compression_stream.c: streamWriterFlush and push-mode feed ==="
rg -n "streamWriterFlush\|streamReaderFeed\|FLUSH_SYNC\|FLUSH_CONTINUE\|FLUSH_END\|feed_cap\b\|stream_started\|errored\b" src/compression_stream.c | head -60
echo ""
echo "=== compression.c: streamDecompressorFeed error handling ==="
rg -n "errored\|frame_done\|input_consumed\|streamDecompressorFeed\|streamCompressorFeed" src/compression.c | head -40Length of output: 613 🏁 Script executed: #!/bin/bash
echo "=== compression_stream.c streamWriterFlush section ==="
grep -n "streamWriterFlush\|FLUSH_SYNC\|FLUSH_CONTINUE\|FLUSH_END\|stream_started\|errored\|streamWriterFeedAndEmit\|LZ4F_flush\|LZ4F_compressEnd" src/compression_stream.c | head -50
echo ""
echo "=== Check VCS_MAGIC vs VKCS_MAGIC naming ==="
grep -rn "VCS_MAGIC\|VKCS_MAGIC\|VCS_ENVELOPE_SIZE\|VKCS_ENVELOPE_SIZE\|VCS_MAGIC_SIZE" src/ | head -20
echo ""
echo "=== io_threads.c: full list of compression guards ==="
grep -n "compress\|decompress\|repl_decompressor\|repl_compression\|provisional" src/io_threads.c | head -30Length of output: 3458 🏁 Script executed: #!/bin/bash
echo "=== replication.c: repl_provisional_compression usage ==="
rg -n "repl_provisional_compression" src/ | head -30
echo ""
echo "=== replication.c: lines around replRefreshDecompression call sites ==="
awk 'NR>=2520 && NR<=2535' src/replication.c
echo "..."
awk 'NR>=3896 && NR<=3910' src/replication.c
echo "..."
awk 'NR>=4428 && NR<=4442' src/replication.c
echo ""
echo "=== server.h: repl_provisional_compression and related server fields ==="
rg -n "repl_provisional_compression\|repl_decompressor\|repl_decompression_errors\|repl_apply_batches\|REPL_STREAM_DECODER" src/server.h | head -30Length of output: 4240 🏁 Script executed: #!/bin/bash
echo "=== LZ4 compressFeed capacity guard comment ==="
awk 'NR>=90 && NR<=160' src/compression_lz4.c
echo ""
echo "=== compression_stream.c: streamWriterFlush full function ==="
awk '/^int streamWriterFlush\b/{found=1; start=NR} found && NR>=start && NR<=start+30{print NR": "$0} NR>start+30{found=0}' src/compression_stream.c
echo ""
echo "=== compression_stream.c: streamWriterFeedAndEmit function ==="
awk '/^static.*streamWriterFeedAndEmit|^static int streamWriterFeedAndEmit/{found=1; start=NR} found && NR>=start && NR<=start+40{print NR": "$0} NR>start+40{found=0}' src/compression_stream.c
echo ""
echo "=== compression.c: streamCompressorFeed full function ==="
awk '/^ssize_t streamCompressorFeed\b/{found=1; start=NR} found && NR>=start && NR<=start+30{print NR": "$0} NR>start+30{found=0}' src/compression.cLength of output: 6083 🏁 Script executed: #!/bin/bash
echo "=== reconcileReplicaCompression: check for ONLINE-only mismatch detection ==="
awk 'NR>=102 && NR<=150' src/replication.c
echo ""
echo "=== replication.c: capability advertisement (REPLCONF capa) ==="
awk 'NR>=4050 && NR<=4075' src/replication.c
echo ""
echo "=== replication.c: REPLICA_STATE_ONLINE and compression init on PSYNC ==="
rg -n "REPLICA_STATE_ONLINE\|replInitCompression\|shouldEnableReplicaCompression" src/replication.c | head -30Length of output: 3930
Replication Compression — Full ReviewWhat Was Fixed Since the Previous Reviews
Issues Still Open1.
|
| # | Severity | Status | Location |
|---|---|---|---|
| LZ4 header/capacity retry bug | Medium | 🔴 Open | compression_lz4.c |
replRefreshDecompression failure not acted on |
Medium | 🔴 Open | replication.c:2526,3902,4434 |
FLUSH_SYNC per-batch overhead |
Low | 🟡 Open (design tradeoff) | compression_stream.c:291 |
assert() vs error return in LZ4 init |
Trivial | 🟡 Open | compression_lz4.c |
|
@CodeRabbit That C_ERR branch for replRefreshDecompression is unreachable (replDecompressorCreate only fails on allocation, which zmalloc aborts on; codec init is lazy in replDecompressorDecode and already returns REPL_DECODE_ERR -> disconnect), so I removed the dead C_ERR plumbing and made replRefreshDecompression void |
|
Tip For best results, initiate chat on the files or code changes.
Issue |
b43ffc7 to
41817a9
Compare
41817a9 to
aaa06ff
Compare
|
Failures on this PR are related to #4612 |
zuiderkwast
left a comment
zuiderkwast
left a comment
There was a problem hiding this comment.
I've done a pass on this. It looks mostly clean and well-structured. I have some question and minor comments.
I haven't read everything very carefully yet. I'll look a little more.
| /* Compressed write path for replicas on either the IO thread or the main thread. */ | ||
| static void writeToReplicaCompressed(client *c) { | ||
| replicaCompressionState *compression = c->repl_data->repl_compression; |
There was a problem hiding this comment.
Just a question.
The replicaCompressionState contains a pending-compressed buffer and some metadata. We would need the exact same struct for a compressed client command/responses stream, right? When we add that, we'd move the struct from c->repl_data->repl_compression to client-level c->repl_compression?
There was a problem hiding this comment.
Not necessarily the exact same struct. replicaCompressionState also contains replication-specific batch accounting, backlog progress, and metrics. For compressed client commands or responses, I would maybe separate client-level state while reusing the lower-level compressor and buffering helpers. Let me look into it again and I can open a followup PR
| size_t output_budget, | ||
| ssize_t *decoded_bytes, | ||
| bool *full_read) { | ||
| uint8_t wire_buf[PROTO_IOBUF_LEN]; |
There was a problem hiding this comment.
Just a note.
I assume it's good to keep this on the stack, so it'll be in L1 cache automatically. The alternative is to have a wire_buf in the client's compression state, but that'd be an L1 cache miss. Right?
There was a problem hiding this comment.
Yes, that was the intent. Keeping it on the stack avoids a persistent allocation and its associated lifetime and accounting. It might benefit from normal stack-cache locality, although being on the stack does not guarantee that it is already in L1.
zuiderkwast
added
release-notes
zuiderkwast
left a comment
zuiderkwast
left a comment
There was a problem hiding this comment.
High-level approve with a few non-blocking comments. Address the easy ones and then feel free to merge.
|
Thanks you so much, I will ask @sarthakaggarwal97 to merge it once I address them |
Add negotiated compression for the incremental replication stream, state-driven runtime reconciliation, explicit checksum policy, resumable replica-side decompression, and hardened recovery behavior. Signed-off-by: Roshan Khatri <rvkhatri@amazon.com>
aaa06ff to
9774d13
Compare
66f9618
into
valkey-io:unstable
|
Thank you @zuiderkwast for taking a look and your review. Thanks @roshkhatri for creating the follow up issues. We can work together and get those follow ups in as well before GA happens! |
|
Thank you so much @sarthakaggarwal97 @zuiderkwast @hpatro for all the valuable reviews and feedbacks! |
|
@sarthakaggarwal97 / @roshkhatri Could one of you add a design-doc for this feature to the core? |
|
Sure @hpatro, I will add it! |
Follow-up to #3853, addresses #3853 (comment). The LZ4 contexts go through the custom zmalloc allocator, so their memory is inside `used_memory` but not attributed to anything and shows up under `used_memory_dataset`. There are two of them: the per-replica `LZ4F_cctx` on the primary, and the stream reader on the replica (`LZ4F_dctx` plus a retained input buffer), which lives on the server struct so per-client accounting never sees it. `zmalloc_size(ctx)` doesn't work here. The context struct is ~200 bytes and the staging buffers it allocates lazily are ~80 KiB. So the allocator callbacks now count the bytes into a `ctx_memory` field on the owning stream, via `LZ4F_CustomMem.opaqueState`. `getClientMemoryUsage()` picks up the compressor context in the existing replica block, and charges the replica-side reader to the primary client, so it shows up in `mem_clients_normal` like other client memory. Added a unit test that checks the counter grows when the streaming buffers are allocated and drops back to zero on free. Signed-off-by: Sarthak Aggarwal <sarthagg@amazon.com>
5 participants
Summary
Adds per-replica streaming compression for replication, with LZ4 as the first codec. This PR is now rebased on the full-sync compression support merged in #4075 and extends the negotiated stream into steady-state replication.
Compression is disabled by default with
repl-compression no. A replica with replication compression enabled advertises the existing codec-specificREPLCONF capa lz4, and a primary compresses only when its own configuration enables LZ4 and the replica advertised support. Older or opted-out replicas continue using plaintext, so mixed deployments fall back safely.The configuration applies as follows:
repl-compression; when it is not selected,rdbcompressionmay still apply per-string LZFrdbcompression, because the generated RDB may become the persisted snapshotrepl-compressionCompression runs on the existing write path, on an IO thread or the main-thread fallback, with a bounded 1 MB raw-input batch. Replication-buffer cursor updates remain on the main thread. On the replica,
streamPushReaderis installed when LZ4 was advertised, detects theVCS_STREAM_REPLenvelope, and decodes directly into the query buffer; plaintext streams pass through unchanged.Headline results (BlockMesh tweets, 3M keys × ~315 B):
Configuration added:
repl-compression:no(default),yes(currently LZ4), orlz4Capability used:
lz4: the replica accepts LZ4 streaming-compressed replication payloadsDesign and testing details
Data flow
Primary: replication backlog → stream compressor → staging buffer → socket
Replica: socket → stream reader → query buffer
The primary compresses at most 1 MB of raw backlog data per batch. Its backlog cursor remains pinned until the corresponding compressed buffer is fully sent, preventing partial writes or
EAGAINfrom losing data.The replica reads wire data in 16 KB chunks and decodes directly into its query buffer. Decoded offsets remain in the logical replication domain, preserving ACK, WAIT, and partial-resynchronization behavior.
Capability negotiation
The capability names the codec rather than the replication phase. A replica advertises
lz4whenrepl-compressionenables LZ4. The same negotiated capability is used by diskless full sync and the steady-state stream. Disk-based full sync selects its codec fromrdbcompressionbut still requires every participating replica to advertise support for that codec.If the primary and replica do not share an enabled codec, steady-state replication remains plaintext; a full-sync RDB may still use per-string LZF according to
rdbcompression.Integrity
LZ4 block checksums are enabled for the steady-state replication stream. Its content checksum is disabled because the frame stays open for the lifetime of the connection and is never finalized. Full-sync and regular RDB streams retain their existing checksum policy.
Invalid envelopes, corrupt payloads, unsupported codecs, and unexpected frame termination disconnect the link.
Runtime changes
Changing
repl-compressionreconnects only links whose active transport no longer matches the committed configuration. The replica remembers whether LZ4 was advertised during the current upstream handshake, and replication cron reconnects only when that state differs from the final configuration.Changes made during handshake or full sync do not interrupt it. The link finishes that sync and reconnects afterward if it still differs from the committed configuration. Reconnection attempts partial resynchronization first, but falls back to a full sync if the backlog no longer covers the replica offset. If the setting changes away and back before cron runs, an already matching link remains connected.
Observability
For active compressed links, each replica entry in
INFO replicationincludes:repl_compressionrepl_compressed_bytesrepl_uncompressed_bytesThese per-link counters cover completed batches and reset when the replica reconnects.
Testing
tests/integration/repl-compression.tclcovers negotiation, plaintext passthrough, batch boundaries, partial resynchronization, cursor pinning, interrupted frames, runtime configuration changes, multiple replicas, chained replication, IO threads, and dual-channel replication.The module API test covers decoder progress while a long-running command yields to the event loop.
The full-sync tests cover diskless and disk-based policy, capability fallback, grouped replicas, piggybacking, AOF fallback, checksums, interrupted transfers, and byte accounting.
Unit tests cover envelope validation, checksum policy, corrupt input, output limits, and buffered decoder output.