Replication compression configs and handshake negotiation - #3577
Closed
roshkhatri wants to merge 15 commits into
Closed
roshkhatri wants to merge 15 commits into
roshkhatri wants to merge 15 commits into
Conversation
❌ Provenance Check AlertPotential code similarities detected with upstream repository.
This check was performed automatically by the Provenance Guard Action. |
roshkhatri
force-pushed
the
repl-compression-handshake-v2
branch
from
April 28, 2026 18:58
5c1c871 to
8cd87d7
Compare
❌ Provenance Check AlertPotential code similarities detected with upstream repository.
This check was performed automatically by the Provenance Guard Action. |
Codecov Report❌ Patch coverage is
Additional details and impacted files@@ Coverage Diff @@
## unstable #3577 +/- ##
============================================
+ Coverage 76.65% 77.08% +0.43%
============================================
Files 162 168 +6
Lines 80662 82707 +2045
============================================
+ Hits 61830 63757 +1927
- Misses 18832 18950 +118
🚀 New features to boost your workflow:
|
roshkhatri
force-pushed
the
repl-compression-handshake-v2
branch
from
May 8, 2026 20:22
8cd87d7 to
4dccdd4
Compare
Add streaming LZ4-backed RDB compression with rio decorators, stream envelope handling, integration changes, and the follow-up fixes and config cleanup needed on top of unstable. Signed-off-by: Sarthak Aggarwal <sarthagg@amazon.com>
Signed-off-by: Sarthak Aggarwal <sarthagg@amazon.com>
- Remove dead code: rdbIsValidMagic() and unused #include <string.h> in rdb.h - Remove redundant first RIO_FLAG_SKIP_RDB_CHECKSUM set in rdbSaveInternal - Remove unrelated changes: config_parse_depth, USE_FAST_FLOAT, write-make-settings - Validate full 8-byte VKCS envelope in aof.c rdbFileUsesStreamingCompression - Add SAFETY comment for rdbRioHasCorruptCompressedInput cast invariant - Rename all snake_case identifiers to camelCase per Valkey conventions: types (compression_algo_t -> compressionAlgo, stream_compressor_t -> streamCompressor, compress_rio_t -> compressRio, etc.), functions (stream_writer_create -> streamWriterCreate, compress_rio_finish -> compressRioFinish, write_vkcs_envelope -> writeVkcsEnvelope, etc.), and static variables (compression_lz4_codec_impl -> compressionLz4CodecImpl) - Drop _t suffix from all types to match Valkey convention - Fix typo: streamWriterIsErrord -> streamWriterIsErrored - Replace silent dummy buffer allocation with assert(needed > 0) in streamWriterEnsureOutBuf Signed-off-by: Sarthak Aggarwal <sarthagg@amazon.com>
* Address streaming RDB compression review * Skip RDB CRC for streaming compression * Remove brittle 32-bit compression unit test --------- Co-authored-by: Sarthak Aggarwal <sarthagg@amazon.com> Signed-off-by: Sarthak Aggarwal <sarthagg@amazon.com>
--------- Signed-off-by: Sarthak Aggarwal <sarthagg@amazon.com>
roshkhatri
marked this pull request as ready for review
May 12, 2026 21:20
Contributor
📝 WalkthroughWalkthroughAdds a vendored LZ4 library and streaming VKCS framing, implements a compression framework with an LZ4 adapter, adds rio partial-read and decorator adapters, integrates streaming compression into RDB save/load and AOF/replication flows, updates build/link wiring, and adds tests and config/docs. ChangesStreaming Compression Integration
Estimated code review effort🎯 5 (Critical) | ⏱️ ~120 minutes Suggested reviewers
|
![coderabbitai[bot]](https://avatars.githubusercontent.com/in/347564?s=60&v=4){kind=small}
Contributor
There was a problem hiding this comment.
Actionable comments posted: 11
Caution
Some comments are outside the diff and can’t be posted inline due to platform limitations.
⚠️ Outside diff range comments (1)
src/aof.c (1)
1009-1054:⚠️ Potential issue | 🟠 Major | ⚡ Quick winRequest
@core-teamarchitectural review for this aof.c change.This modifies AOF restart logic and persistence-path decision making in a protected file scope. Per coding guidelines, changes to
src/aof.crequire explicit core-team architectural sign-off before merge.🤖 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/aof.c` around lines 1009 - 1054, This change touches core AOF restart and persistence-path logic (functions restartAOFWithSyncRdb and rdbFileUsesStreamingCompression, referencing server.rdb_filename) and therefore needs an explicit core-team architectural review before merging; please add a clear PR-level request for core-team sign-off (e.g., tag core-team reviewers and add an "architectural review required" note in the PR description), include an explanation of the behavioral change and rationale, and ensure the commit message references the review request so CI/maintainers do not merge without the required approval.
🧹 Nitpick comments (5)
src/replication.c (1)
1449-1450: Please get@core-teamreview on the replication handshake changes.This touches capability negotiation in
replication.c, which is exactly the path the repo asks to route through architectural review.As per coding guidelines, "Request
@core-teamarchitectural review for changes to cluster*.c, replication.c, rdb.c, or aof.c".Also applies to: 3859-3886
🤖 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 1449 - 1450, This change touches replication capability negotiation in replication.c (the block using strcasecmp(objectGetVal(c->argv[j + 1]), REPLICA_CAPA_COMPRESSION_STR) and the REPLICA_CAPA_COMPRESSION flag) and must get an architectural review from the `@core-team`; update the PR by explicitly requesting `@core-team` review and add a short in-code comment/TODO next to the capability negotiation (and the related range you modified around lines 3859-3886) noting "requires `@core-team` architectural review" so reviewers see it in-context and the PR cannot be merged without that review.src/aof.c (1)
1009-1026: ⚡ Quick winAdd a short rationale comment above
rdbFileUsesStreamingCompression.The helper drives a non-obvious fallback path in AOF restart; please document why envelope detection is sufficient for this decision.
Suggested patch
+/* Detect whether the synced RDB starts with a streaming envelope. + * We use this to avoid reusing a streamed/compressed RDB as AOF base and + * force BGREWRITEAOF fallback for a plain, reusable base file. */ static int rdbFileUsesStreamingCompression(const char *filename) {As per coding guidelines
src/**/*.{c,h}: "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/aof.c` around lines 1009 - 1026, Add a short rationale comment above rdbFileUsesStreamingCompression explaining why inspecting only the VKCS envelope/header is sufficient to decide the streaming-compression fallback during AOF restart: note that the envelope contains the compression/streaming metadata used by streamReadEnvelopeInfo, that this probe is cheap and reliable without scanning the whole RDB, and that the outcome drives the non-obvious AOF restart fallback path; reference streamReadEnvelopeInfo and STREAM_KIND_RDB so future readers understand the trust boundary and intent.tests/integration/replication-aof-sync.tcl (1)
192-194: ⚡ Quick winReplace fixed sleep with condition-based synchronization.
Line 192 uses
after 1000, which makes this test timing-sensitive and flaky. Gate the negative log assertion on a deterministic state transition (e.g., afterwaitForBgrewriteaofor another explicit completion signal), then assert the log condition.As per coding guidelines: "Avoid timing-dependent tests; use proper synchronization (test reliability)".
🤖 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/replication-aof-sync.tcl` around lines 192 - 194, The fixed sleep `after 1000` is making the test flaky; replace it with a deterministic synchronization by calling `waitForBgrewriteaof $replica` before performing the negative log assertion, then perform `assert {![log_file_matches $replica_log "*Reused RDB file from primary sync as AOF base file*"]}`; locate the block that currently has `after 1000` and move or replace that line with the existing `waitForBgrewriteaof` call so the log check runs only after the AOF rewrite completion.tests/integration/repl-compression.tcl (1)
122-136: ⚡ Quick winRename the backward-compatibility test to match actual coverage.
This test does not run an older binary; it validates a same-version replica that does not advertise compression. Rename it to avoid over-claiming compatibility scope.
As per coding guidelines: "Use descriptive test names that explain what is being tested".
🤖 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/repl-compression.tcl` around lines 122 - 136, Rename the test case title string in the test block that currently reads "Backward compatibility - older replica without capa compression connects successfully" to a descriptive name reflecting what it actually verifies, e.g. "Replica without advertised compression connects successfully"; update the test header (the test { ... } line) that wraps the start_server / $replica replicaof / wait_for_condition sequence so the new name appears in test output and logs (no code changes inside start_server, $replica replicaof, wait_for_condition, s 0 master_link_status, or $replica replicaof no one are required).src/rdb.c (1)
1565-3792: Request@core-teamarchitectural review for thisrdb.cchange set.This touches RDB save/load framing, checksum semantics, and streaming paths; please explicitly route this PR through
@core-teambefore merge.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/rdb.c` around lines 1565 - 3792, The change set modifies critical persistence and replication code (rdbSaveInternal, rdbLoadRio, rdbLoadRioWithLoadingCtxScopedRdb and related framing/checksum/streaming functions) and therefore must be routed for an architecture review by the core-team before merging; update the PR by adding an explicit request-for-review line tagging `@core-team` in the PR description, add the "requires-arch-review" label (or equivalent), and include a short summary of the affected symbols (rdbSaveInternal, rdbSaveToFile/rdbSaveBackground, rdbLoadRio/rdbLoadRioWithLoadingCtx, rdbInputStreamPrepare/rdbInputStreamValidateEnd) plus the rationale that framing/checksum/streaming semantics changed so reviewers can focus their review.
🤖 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-1623: The computation of srcEnd uses pointer arithmetic on
srcStart even when srcBuffer is NULL in LZ4F_decompress, causing undefined
behavior; update the initialization in LZ4F_decompress so srcStart and srcEnd
are computed safely by treating a NULL srcBuffer as a valid "no input" sentinel
(e.g. set srcStart to NULL and srcEnd to srcStart when srcBuffer is NULL, or
derive srcEnd using a conditional like dstEnd does), ensuring you adjust
variables srcStart, srcEnd and srcPtr consistently so subsequent logic (which
reads from srcPtr up to srcEnd) safely handles the NULL/no-input case without
performing NULL + offset arithmetic.
In `@src/compression_lz4.c`:
- Around line 85-86: The functions compressionLz4CompressFeed and
compressionLz4DecompressFeed (and the other LZ4 feed/finish entry points around
the same area) must guard against reuse of a context that has previously
errored: at each function entry check sc and sc->ctx as well as sc->errored and
immediately return an error (-1) if errored is true to avoid calling LZ4 APIs on
a UB state; keep the existing behavior of setting sc->errored = true on
failures, but add the sticky-error guard at the top of each relevant function
(e.g., compressionLz4CompressFeed, compressionLz4DecompressFeed and the related
feed/finish helpers) so once errored no further LZ4 calls are made until the
context is reinitialized or freed.
- Around line 69-74: The addition in compressionLz4OutputBound can overflow
size_t; update compressionLz4OutputBound to perform checked arithmetic when
summing LZ4F_compressBound(input_len, &lz4f_prefs), LZ4F_HEADER_SIZE_MAX and
LZ4F_compressBound(0, &lz4f_prefs) — detect overflow at each add (e.g., check if
a + b < a) and if overflow would occur return SIZE_MAX (or another agreed
sentinel) so callers know allocation is impossible, otherwise return the safe
sum; reference the function name compressionLz4OutputBound and the symbols
lz4f_prefs, LZ4F_HEADER_SIZE_MAX, and LZ4F_compressBound when locating the
change.
In `@src/compression.c`:
- Around line 163-188: The function streamDecompressFeed can call
codec_impl->decompress_feed with a NULL input pointer when input_len > 0; add a
guard in streamDecompressFeed to detect this case (e.g. if input_len > 0 &&
input == NULL) and mark sd->errored = true then return -1 before looking up or
invoking codec_impl->decompress_feed so the codec never receives a NULL buffer
with non-zero length.
In `@src/replication.c`:
- Around line 3880-3886: The dual-channel handshake in
dualChannelReplHandleHandshake() omits the REPLICA_CAPA_COMPRESSION capability
so replicas that fall back to dedicated RDB-channel sync never negotiate
compression; update dualChannelReplHandleHandshake() to include
REPLICA_CAPA_COMPRESSION_STR in its REPLCONF/capa capability list whenever
server.repl_compression && use_diskless_load (or mirror the same condition used
where argv/lens/argc are set for the single-channel path), or refactor the
capability-list construction into a small shared helper used by both the
single-channel handshake and dualChannelReplHandleHandshake() so both paths
consistently include REPLICA_CAPA_COMPRESSION.
In `@src/rio.c`:
- Around line 249-275: The additive checks on r->io.conn.read_limit can
overflow; replace any comparisons that use read_so_far + X with
subtraction-based guards: compute size_t remaining = (r->io.conn.read_limit >
r->io.conn.read_so_far) ? r->io.conn.read_limit - r->io.conn.read_so_far : 0 and
use remaining for comparisons (e.g. for the strict_limit check compare remaining
< min_read and return EOVERFLOW), and in the read loop cap toread by
remaining-after-avail only after verifying remaining > avail (e.g.
remaining_after_avail = remaining > avail ? remaining - avail : 0; if (toread >
remaining_after_avail) toread = remaining_after_avail). Apply this in the checks
referencing r->io.conn.read_limit, r->io.conn.read_so_far, min_read, avail, and
toread to avoid size_t wraparound.
In `@tests/integration/rdb-compression.tcl`:
- Around line 250-304: The test leaves the temporary partial snapshot file
partial-vkcs.rdb (variable partial_rdb) on disk; delete it at the end of the
test to avoid residue by adding a safe cleanup call such as catch {file delete
-force $partial_rdb} after the final assertion (after assert_match "*Error*"
$err) so the file is always removed even if delete fails; you can also wrap
earlier early-failure paths to delete partial_rdb where appropriate to ensure
cleanup in all exit paths.
- Around line 459-466: This test relies on prior tests to have established
replication; make it self-contained by explicitly initializing a full sync
within the test before setting repl:post-sync: configure/start replication from
$primary to $replica (e.g., issue the replication setup/replicaof call or start
the replica process), perform a write on $primary that you can wait for on
$replica to confirm the full sync completed, then proceed to set $primary
repl:post-sync and use the existing wait_for_condition on $replica; update the
test named "Incremental replication continues after LZ4 full sync" to include
these setup steps so it does not depend on test execution order.
In `@tests/integration/repl-compression.tcl`:
- Around line 42-95: After the replica reaches up, assert the negotiated
capability on the primary-side replica metadata instead of only checking
master_link_status: call set info [$primary info replication] (as already used
in the second test) and then use assert_match/assert_no_match to verify the
presence or absence of the compression capability token for slave0; e.g. in the
"replcompression yes and diskless load" test add assert_match
"*slave0:*capa=*compression*" (or the exact capability substring your server
reports) and in the "replcompression no" and "replcompression yes but
disk-backed load" tests add assert_no_match "*slave0:*capa=*compression*" to
make the tests deterministic, using the existing assert_match/assert_equal
helpers to provide clear failure messages.
In `@tests/integration/valkey-check-rdb.tcl`:
- Around line 64-65: Remove the misplaced "checksum disabled" assertion from
this test: delete the assert_match line that expects "RDB file was saved with
checksum disabled: skipped checksum for this transfer" and leave only the
relevant assertions (e.g., the existing assert_no_match {*Checksum OK*} $result)
so that the dedicated rdbchecksum no test block remains the single place
checking the checksum-disabled message; update any surrounding comments if
necessary to reflect that this test does not disable checksum.
In `@valkey.conf`:
- Around line 834-847: Remove the no-op user-facing config entry
"replcompression" from valkey.conf (or at minimum revert it to an
internal/commented example) so we don't ship a persistent option that the docs
state has no effect; locate the "replcompression" knob in the provided
diff/block and either delete the "replcompression no" line or convert it to an
internal-only commented note explaining it's reserved for a future release,
ensuring no new user-visible config option is introduced until the transport
compression behavior and trade-offs are implemented.
---
Outside diff comments:
In `@src/aof.c`:
- Around line 1009-1054: This change touches core AOF restart and
persistence-path logic (functions restartAOFWithSyncRdb and
rdbFileUsesStreamingCompression, referencing server.rdb_filename) and therefore
needs an explicit core-team architectural review before merging; please add a
clear PR-level request for core-team sign-off (e.g., tag core-team reviewers and
add an "architectural review required" note in the PR description), include an
explanation of the behavioral change and rationale, and ensure the commit
message references the review request so CI/maintainers do not merge without the
required approval.
---
Nitpick comments:
In `@src/aof.c`:
- Around line 1009-1026: Add a short rationale comment above
rdbFileUsesStreamingCompression explaining why inspecting only the VKCS
envelope/header is sufficient to decide the streaming-compression fallback
during AOF restart: note that the envelope contains the compression/streaming
metadata used by streamReadEnvelopeInfo, that this probe is cheap and reliable
without scanning the whole RDB, and that the outcome drives the non-obvious AOF
restart fallback path; reference streamReadEnvelopeInfo and STREAM_KIND_RDB so
future readers understand the trust boundary and intent.
In `@src/rdb.c`:
- Around line 1565-3792: The change set modifies critical persistence and
replication code (rdbSaveInternal, rdbLoadRio, rdbLoadRioWithLoadingCtxScopedRdb
and related framing/checksum/streaming functions) and therefore must be routed
for an architecture review by the core-team before merging; update the PR by
adding an explicit request-for-review line tagging `@core-team` in the PR
description, add the "requires-arch-review" label (or equivalent), and include a
short summary of the affected symbols (rdbSaveInternal,
rdbSaveToFile/rdbSaveBackground, rdbLoadRio/rdbLoadRioWithLoadingCtx,
rdbInputStreamPrepare/rdbInputStreamValidateEnd) plus the rationale that
framing/checksum/streaming semantics changed so reviewers can focus their
review.
In `@src/replication.c`:
- Around line 1449-1450: This change touches replication capability negotiation
in replication.c (the block using strcasecmp(objectGetVal(c->argv[j + 1]),
REPLICA_CAPA_COMPRESSION_STR) and the REPLICA_CAPA_COMPRESSION flag) and must
get an architectural review from the `@core-team`; update the PR by explicitly
requesting `@core-team` review and add a short in-code comment/TODO next to the
capability negotiation (and the related range you modified around lines
3859-3886) noting "requires `@core-team` architectural review" so reviewers see it
in-context and the PR cannot be merged without that review.
In `@tests/integration/repl-compression.tcl`:
- Around line 122-136: Rename the test case title string in the test block that
currently reads "Backward compatibility - older replica without capa compression
connects successfully" to a descriptive name reflecting what it actually
verifies, e.g. "Replica without advertised compression connects successfully";
update the test header (the test { ... } line) that wraps the start_server /
$replica replicaof / wait_for_condition sequence so the new name appears in test
output and logs (no code changes inside start_server, $replica replicaof,
wait_for_condition, s 0 master_link_status, or $replica replicaof no one are
required).
In `@tests/integration/replication-aof-sync.tcl`:
- Around line 192-194: The fixed sleep `after 1000` is making the test flaky;
replace it with a deterministic synchronization by calling `waitForBgrewriteaof
$replica` before performing the negative log assertion, then perform `assert
{![log_file_matches $replica_log "*Reused RDB file from primary sync as AOF base
file*"]}`; locate the block that currently has `after 1000` and move or replace
that line with the existing `waitForBgrewriteaof` call so the log check runs
only after the AOF rewrite completion.
🪄 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: defaults
Review profile: CHILL
Plan: Pro Plus
Run ID: 491e41fe-c726-4767-a876-48e14d6b2df6
📒 Files selected for processing (43)
cmake/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/rdb.csrc/rdb.hsrc/replication.csrc/rio.csrc/rio.hsrc/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
Comment on lines
+1619
to
+1623
| const BYTE* const srcStart = (const BYTE*)srcBuffer; | ||
| const BYTE* const srcEnd = srcStart + *srcSizePtr; | ||
| const BYTE* srcPtr = srcStart; | ||
| BYTE* const dstStart = (BYTE*)dstBuffer; | ||
| BYTE* const dstEnd = dstStart ? dstStart + *dstSizePtr : NULL; |
Contributor
There was a problem hiding this comment.
🧩 Analysis chain
🏁 Script executed:
#!/bin/bash
set -euo pipefail
# Verify UB site
rg -n -C2 'srcEnd = srcStart \+ \*srcSizePtr' deps/lz4/lz4frame.c
# Verify reachable NULL call site
rg -n -C3 'LZ4F_decompress\(dctx, NULL, &o, NULL, &i, NULL\)' deps/lz4/lz4frame.cRepository: valkey-io/valkey
Length of output: 714
🏁 Script executed:
#!/bin/bash
set -euo pipefail
# 1. Find the function signature of LZ4F_decompress
echo "=== LZ4F_decompress Function Signature ==="
rg -A 10 'LZ4F_decompress\s*\(' deps/lz4/lz4frame.c | head -30
# 2. Get broader context around line 1620 (inside function)
echo -e "\n=== Context around line 1620 (pointer arithmetic) ==="
sed -n '1600,1650p' deps/lz4/lz4frame.c | cat -n
# 3. Get context around line 1496 (call site)
echo -e "\n=== Context around line 1496 (call site) ==="
sed -n '1480,1510p' deps/lz4/lz4frame.c | cat -n
# 4. Check for NULL checks before line 1620
echo -e "\n=== NULL/sanity checks in the function ==="
rg -B 30 'srcEnd = srcStart \+ \*srcSizePtr' deps/lz4/lz4frame.c | grep -i 'null\|check\|return' | head -20Repository: valkey-io/valkey
Length of output: 6715
🏁 Script executed:
#!/bin/bash
set -euo pipefail
# Check for early returns or guards before line 1620
echo "=== Full LZ4F_decompress function head (before pointer arithmetic) ==="
sed -n '1614,1630p' deps/lz4/lz4frame.c | cat -n
# Check what happens after line 1620 - are srcStart/srcEnd used before any validation?
echo -e "\n=== Usage of srcStart/srcEnd after line 1620 ==="
sed -n '1620,1700p' deps/lz4/lz4frame.c | cat -n | head -50
# Check: does the call at line 1496 always have i=0?
echo -e "\n=== Context: what are the values of i and o when calling? ==="
sed -n '1490,1500p' deps/lz4/lz4frame.c | cat -n
# Look for any existing checks for NULL srcBuffer elsewhere in the function
echo -e "\n=== Search for srcBuffer NULL checks in LZ4F_decompress ==="
sed -n '1614,1800p' deps/lz4/lz4frame.c | grep -n -i 'srcbuffer\|srcstart' | head -30Repository: valkey-io/valkey
Length of output: 4567
Fix NULL pointer arithmetic in LZ4F_decompress.
Line 1620 computes srcStart + *srcSizePtr even when srcBuffer is NULL, which is undefined behavior in C. This is reachable from line 1496 (LZ4F_decompress(dctx, NULL, &o, NULL, &i, NULL)) and violates pointer safety.
Notably, dstBuffer is protected with a ternary conditional (line 1624), but srcBuffer is not. The fix ensures consistent NULL-safety handling:
Proposed fix
const BYTE* const srcStart = (const BYTE*)srcBuffer;
+ if (srcStart == NULL && *srcSizePtr != 0) {
+ return (size_t)-1; /* or use RETURN_ERROR macro */
+ }
const BYTE* const srcEnd = srcStart + (*srcSizePtr ? *srcSizePtr : 1);Alternatively, treat NULL as a valid "no input" case with a safe sentinel:
const BYTE* const srcStart = (const BYTE*)srcBuffer;
+ BYTE dummyByte = 0;
+ const BYTE* const effectiveStart = srcStart ? srcStart : &dummyByte;
- const BYTE* const srcEnd = srcStart + *srcSizePtr;
+ const BYTE* const srcEnd = effectiveStart + *srcSizePtr;🧰 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 - 1623, The computation of srcEnd uses
pointer arithmetic on srcStart even when srcBuffer is NULL in LZ4F_decompress,
causing undefined behavior; update the initialization in LZ4F_decompress so
srcStart and srcEnd are computed safely by treating a NULL srcBuffer as a valid
"no input" sentinel (e.g. set srcStart to NULL and srcEnd to srcStart when
srcBuffer is NULL, or derive srcEnd using a conditional like dstEnd does),
ensuring you adjust variables srcStart, srcEnd and srcPtr consistently so
subsequent logic (which reads from srcPtr up to srcEnd) safely handles the
NULL/no-input case without performing NULL + offset arithmetic.
Comment on lines
+69
to
+74
| size_t compressionLz4OutputBound(size_t input_len) { | ||
| /* Conservative worst-case: data bound + frame header + flush/end overhead. | ||
| * Always includes all components so the caller can allocate once and reuse | ||
| * for any flush mode and frame state. */ | ||
| return LZ4F_compressBound(input_len, &lz4f_prefs) + LZ4F_HEADER_SIZE_MAX + LZ4F_compressBound(0, &lz4f_prefs); | ||
| } |
Contributor
There was a problem hiding this comment.
Guard size_t overflow in output-bound arithmetic.
The bound calculation adds multiple size_t values without overflow checks; very large inputs can wrap and under-size caller buffers.
Proposed hardening
size_t compressionLz4OutputBound(size_t input_len) {
@@
- return LZ4F_compressBound(input_len, &lz4f_prefs) + LZ4F_HEADER_SIZE_MAX + LZ4F_compressBound(0, &lz4f_prefs);
+ size_t a = LZ4F_compressBound(input_len, &lz4f_prefs);
+ size_t b = LZ4F_HEADER_SIZE_MAX;
+ size_t c = LZ4F_compressBound(0, &lz4f_prefs);
+ if (a > SIZE_MAX - b) return SIZE_MAX;
+ a += b;
+ if (a > SIZE_MAX - c) return SIZE_MAX;
+ return a + c;
}📝 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.
Suggested change
| size_t compressionLz4OutputBound(size_t input_len) { | |
| /* Conservative worst-case: data bound + frame header + flush/end overhead. | |
| * Always includes all components so the caller can allocate once and reuse | |
| * for any flush mode and frame state. */ | |
| return LZ4F_compressBound(input_len, &lz4f_prefs) + LZ4F_HEADER_SIZE_MAX + LZ4F_compressBound(0, &lz4f_prefs); | |
| } | |
| size_t compressionLz4OutputBound(size_t input_len) { | |
| /* Conservative worst-case: data bound + frame header + flush/end overhead. | |
| * Always includes all components so the caller can allocate once and reuse | |
| * for any flush mode and frame state. */ | |
| size_t a = LZ4F_compressBound(input_len, &lz4f_prefs); | |
| size_t b = LZ4F_HEADER_SIZE_MAX; | |
| size_t c = LZ4F_compressBound(0, &lz4f_prefs); | |
| if (a > SIZE_MAX - b) return SIZE_MAX; | |
| a += b; | |
| if (a > SIZE_MAX - c) return SIZE_MAX; | |
| return a + 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/compression_lz4.c` around lines 69 - 74, The addition in
compressionLz4OutputBound can overflow size_t; update compressionLz4OutputBound
to perform checked arithmetic when summing LZ4F_compressBound(input_len,
&lz4f_prefs), LZ4F_HEADER_SIZE_MAX and LZ4F_compressBound(0, &lz4f_prefs) —
detect overflow at each add (e.g., check if a + b < a) and if overflow would
occur return SIZE_MAX (or another agreed sentinel) so callers know allocation is
impossible, otherwise return the safe sum; reference the function name
compressionLz4OutputBound and the symbols lz4f_prefs, LZ4F_HEADER_SIZE_MAX, and
LZ4F_compressBound when locating the change.
Comment on lines
+85
to
+86
| if (!sc || !sc->ctx) return -1; | ||
|
|
Contributor
There was a problem hiding this comment.
🧩 Analysis chain
🏁 Script executed:
#!/bin/bash
set -euo pipefail
echo "1) Verify missing sticky-error guards in changed file:"
rg -n --type=c 'compressionLz4(CompressFeed|DecompressFeed)|errored' src/compression_lz4.c -C2
echo
echo "2) Verify LZ4 frame API notes about error-state behavior:"
LZ4HDR="$(fd -i 'lz4frame.h' deps src | head -n1 || true)"
if [ -z "${LZ4HDR}" ]; then
echo "lz4frame.h not found in repository paths."
exit 0
fi
echo "Header: ${LZ4HDR}"
rg -n 'undefined|error|LZ4F_.*Error' "${LZ4HDR}" -C2Repository: valkey-io/valkey
Length of output: 9689
Prevent re-entry into LZ4 contexts after errors to avoid undefined state.
LZ4 Frame API documentation states: "After an error, the state is left in a UB state, and must be re-initialized or freed" and "After a decompression error, the dctx context is not resumable." Currently, compressionLz4CompressFeed and compressionLz4DecompressFeed set errored = true on failures but do not check this flag at function entry. Subsequent calls with an errored context violate the LZ4 API contract and invoke undefined behavior.
Add sticky-error guards at function entry:
Proposed fix
ssize_t compressionLz4CompressFeed(streamCompressor *sc,
uint8_t *output,
size_t output_capacity,
const uint8_t *input,
size_t input_len,
compressFlushMode flush_mode) {
@@
- if (!sc || !sc->ctx) return -1;
+ if (!sc || !sc->ctx || sc->errored) return -1;
@@
ssize_t compressionLz4DecompressFeed(streamDecompressor *sd,
uint8_t *output,
size_t output_capacity,
const uint8_t *input,
size_t input_len,
size_t *input_consumed) {
@@
- if (!sd || !sd->ctx || !input_consumed) return -1;
+ if (!sd || !sd->ctx || sd->errored || !input_consumed) return -1;Also applies to: 149–155, 169–170, 175–177
🤖 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_lz4.c` around lines 85 - 86, The functions
compressionLz4CompressFeed and compressionLz4DecompressFeed (and the other LZ4
feed/finish entry points around the same area) must guard against reuse of a
context that has previously errored: at each function entry check sc and sc->ctx
as well as sc->errored and immediately return an error (-1) if errored is true
to avoid calling LZ4 APIs on a UB state; keep the existing behavior of setting
sc->errored = true on failures, but add the sticky-error guard at the top of
each relevant function (e.g., compressionLz4CompressFeed,
compressionLz4DecompressFeed and the related feed/finish helpers) so once
errored no further LZ4 calls are made until the context is reinitialized or
freed.
Comment on lines
+250
to
+304
| test {Partial VKCS snapshot copied from an interrupted BGSAVE is rejected on load} { | ||
| r config set rdbcompression yes | ||
| r config set rdb-compression-algo lz4 | ||
| r config set rdb-key-save-delay 10000 | ||
| r flushall | ||
| set noisy_payload "" | ||
| for {set j 0} {$j < 32768} {incr j} { | ||
| append noisy_payload [format %c [expr {(($j * 31) + 17) % 94 + 33}]] | ||
| } | ||
| for {set i 0} {$i < 128} {incr i} { | ||
| r set "partial:$i" "${noisy_payload}:$i" | ||
| } | ||
|
|
||
| assert_match {*Background saving started*} [r bgsave] | ||
| wait_for_condition 200 10 { | ||
| [s rdb_bgsave_in_progress] eq 1 | ||
| } else { | ||
| r config set rdb-key-save-delay 0 | ||
| fail "BGSAVE did not start in time" | ||
| } | ||
|
|
||
| wait_for_condition 200 10 { | ||
| [get_child_pid 0] ne "" | ||
| } else { | ||
| r config set rdb-key-save-delay 0 | ||
| fail "Timed out waiting for BGSAVE child pid" | ||
| } | ||
| set child_pid [get_child_pid 0] | ||
| set dir [lindex [r config get dir] 1] | ||
| set temp_rdb [file join $dir temp-${child_pid}.rdb] | ||
| set partial_rdb [file join $dir partial-vkcs.rdb] | ||
| wait_for_condition 500 10 { | ||
| [file exists $temp_rdb] && [file size $temp_rdb] > 4096 | ||
| } else { | ||
| r config set rdb-key-save-delay 0 | ||
| catch {exec kill -9 $child_pid} | ||
| fail "Timed out waiting for partial VKCS snapshot" | ||
| } | ||
|
|
||
| file copy -force $temp_rdb $partial_rdb | ||
| assert_equal "VKCS" [string range [read_binary_file_prefix $partial_rdb 8] 0 3] | ||
|
|
||
| catch {exec kill -9 $child_pid} | ||
| wait_for_condition 500 10 { | ||
| [s rdb_bgsave_in_progress] eq 0 | ||
| } else { | ||
| r config set rdb-key-save-delay 0 | ||
| fail "Interrupted BGSAVE child was not collected in time" | ||
| } | ||
| r config set rdb-key-save-delay 0 | ||
|
|
||
| file copy -force $partial_rdb [dump_rdb_path r] | ||
| catch {r debug reload nosave} err | ||
| assert_match "*Error*" $err | ||
| } |
Contributor
There was a problem hiding this comment.
Clean up the temporary partial snapshot file after the test.
Line 280 creates partial-vkcs.rdb, but it is never deleted. Please remove it at test end to avoid filesystem residue across runs.
Suggested change
file copy -force $partial_rdb [dump_rdb_path r]
catch {r debug reload nosave} err
assert_match "*Error*" $err
+ catch {file delete -force $partial_rdb}As per coding guidelines, Ensure proper cleanup of resources and temporary files in tests.
📝 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.
Suggested change
| test {Partial VKCS snapshot copied from an interrupted BGSAVE is rejected on load} { | |
| r config set rdbcompression yes | |
| r config set rdb-compression-algo lz4 | |
| r config set rdb-key-save-delay 10000 | |
| r flushall | |
| set noisy_payload "" | |
| for {set j 0} {$j < 32768} {incr j} { | |
| append noisy_payload [format %c [expr {(($j * 31) + 17) % 94 + 33}]] | |
| } | |
| for {set i 0} {$i < 128} {incr i} { | |
| r set "partial:$i" "${noisy_payload}:$i" | |
| } | |
| assert_match {*Background saving started*} [r bgsave] | |
| wait_for_condition 200 10 { | |
| [s rdb_bgsave_in_progress] eq 1 | |
| } else { | |
| r config set rdb-key-save-delay 0 | |
| fail "BGSAVE did not start in time" | |
| } | |
| wait_for_condition 200 10 { | |
| [get_child_pid 0] ne "" | |
| } else { | |
| r config set rdb-key-save-delay 0 | |
| fail "Timed out waiting for BGSAVE child pid" | |
| } | |
| set child_pid [get_child_pid 0] | |
| set dir [lindex [r config get dir] 1] | |
| set temp_rdb [file join $dir temp-${child_pid}.rdb] | |
| set partial_rdb [file join $dir partial-vkcs.rdb] | |
| wait_for_condition 500 10 { | |
| [file exists $temp_rdb] && [file size $temp_rdb] > 4096 | |
| } else { | |
| r config set rdb-key-save-delay 0 | |
| catch {exec kill -9 $child_pid} | |
| fail "Timed out waiting for partial VKCS snapshot" | |
| } | |
| file copy -force $temp_rdb $partial_rdb | |
| assert_equal "VKCS" [string range [read_binary_file_prefix $partial_rdb 8] 0 3] | |
| catch {exec kill -9 $child_pid} | |
| wait_for_condition 500 10 { | |
| [s rdb_bgsave_in_progress] eq 0 | |
| } else { | |
| r config set rdb-key-save-delay 0 | |
| fail "Interrupted BGSAVE child was not collected in time" | |
| } | |
| r config set rdb-key-save-delay 0 | |
| file copy -force $partial_rdb [dump_rdb_path r] | |
| catch {r debug reload nosave} err | |
| assert_match "*Error*" $err | |
| } | |
| test {Partial VKCS snapshot copied from an interrupted BGSAVE is rejected on load} { | |
| r config set rdbcompression yes | |
| r config set rdb-compression-algo lz4 | |
| r config set rdb-key-save-delay 10000 | |
| r flushall | |
| set noisy_payload "" | |
| for {set j 0} {$j < 32768} {incr j} { | |
| append noisy_payload [format %c [expr {(($j * 31) + 17) % 94 + 33}]] | |
| } | |
| for {set i 0} {$i < 128} {incr i} { | |
| r set "partial:$i" "${noisy_payload}:$i" | |
| } | |
| assert_match {*Background saving started*} [r bgsave] | |
| wait_for_condition 200 10 { | |
| [s rdb_bgsave_in_progress] eq 1 | |
| } else { | |
| r config set rdb-key-save-delay 0 | |
| fail "BGSAVE did not start in time" | |
| } | |
| wait_for_condition 200 10 { | |
| [get_child_pid 0] ne "" | |
| } else { | |
| r config set rdb-key-save-delay 0 | |
| fail "Timed out waiting for BGSAVE child pid" | |
| } | |
| set child_pid [get_child_pid 0] | |
| set dir [lindex [r config get dir] 1] | |
| set temp_rdb [file join $dir temp-${child_pid}.rdb] | |
| set partial_rdb [file join $dir partial-vkcs.rdb] | |
| wait_for_condition 500 10 { | |
| [file exists $temp_rdb] && [file size $temp_rdb] > 4096 | |
| } else { | |
| r config set rdb-key-save-delay 0 | |
| catch {exec kill -9 $child_pid} | |
| fail "Timed out waiting for partial VKCS snapshot" | |
| } | |
| file copy -force $temp_rdb $partial_rdb | |
| assert_equal "VKCS" [string range [read_binary_file_prefix $partial_rdb 8] 0 3] | |
| catch {exec kill -9 $child_pid} | |
| wait_for_condition 500 10 { | |
| [s rdb_bgsave_in_progress] eq 0 | |
| } else { | |
| r config set rdb-key-save-delay 0 | |
| fail "Interrupted BGSAVE child was not collected in time" | |
| } | |
| r config set rdb-key-save-delay 0 | |
| file copy -force $partial_rdb [dump_rdb_path r] | |
| catch {r debug reload nosave} err | |
| assert_match "*Error*" $err | |
| catch {file delete -force $partial_rdb} | |
| } |
🤖 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 250 - 304, The test
leaves the temporary partial snapshot file partial-vkcs.rdb (variable
partial_rdb) on disk; delete it at the end of the test to avoid residue by
adding a safe cleanup call such as catch {file delete -force $partial_rdb} after
the final assertion (after assert_match "*Error*" $err) so the file is always
removed even if delete fails; you can also wrap earlier early-failure paths to
delete partial_rdb where appropriate to ensure cleanup in all exit paths.
Comment on lines
+42
to
+95
| test {Replica with replcompression no does NOT send capa compression} { | ||
| start_server {overrides {save "" replcompression no}} { | ||
| set replica [srv 0 client] | ||
| $replica replicaof $primary_host $primary_port | ||
|
|
||
| wait_for_condition 50 100 { | ||
| [s 0 master_link_status] eq {up} | ||
| } else { | ||
| fail "Replication not started" | ||
| } | ||
|
|
||
| # Full sync completes normally without compression capability | ||
| assert_equal {up} [s 0 master_link_status] | ||
|
|
||
| $replica replicaof no one | ||
| } | ||
| } | ||
|
|
||
| test {Replica with replcompression yes and diskless load sends capa compression} { | ||
| start_server {overrides {save "" replcompression yes repl-diskless-load swapdb}} { | ||
| set replica [srv 0 client] | ||
| $replica replicaof $primary_host $primary_port | ||
|
|
||
| wait_for_condition 50 100 { | ||
| [s 0 master_link_status] eq {up} | ||
| } else { | ||
| fail "Replication not started" | ||
| } | ||
|
|
||
| set info [$primary info replication] | ||
| assert_match "*slave0:*" $info | ||
| assert_equal {up} [s 0 master_link_status] | ||
|
|
||
| $replica replicaof no one | ||
| } | ||
| } | ||
|
|
||
| test {Replica with replcompression yes but disk-backed load does NOT send capa compression} { | ||
| start_server {overrides {save "" replcompression yes repl-diskless-load disabled}} { | ||
| set replica [srv 0 client] | ||
| $replica replicaof $primary_host $primary_port | ||
|
|
||
| wait_for_condition 50 100 { | ||
| [s 0 master_link_status] eq {up} | ||
| } else { | ||
| fail "Replication not started" | ||
| } | ||
|
|
||
| # Full sync completes normally — disk-backed replica does not advertise compression | ||
| assert_equal {up} [s 0 master_link_status] | ||
|
|
||
| $replica replicaof no one | ||
| } | ||
| } |
Contributor
There was a problem hiding this comment.
Capability-negotiation tests currently validate link-up, not capability state.
These cases are named around sending/not-sending capa compression, but they only assert replication reaches up. That can pass even if capability negotiation is broken. Add explicit assertions for the negotiated capability bit/string on the primary-side replica metadata (or another deterministic observable).
As per coding guidelines: "Integration tests live under tests/ and are written in Tcl. End-to-end behavior changes should be covered by integration tests." and "Use clear assertions with meaningful error messages in tests".
Also applies to: 139-166
🤖 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/repl-compression.tcl` around lines 42 - 95, After the
replica reaches up, assert the negotiated capability on the primary-side replica
metadata instead of only checking master_link_status: call set info [$primary
info replication] (as already used in the second test) and then use
assert_match/assert_no_match to verify the presence or absence of the
compression capability token for slave0; e.g. in the "replcompression yes and
diskless load" test add assert_match "*slave0:*capa=*compression*" (or the exact
capability substring your server reports) and in the "replcompression no" and
"replcompression yes but disk-backed load" tests add assert_no_match
"*slave0:*capa=*compression*" to make the tests deterministic, using the
existing assert_match/assert_equal helpers to provide clear failure messages.
Comment on lines
+834
to
+847
| # Enable compression for the replication transport between primary and replica. | ||
| # When enabled on a replica, the replica advertises the "compression" capability | ||
| # during the PSYNC handshake. The primary records this capability and, in a | ||
| # future release, will compress the incremental replication stream sent to that | ||
| # replica. | ||
| # | ||
| # Requires repl-diskless-load to be enabled (not "disabled") on the replica for | ||
| # the capability to be advertised. | ||
| # | ||
| # No effect yet in this release — compressed replication transport is introduced | ||
| # in a later release. The compression algorithm (LZ4) and level are fixed | ||
| # internally and will become configurable in a later release. | ||
| replcompression no | ||
|
|
Contributor
There was a problem hiding this comment.
Avoid shipping a no-op user config knob in valkey.conf.
This introduces a persistent user-facing option whose own docs say it has no runtime effect in this release. That increases config/downgrade surface without user value yet. Prefer keeping this internal/implicit until transport behavior is active, then expose it with a concrete trade-off.
As per coding guidelines: "valkey.conf: Avoid new configuration options unless heuristics are insufficient; only add for explicit trade-offs between CPU and memory".
🤖 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 `@valkey.conf` around lines 834 - 847, Remove the no-op user-facing config
entry "replcompression" from valkey.conf (or at minimum revert it to an
internal/commented example) so we don't ship a persistent option that the docs
state has no effect; locate the "replcompression" knob in the provided
diff/block and either delete the "replcompression no" line or convert it to an
internal-only commented note explaining it's reserved for a future release,
ensuring no new user-visible config option is introduced until the transport
compression behavior and trade-offs are implemented.
Signed-off-by: Sarthak Aggarwal <sarthagg@amazon.com>
Signed-off-by: Sarthak Aggarwal <sarthagg@amazon.com>
Signed-off-by: Sarthak Aggarwal <sarthagg@amazon.com>
- rdbSaveInternal: the old comment claimed per-string LZF was disabled 'when algo != LZF', but the actual gate is RIO_FLAG_STREAMING_COMPRESSION on the wrapper rio. Standalone rios (DUMP, AOF rewrite, diskless) keep using LZF regardless of algo. - rdbInputStreamPrepare: flag the synchronous probe IO so a future non-blocking caller (replication) doesn't accidentally block the loop. - rdbRioHasCorruptCompressedInput: the cast is sound today only because one producer sets RIO_FLAG_STREAMING_DECOMPRESSION. Replace the 'SAFETY' assertion with a note telling the next person to add a type discriminator before adding a second producer. - rdbSaveRawString: minor wording cleanup on the per-string LZF gate. Signed-off-by: Sarthak Aggarwal <sarthagg@amazon.com>
Match the surrounding Valkey style: drop comments that restate the code, drop the Ownership/Threading/Returns boilerplate from headers, collapse repeated 'capacity-shortage is retriable' notes into one explanation per function. Behavior is unchanged. Net -285 lines; integration tests (21/21) and build are clean. Signed-off-by: Sarthak Aggarwal <sarthagg@amazon.com>
Signed-off-by: Sarthak Aggarwal <sarthagg@amazon.com>
Signed-off-by: Sarthak Aggarwal <sarthagg@amazon.com>
Signed-off-by: Sarthak Aggarwal <sarthagg@amazon.com>
… config
Introduces REPLICA_CAPA_COMPRESSION handshake negotiation and a
replcompression boolean config. A replica with replcompression=yes
advertises capa compression during the PSYNC handshake when using
diskless load. The primary records REPLICA_CAPA_COMPRESSION on the
replica's capa bitmask but takes no action on it yet — compressed
replication transport lands in a follow-up.
Compression parameters (algorithm, level) are fixed internally via
REPL_COMPRESSION_ALGO (ALGO_LZ4) and REPL_COMPRESSION_LEVEL (5, HC
mode) and will become configurable in a later release.
- src/server.h: new REPLICA_CAPA_COMPRESSION (1<<4) flag, matching
_STR, repl_compression field, and REPL_COMPRESSION_{ALGO,LEVEL}
constants
- src/config.c: new replcompression bool config
- src/replication.c: primary records capa in replconfCommand; replica
advertises capa compression when server.repl_compression and
useDisklessLoad(); argv/lens arrays grown from 9 to 11 entries;
useDisklessLoad() extracted into a local reused by skip-rdb-checksum
and compression checks
- valkey.conf: new replcompression config documentation block
- src/unit/test_repl_compression.cpp: 3 GTest cases covering capa bit
uniqueness, capa string, and ALGO_LZ4 non-zero
- tests/integration/repl-compression.tcl: 9 TCL integration tests
covering config CRUD, CONFIG REWRITE persistence, and handshake
behavior
Signed-off-by: Roshan Khatri <rvkhatri@amazon.com>
roshkhatri
force-pushed
the
repl-compression-handshake-v2
branch
from
May 22, 2026 19:38
c5ecfeb to
92db9be
Compare
Contributor
There was a problem hiding this comment.
Actionable comments posted: 2
♻️ Duplicate comments (2)
valkey.conf (1)
837-850:⚠️ Potential issue | 🟠 Major | 🏗️ Heavy liftDefer exposing
replcompressionas a user config until it has real runtime trade-offs.Line 846 explicitly says this has no effect in the current release, but the option is still persisted in
valkey.conf, which expands config and downgrade surface without a present CPU/memory trade-off.As per coding guidelines: "valkey.conf: Avoid new configuration options unless heuristics are insufficient; only add for explicit trade-offs between CPU and memory".
🤖 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 `@valkey.conf` around lines 837 - 850, The config option replcompression is documented as having no effect in this release and should not be exposed in valkey.conf yet; remove (or comment out) the replcompression entry and its user-facing documentation from valkey.conf and keep any internal placeholder logic hidden until the feature is implemented, adding a short TODO/NOTE in the codebase/docs referencing replcompression so it can be re-introduced when runtime CPU/memory trade-offs exist; also update any config schema or docs that list replcompression to avoid exposing it to users prematurely.src/replication.c (1)
3880-3887:⚠️ Potential issue | 🟠 Major | ⚡ Quick winCompression capability is still missing on the dual-channel handshake path.
Line 3880 adds
capa compressionfor single-channel, butdualChannelReplHandleHandshake()(Line 3059) still sends a fixed REPLCONF without it, so capability negotiation can diverge by path.Suggested fix
- *err = sendCommand(conn, "REPLCONF", "capa", "eof", "rdb-only", "1", "rdb-channel", "1", "listening-port", portstr, - NULL); + if (server.repl_compression && useDisklessLoad()) { + *err = sendCommand(conn, "REPLCONF", + "capa", "eof", + "capa", REPLICA_CAPA_COMPRESSION_STR, + "rdb-only", "1", + "rdb-channel", "1", + "listening-port", portstr, + NULL); + } else { + *err = sendCommand(conn, "REPLCONF", + "capa", "eof", + "rdb-only", "1", + "rdb-channel", "1", + "listening-port", portstr, + NULL); + }🤖 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 3880 - 3887, dual-channel handshake in dualChannelReplHandleHandshake() doesn't include the "capa compression" token causing capability mismatch; update the REPLCONF/handshake construction in dualChannelReplHandleHandshake() to append the same tokens used in the single-channel path when server.repl_compression && use_diskless_load are true — specifically add REPLICA_CAPA_COMPRESSION_STR (and its length handling) to the argv/lens/argc sequence or include it in the formatted REPLCONF string in the same conditional used in the single-channel code so both handshake paths advertise compression consistently.
🧹 Nitpick comments (2)
src/replication.c (1)
3859-3862: ⚡ Quick winUse C-style block comments in
src/*.c.Line 3858 uses
// ...in C code undersrc/; switch it to/* ... */to match project conventions.As per coding guidelines: "
src/**/*.{c,h}: Use C-style /* */ comments for single or multi-line comments 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/replication.c` around lines 3859 - 3862, Replace the C++-style // comment on the nearby line in src/replication.c with a C-style block comment (/* ... */); locate the area around the use_diskless_load and replicationSupportSkipRDBChecksum calls (variables use_diskless_load, send_skip_rdb_checksum_capa, argv, lens) and convert any // comment(s) there to /* ... */ to follow the project's src/*.c comment convention.src/rdb.c (1)
3151-3200: ⚡ Quick winMake the new input-stream helpers file-local.
rdbInputStreamInit(),rdbInputStreamPrepare(),rdbInputStreamDestroy(),rdbInputStreamValidateEnd(), andrdbRioHasCorruptCompressedInput()all look local tosrc/rdb.c; leaving them non-staticwidens the symbol surface for no benefit.As per coding guidelines, "Use static keyword for file-local 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 - 3200, These helper functions are file-local and should be marked static to limit symbol visibility; change the function declarations for rdbInputStreamInit, rdbInputStreamPrepare, rdbInputStreamDestroy, rdbInputStreamValidateEnd, and rdbRioHasCorruptCompressedInput to be static (i.e., add the static keyword to each definition) so they are internal to src/rdb.c and follow the project's coding guideline for file-local functions.
🤖 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/aof.c`:
- Around line 1009-1026: rdbFileUsesStreamingCompression currently does a single
read() into header and treats any short read as “uncompressed”, which
misclassifies compressed RDBs when read() is interrupted or returns partial
data; change the single read to a loop that repeatedly calls read() (handling
EINTR by continuing), accumulates bytes into header until sizeof(header) is
filled or read() returns 0/-1, preserve the original errno in read_errno on
failure, close fd before returning, and only consider the file “short” (return
0) if the total bytes read is less than sizeof(header) (i.e. EOF) — otherwise
call streamReadEnvelopeInfo(header, sizeof(header), STREAM_KIND_RDB, &info) as
before; refer to function rdbFileUsesStreamingCompression, variables header,
nread/read_errno, fd, and streamReadEnvelopeInfo.
In `@src/rdb.c`:
- Around line 521-530: rdbSavedObjectLen currently models per-string LZF even
when the real save path will use whole-stream (rio) compression and thus skip
per-string LZF; update rdbSavedObjectLen so its length calculation matches
rdbSaveObject by not counting LZF when the save will use streaming compression:
detect the same condition used in rdbSaveObject (RIO_FLAG_STREAMING_COMPRESSION)
and when rdb==NULL assume whole-stream compression is active (or better, add a
parameter/flag to rdbSavedObjectLen to indicate streaming-compression mode) so
rdbSavedObjectLen and rdbSaveObject (the functions referenced) return the same
encoded size. Ensure you update callers if you add a parameter.
---
Duplicate comments:
In `@src/replication.c`:
- Around line 3880-3887: dual-channel handshake in
dualChannelReplHandleHandshake() doesn't include the "capa compression" token
causing capability mismatch; update the REPLCONF/handshake construction in
dualChannelReplHandleHandshake() to append the same tokens used in the
single-channel path when server.repl_compression && use_diskless_load are true —
specifically add REPLICA_CAPA_COMPRESSION_STR (and its length handling) to the
argv/lens/argc sequence or include it in the formatted REPLCONF string in the
same conditional used in the single-channel code so both handshake paths
advertise compression consistently.
In `@valkey.conf`:
- Around line 837-850: The config option replcompression is documented as having
no effect in this release and should not be exposed in valkey.conf yet; remove
(or comment out) the replcompression entry and its user-facing documentation
from valkey.conf and keep any internal placeholder logic hidden until the
feature is implemented, adding a short TODO/NOTE in the codebase/docs
referencing replcompression so it can be re-introduced when runtime CPU/memory
trade-offs exist; also update any config schema or docs that list
replcompression to avoid exposing it to users prematurely.
---
Nitpick comments:
In `@src/rdb.c`:
- Around line 3151-3200: These helper functions are file-local and should be
marked static to limit symbol visibility; change the function declarations for
rdbInputStreamInit, rdbInputStreamPrepare, rdbInputStreamDestroy,
rdbInputStreamValidateEnd, and rdbRioHasCorruptCompressedInput to be static
(i.e., add the static keyword to each definition) so they are internal to
src/rdb.c and follow the project's coding guideline for file-local functions.
In `@src/replication.c`:
- Around line 3859-3862: Replace the C++-style // comment on the nearby line in
src/replication.c with a C-style block comment (/* ... */); locate the area
around the use_diskless_load and replicationSupportSkipRDBChecksum calls
(variables use_diskless_load, send_skip_rdb_checksum_capa, argv, lens) and
convert any // comment(s) there to /* ... */ to follow the project's src/*.c
comment convention.
🪄 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: e7b855c3-7317-4398-9b06-a9df2f0260e0
📒 Files selected for processing (43)
cmake/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/rdb.csrc/rdb.hsrc/replication.csrc/rio.csrc/rio.hsrc/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
✅ Files skipped from review due to trivial changes (1)
- deps/lz4/LICENSE
Comment on lines
+1009
to
+1026
| static int rdbFileUsesStreamingCompression(const char *filename) { | ||
| unsigned char header[VKCS_ENVELOPE_SIZE]; | ||
| int fd = open(filename, O_RDONLY); | ||
| if (fd == -1) return -1; | ||
|
|
||
| ssize_t nread = read(fd, header, sizeof(header)); | ||
| int read_errno = errno; | ||
| close(fd); | ||
|
|
||
| if (nread == -1) { | ||
| errno = read_errno; | ||
| return -1; | ||
| } | ||
| if (nread < (ssize_t)sizeof(header)) return 0; | ||
|
|
||
| streamReaderInfo info = {0}; | ||
| return streamReadEnvelopeInfo(header, sizeof(header), STREAM_KIND_RDB, &info) == 0; | ||
| } |
Contributor
There was a problem hiding this comment.
🧩 Analysis chain
🏁 Script executed:
#!/bin/bash
# Verify current implementation uses a single read and no EINTR/short-read loop.
rg -n -A20 -B5 'static int rdbFileUsesStreamingCompression|read\(' src/aof.cRepository: valkey-io/valkey
Length of output: 4080
🏁 Script executed:
#!/bin/bash
set -euo pipefail
rg -n 'rdbFileUsesStreamingCompression\s*\(' src/aof.c
# Show surrounding logic near the call site(s)
rg -n -A30 -B10 'rdbFileUsesStreamingCompression\s*\(' src/aof.cRepository: valkey-io/valkey
Length of output: 3311
🏁 Script executed:
#!/bin/bash
# Print exact call-site blocks with line numbers (keep output small)
# (Assumes repo is valkey-io/valkey and file exists)
nl -ba src/aof.c | sed -n '1010,1105p'Repository: valkey-io/valkey
Length of output: 101
🏁 Script executed:
#!/bin/bash
set -euo pipefail
# Show the remainder of restartAOFWithSyncRdb including cleanup label; avoid nl
cat -n src/aof.c | sed -n '1030,1135p'Repository: valkey-io/valkey
Length of output: 4875
🏁 Script executed:
#!/bin/bash
set -euo pipefail
cat -n src/aof.c | sed -n '1135,1210p'Repository: valkey-io/valkey
Length of output: 3513
Handle short (and EINTR) reads when probing the VKCS header in rdbFileUsesStreamingCompression()
restartAOFWithSyncRdb() reuses the sync RDB as the AOF base when rdbFileUsesStreamingCompression() returns 0. Since the helper does a single read() and returns 0 on any short read, a compressed RDB can be misclassified as uncompressed and incorrectly reused.
🔧 Proposed fix
static int rdbFileUsesStreamingCompression(const char *filename) {
unsigned char header[VKCS_ENVELOPE_SIZE];
int fd = open(filename, O_RDONLY);
if (fd == -1) return -1;
- ssize_t nread = read(fd, header, sizeof(header));
- int read_errno = errno;
- close(fd);
-
- if (nread == -1) {
- errno = read_errno;
- return -1;
- }
- if (nread < (ssize_t)sizeof(header)) return 0;
+ size_t total = 0;
+ while (total < sizeof(header)) {
+ ssize_t nread = read(fd, header + total, sizeof(header) - total);
+ if (nread == -1) {
+ if (errno == EINTR) continue;
+ int read_errno = errno;
+ close(fd);
+ errno = read_errno;
+ return -1;
+ }
+ if (nread == 0) break;
+ total += (size_t)nread;
+ }
+ close(fd);
+ if (total < sizeof(header)) return 0;
streamReaderInfo info = {0};
return streamReadEnvelopeInfo(header, sizeof(header), STREAM_KIND_RDB, &info) == 0;
}📝 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.
Suggested change
| static int rdbFileUsesStreamingCompression(const char *filename) { | |
| unsigned char header[VKCS_ENVELOPE_SIZE]; | |
| int fd = open(filename, O_RDONLY); | |
| if (fd == -1) return -1; | |
| ssize_t nread = read(fd, header, sizeof(header)); | |
| int read_errno = errno; | |
| close(fd); | |
| if (nread == -1) { | |
| errno = read_errno; | |
| return -1; | |
| } | |
| if (nread < (ssize_t)sizeof(header)) return 0; | |
| streamReaderInfo info = {0}; | |
| return streamReadEnvelopeInfo(header, sizeof(header), STREAM_KIND_RDB, &info) == 0; | |
| } | |
| static int rdbFileUsesStreamingCompression(const char *filename) { | |
| unsigned char header[VKCS_ENVELOPE_SIZE]; | |
| int fd = open(filename, O_RDONLY); | |
| if (fd == -1) return -1; | |
| size_t total = 0; | |
| while (total < sizeof(header)) { | |
| ssize_t nread = read(fd, header + total, sizeof(header) - total); | |
| if (nread == -1) { | |
| if (errno == EINTR) continue; | |
| int read_errno = errno; | |
| close(fd); | |
| errno = read_errno; | |
| return -1; | |
| } | |
| if (nread == 0) break; | |
| total += (size_t)nread; | |
| } | |
| close(fd); | |
| if (total < sizeof(header)) return 0; | |
| streamReaderInfo info = {0}; | |
| return streamReadEnvelopeInfo(header, sizeof(header), STREAM_KIND_RDB, &info) == 0; | |
| } |
🤖 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/aof.c` around lines 1009 - 1026, rdbFileUsesStreamingCompression
currently does a single read() into header and treats any short read as
“uncompressed”, which misclassifies compressed RDBs when read() is interrupted
or returns partial data; change the single read to a loop that repeatedly calls
read() (handling EINTR by continuing), accumulates bytes into header until
sizeof(header) is filled or read() returns 0/-1, preserve the original errno in
read_errno on failure, close fd before returning, and only consider the file
“short” (return 0) if the total bytes read is less than sizeof(header) (i.e.
EOF) — otherwise call streamReadEnvelopeInfo(header, sizeof(header),
STREAM_KIND_RDB, &info) as before; refer to function
rdbFileUsesStreamingCompression, variables header, nread/read_errno, fd, and
streamReadEnvelopeInfo.
Comment on lines
+521
to
+530
| /* Try LZF compression — values under 20 bytes don't compress, skip those. | ||
| * Skip per-string LZF when the rio has whole-stream compression so we | ||
| * don't compress twice; standalone rios (DUMP, AOF rewrite, diskless) | ||
| * still hit this path. */ | ||
| if (server.rdb_compression && len > 20 && | ||
| !(rdb && (rdb->flags & RIO_FLAG_STREAMING_COMPRESSION))) { | ||
| n = rdbSaveLzfStringObject(rdb, s, len); | ||
| if (n == -1) return -1; | ||
| if (n > 0) return n; | ||
| /* Return value of 0 means data can't be compressed, save the old way */ | ||
| /* 0 means data can't be compressed; fall through and store verbatim. */ |
Contributor
There was a problem hiding this comment.
Keep rdbSavedObjectLen() aligned with the real save encoding.
This gate only disables per-string LZF when a live rio advertises RIO_FLAG_STREAMING_COMPRESSION. rdbSavedObjectLen() still reaches rdbSaveObject() with rdb == NULL, so once whole-RDB streaming compression is enabled it can keep modeling the old LZF path and return a different size than the actual save path writes. Any caller using that helper for sizing or on-disk length reporting will drift in this mode.
🤖 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 521 - 530, rdbSavedObjectLen currently models
per-string LZF even when the real save path will use whole-stream (rio)
compression and thus skip per-string LZF; update rdbSavedObjectLen so its length
calculation matches rdbSaveObject by not counting LZF when the save will use
streaming compression: detect the same condition used in rdbSaveObject
(RIO_FLAG_STREAMING_COMPRESSION) and when rdb==NULL assume whole-stream
compression is active (or better, add a parameter/flag to rdbSavedObjectLen to
indicate streaming-compression mode) so rdbSavedObjectLen and rdbSaveObject (the
functions referenced) return the same encoded size. Ensure you update callers if
you add a parameter.
Member
Author
|
Closing this, discussed with @sarthakaggarwal97 offline and decided to merge it with Streaming Compression support for Replication |
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
2 participants
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
This PR is implemented over #3531 as part of #3195
To review this before #3531 look at the actual changes of this PR here: https://github.com/roshkhatri/valkey/pull/16/changes
Scope: This PR adds only the replication compression handshake and a boolean config. No replication logic. The primary records a new capability bit but takes no action on it. Compressed replication transport lands in a follow-up PR.
This is the negotiation layer for an upcoming compressed replication feature. Separating negotiation from transport keeps each PR small and independently reviewable
What's in this PR
New
REPLICA_CAPA_COMPRESSION (1 << 4)capability flag and matchingREPLICA_CAPA_COMPRESSION_STR "compression"New
repl_compressionfield onstruct valkeyServerREPL_COMPRESSION_ALGO(fixed toALGO_LZ4) andREPL_COMPRESSION_LEVEL(fixed to5). This is internal constants consumed by the future transport path. These will become user-configurable (repl-compression-algo,repl-compression-level) in a later release once benchmarks inform sensible defaults.New
replcompressionboolean config (defaultno)Naming follows the existing
rdbcompressionin Streaming Compression support for RDB #3531src/replication.cREPLICA_CAPA_COMPRESSIONinreplconfCommandwhen a replica advertisescapa compressioncapa compressioninsyncWithPrimaryHandleSendHandshakeStatewhenserver.repl_compression && useDisklessLoad()argv[]/lens[]grown from 9 to 11 entries to accommodate the new capa pair for compression capauseDisklessLoad()extracted into a local reused by both the existing skip-rdb-checksum check and the new compression check (may change after Streaming Compression support for RDB #3531 )This will be backward compatible.
capa compressionis a new REPLCONF capa pair; unrecognized capas are silently ignored by older primaries per existing convention.replcompression yesis persisted viaCONFIG REWRITE, a downgrade to a pre-PR binary will fail to start on the unknown directive. Remove the directive fromvalkey.confbefore downgrading.Follow-up next PR (out of scope)