From 5442bed8a7d823e62a6b7ed3dd4864cce8bd790d Mon Sep 17 00:00:00 2001 From: Roshan Khatri Date: Thu, 30 Jul 2026 08:24:05 +0000 Subject: [PATCH] Streaming Compression support for replication stream Signed-off-by: Roshan Khatri --- .github/workflows/ci.yml | 33 + cmake/Modules/SourceFiles.cmake | 3 +- src/Makefile | 1 + src/compression_repl.c | 207 ++++ src/compression_repl.h | 91 ++ src/compression_stream.c | 71 +- src/compression_stream.h | 13 + src/config.c | 27 + src/io_threads.c | 6 + src/networking.c | 206 +++- src/replication.c | 247 +++- src/server.c | 39 +- src/server.h | 85 +- src/unit/test_repl_compression.cpp | 87 ++ tests/integration/block-repl.tcl | 2 +- .../integration/dual-channel-replication.tcl | 34 +- tests/integration/repl-compression.tcl | 1001 +++++++++++++++++ tests/integration/replication-2.tcl | 2 +- tests/integration/replication-3.tcl | 2 +- tests/integration/replication-4.tcl | 10 +- tests/integration/replication-aof-sync.tcl | 2 +- tests/integration/replication.tcl | 16 +- tests/integration/skip-rdb-checksum.tcl | 2 +- tests/support/server.tcl | 2 +- tests/test_helper.tcl | 2 +- valkey.conf | 14 + 26 files changed, 2113 insertions(+), 92 deletions(-) create mode 100644 src/compression_repl.c create mode 100644 src/compression_repl.h create mode 100644 src/unit/test_repl_compression.cpp create mode 100644 tests/integration/repl-compression.tcl diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index 350bfaf5748..86f2aa31a6a 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -197,6 +197,39 @@ jobs: - name: test-tls-builtin run: ./runtest --verbose --single unit/tls --dump-logs --tls + test-replication-compression: + # Re-runs the replication suite with repl-compression=lz4-stream applied + # globally via --config. Test files opt in by carrying the top-level "repl-compression" + # tag (see tests/support/server.tcl), so this job selects them with + # --tags repl-compression instead of a hardcoded filename list that would + # silently drift if a file is renamed. Exercises the streaming-compression + # transport across the replication surface (full sync, dual-channel, buffer + # management, AOF-sync, etc.) to catch regressions that only surface under + # compressed replication. + # Not tagged for this job: integration/replication-psync (start_server lives + # inside a proc, top-level tags cannot apply) and the replication-buffer / + # dual-channel buffer-memory tests (assert exact byte volumes that compression changes). + runs-on: ubuntu-latest + steps: + - name: Install libbacktrace + uses: actions/checkout@de0fac2e4500dabe0009e67214ff5f5447ce83dd # v6.0.2 + with: + repository: ianlancetaylor/libbacktrace + ref: b9e40069c0b47a722286b94eb5231f7f05c08713 + path: libbacktrace + - run: cd libbacktrace && ./configure && make && sudo make install + - name: Checkout Valkey + uses: actions/checkout@de0fac2e4500dabe0009e67214ff5f5447ce83dd # v6.0.2 + - name: make + run: make -j4 SERVER_CFLAGS='-Werror' USE_LIBBACKTRACE=yes + - name: install test dependencies + run: sudo apt-get install -y tcl8.6 tclx + - name: replication tests with streaming compression enabled + run: | + ./runtest --verbose --dump-logs \ + --tags "repl-compression -slow" \ + --config repl-compression lz4-stream + build-debian-old: runs-on: ubuntu-latest container: debian:bullseye diff --git a/cmake/Modules/SourceFiles.cmake b/cmake/Modules/SourceFiles.cmake index 7fb2264ed58..6dab7c0c6b7 100644 --- a/cmake/Modules/SourceFiles.cmake +++ b/cmake/Modules/SourceFiles.cmake @@ -124,7 +124,8 @@ set(VALKEY_SERVER_SRCS ${CMAKE_SOURCE_DIR}/src/queues.c ${CMAKE_SOURCE_DIR}/src/compression.c ${CMAKE_SOURCE_DIR}/src/compression_lz4.c - ${CMAKE_SOURCE_DIR}/src/compression_stream.c) + ${CMAKE_SOURCE_DIR}/src/compression_stream.c + ${CMAKE_SOURCE_DIR}/src/compression_repl.c) # valkey-cli diff --git a/src/Makefile b/src/Makefile index ac08367ad00..9171ed90ab2 100644 --- a/src/Makefile +++ b/src/Makefile @@ -491,6 +491,7 @@ ENGINE_SERVER_OBJ = \ compression.o \ compression_lz4.o \ compression_stream.o \ + compression_repl.o \ config.o \ connection.o \ crc16.o \ diff --git a/src/compression_repl.c b/src/compression_repl.c new file mode 100644 index 00000000000..8095f866615 --- /dev/null +++ b/src/compression_repl.c @@ -0,0 +1,207 @@ +/* + * Copyright (c) Valkey Contributors + * All rights reserved. + * SPDX-License-Identifier: BSD-3-Clause + */ + +#include "compression_repl.h" +#include "zmalloc.h" + +/* Release the decode scratch buffer once it grows past this so a replica does + * not retain peak allocation. ~16 x PROTO_REPLY_CHUNK_BYTES, expressed locally + * to keep this adapter independent of server.h. */ +#define REPL_COMPRESSION_RETAIN_LIMIT (256 * 1024) + +/* Retain the encode staging buffer up to a full batch's LZ4 worst-case output. */ +#define REPL_COMPRESSION_ENCODE_RETAIN_LIMIT \ + (REPL_COMPRESSION_BATCH_LIMIT + (REPL_COMPRESSION_BATCH_LIMIT / 255) + 1024) + +/* Decoded-output room offered to the codec per feed iteration. */ +#define REPL_DECODE_CHUNK (16 * 1024) + +/* ===== Primary-side per-replica compressor ===== */ + +replCompressor *replCompressorCreate(compressionAlgo algo, int level) { + replCompressor *rc = zcalloc(sizeof(*rc)); + + streamWriterConfig cfg = { + .algo = algo, + .level = level, + .codec_checksum_enabled = false, + .stream_kind = VCS_STREAM_REPL, + }; + rc->out_buf = sdsempty(); + if (streamWriterInit(&rc->writer, &cfg, NULL, NULL) != 0) { + sdsfree(rc->out_buf); + zfree(rc); + return NULL; + } + streamWriterSetSink(&rc->writer, &rc->out_buf); + return rc; +} + +void replCompressorDestroy(replCompressor *rc) { + if (!rc) return; + streamWriterFree(&rc->writer); + sdsfree(rc->out_buf); + zfree(rc); +} + +int replCompressorWrite(replCompressor *rc, const void *buf, size_t len) { + return streamWriterWrite(&rc->writer, buf, len); +} + +int replCompressorFlush(replCompressor *rc) { + return streamWriterFlush(&rc->writer); +} + +void replCompressorResetBatch(replCompressor *rc) { + /* Decide retention from the payload length, not sdsalloc: SDS grows capacity + * greedily, so a normal batch would otherwise be reclaimed and reallocated + * every cycle. A batch is capped at REPL_COMPRESSION_BATCH_LIMIT, so its + * output stays within the bound; only an oversized payload is reclaimed. */ + size_t used = sdslen(rc->out_buf); + sdsclear(rc->out_buf); + rc->out_buf_pos = 0; + rc->raw_bytes = 0; + if (used > REPL_COMPRESSION_ENCODE_RETAIN_LIMIT) { + sdsfree(rc->out_buf); + rc->out_buf = sdsempty(); + } +} + +size_t replCompressorMemUsage(const replCompressor *rc) { + if (!rc) return 0; + /* Codec context memory is small and fixed; only the staging SDS is measured. */ + size_t total = sizeof(*rc); + if (rc->out_buf) total += sdsalloc(rc->out_buf); + return total; +} + +compressionAlgo replCompressorAlgo(const replCompressor *rc) { + return rc ? rc->writer.compressor.algo : ALGO_NONE; +} + +/* ===== Replica-side decompressor ===== */ + +replDecompressor *replDecompressorCreate(void) { + replDecompressor *rd = zcalloc(sizeof(*rd)); + rd->decode_buf = sdsempty(); + return rd; +} + +void replDecompressorDestroy(replDecompressor *rd) { + if (!rd) return; + if (rd->mode == REPL_DECODE_MODE_COMPRESSED) streamDecompressorFree(&rd->decompressor); + sdsfree(rd->decode_buf); + zfree(rd); +} + +/* Append raw bytes to the decode buffer (passthrough), enforcing output_max. */ +static replDecodeResult replDecodeEmit(replDecompressor *rd, const uint8_t *in, size_t len, size_t output_max) { + if (len == 0) return REPL_DECODE_OK; + if (sdslen(rd->decode_buf) + len > output_max) return REPL_DECODE_OVERFLOW; + rd->decode_buf = sdscatlen(rd->decode_buf, in, len); + return REPL_DECODE_OK; +} + +/* Drain compressed bytes [in, in+len) through the codec into rd->decode_buf, + * bounded by output_max (decompression-bomb guard). */ +static replDecodeResult replDecodeFeed(replDecompressor *rd, const uint8_t *in, size_t len, size_t output_max) { + size_t off = 0; + while (off < len) { + size_t used = sdslen(rd->decode_buf); + /* Budget exhausted with input still pending: the stream expands past + * the bomb-guard cap. Checked up front so the codec is never handed + * more room than the remaining budget allows. */ + if (used >= output_max) return REPL_DECODE_OVERFLOW; + size_t room = REPL_DECODE_CHUNK; + if (room > output_max - used) room = output_max - used; + rd->decode_buf = sdsMakeRoomFor(rd->decode_buf, room); + size_t consumed = 0; + ssize_t produced = streamDecompressorFeed(&rd->decompressor, + (uint8_t *)rd->decode_buf + used, + room, + in + off, len - off, &consumed); + if (produced < 0 || consumed > len - off) return REPL_DECODE_ERR; + if (produced > 0) sdsIncrLen(rd->decode_buf, (size_t)produced); + off += consumed; + /* A long-lived replication stream must never reach a compressed frame + * end. If it does, the stream is corrupt or the primary sent an + * unexpected terminator: the caller should disconnect. */ + if (rd->decompressor.frame_done) return REPL_DECODE_FRAME_DONE; + /* The codec always makes progress given input and output room; no + * progress with input still pending is a stuck state. Fail rather + * than let the caller drop the unconsumed tail. */ + if (consumed == 0 && produced == 0) return REPL_DECODE_ERR; + } + return REPL_DECODE_OK; +} + +replDecodeResult replDecompressorDecode(replDecompressor *rd, + const void *src, + size_t len, + size_t output_max, + size_t *out_len) { + static const uint8_t vcs_magic[VCS_MAGIC_SIZE] = {VCS_MAGIC_0, VCS_MAGIC_1, VCS_MAGIC_2}; + + if (out_len) *out_len = 0; + sdsclear(rd->decode_buf); + + const uint8_t *in = src; + size_t off = 0; + + /* Probe phase: classify the stream from its leading bytes. The magic may + * arrive split across feeds, so bytes accumulate until the prefix matches + * or rules out the VCS magic. */ + if (rd->mode == REPL_DECODE_MODE_PROBE) { + while (rd->envelope_len < VCS_MAGIC_SIZE && off < len) { + if (in[off] != vcs_magic[rd->envelope_len]) { + rd->mode = REPL_DECODE_MODE_PASSTHROUGH; /* Not a VCS stream. */ + break; + } + rd->envelope[rd->envelope_len++] = in[off++]; + } + + if (rd->mode == REPL_DECODE_MODE_PROBE) { + /* Magic matches so far; gather the rest of the envelope. */ + while (rd->envelope_len < VCS_ENVELOPE_SIZE && off < len) rd->envelope[rd->envelope_len++] = in[off++]; + if (rd->envelope_len < VCS_ENVELOPE_SIZE) return REPL_DECODE_OK; /* Need more header. */ + + compressionAlgo algo = ALGO_NONE; + if (streamParseVcsEnvelope(rd->envelope, VCS_ENVELOPE_SIZE, VCS_STREAM_REPL, &algo) != 0) + return REPL_DECODE_ERR; + if (streamDecompressorInit(&rd->decompressor, algo, false) != 0) return REPL_DECODE_ERR; + rd->mode = REPL_DECODE_MODE_COMPRESSED; + } + } + + if (rd->mode == REPL_DECODE_MODE_PASSTHROUGH) { + /* Replay any buffered magic-prefix bytes once, then forward the rest. */ + replDecodeResult r = replDecodeEmit(rd, rd->envelope, rd->envelope_len, output_max); + rd->envelope_len = 0; + if (r != REPL_DECODE_OK) return r; + r = replDecodeEmit(rd, in + off, len - off, output_max); + if (r != REPL_DECODE_OK) return r; + } else if (off < len) { + replDecodeResult r = replDecodeFeed(rd, in + off, len - off, output_max); + if (r != REPL_DECODE_OK) return r; + } + + /* Shrink the scratch buffer if it grew large and is now mostly empty. */ + if (sdsalloc(rd->decode_buf) > REPL_COMPRESSION_RETAIN_LIMIT && + sdslen(rd->decode_buf) < sdsalloc(rd->decode_buf) / 4) { + rd->decode_buf = sdsRemoveFreeSpace(rd->decode_buf, 0); + } + + if (out_len) *out_len = sdslen(rd->decode_buf); + return REPL_DECODE_OK; +} + +sds replDecompressorBuf(replDecompressor *rd) { + return rd ? rd->decode_buf : NULL; +} + +bool replDecompressorIsPassthrough(const replDecompressor *rd) { + return rd && rd->mode == REPL_DECODE_MODE_PASSTHROUGH; +} diff --git a/src/compression_repl.h b/src/compression_repl.h new file mode 100644 index 00000000000..6df545c7ed8 --- /dev/null +++ b/src/compression_repl.h @@ -0,0 +1,91 @@ +/* + * Copyright (c) Valkey Contributors + * All rights reserved. + * SPDX-License-Identifier: BSD-3-Clause + */ + +#ifndef COMPRESSION_REPL_H +#define COMPRESSION_REPL_H + +/* Replication compression adapter on top of compression_stream. The rio-based + * streamWriter/streamReader path serves blocking RDB save/load; replication + * reads a non-blocking socket in the event loop where a pull-mode reader + * cannot work, so this adapter is push-mode. The incoming stream may be + * compressed (VCS envelope) or plaintext (primary compresses only if its own + * config is enabled) and is classified from its leading bytes. */ + +#include "compression.h" +#include "compression_stream.h" +#include "sds.h" + +/* Max raw replication-backlog bytes compressed per write dispatch cycle; + * bounds per-batch latency and the staging buffer. */ +#define REPL_COMPRESSION_BATCH_LIMIT (1024 * 1024) + +/* ===== Primary-side per-replica compressor ===== */ + +typedef struct replCompressor { + streamWriter writer; /* VCS/LZ4 frame encoder (VCS_STREAM_REPL). */ + sds out_buf; /* Compressed bytes staged for the socket. */ + size_t out_buf_pos; /* Next unsent byte offset within out_buf. */ + size_t raw_bytes; /* Raw backlog bytes represented by out_buf. */ +} replCompressor; + +/* Compressor API. Write and Flush return 0 on success and -1 on error; the + * encoder compresses directly into out_buf. Create returns NULL on error; + * Destroy is NULL-safe. */ +replCompressor *replCompressorCreate(compressionAlgo algo, int level); +void replCompressorDestroy(replCompressor *rc); +int replCompressorWrite(replCompressor *rc, const void *buf, size_t len); +int replCompressorFlush(replCompressor *rc); +/* Clears the staging buffer for a new batch; reclaims only oversized payloads. */ +void replCompressorResetBatch(replCompressor *rc); +/* Approximate heap usage for client-output-buffer accounting. */ +size_t replCompressorMemUsage(const replCompressor *rc); +/* Algorithm the compressor was initialized with; ALGO_NONE if rc is NULL. */ +compressionAlgo replCompressorAlgo(const replCompressor *rc); + +/* ===== Replica-side decompressor ===== */ + +typedef enum { + REPL_DECODE_OK = 0, /* Decoded (possibly 0) bytes; need more input next tick. */ + REPL_DECODE_ERR = -1, /* IO/feed/decoder error: disconnect. */ + REPL_DECODE_FRAME_DONE = -2, /* Frame ended on a live link: protocol corruption. */ + REPL_DECODE_OVERFLOW = -3, /* Decoded output exceeded the bomb-guard cap. */ +} replDecodeResult; + +typedef enum { + REPL_DECODE_MODE_PROBE = 0, /* Still classifying the stream. */ + REPL_DECODE_MODE_COMPRESSED, /* VCS envelope seen; decoder initialized. */ + REPL_DECODE_MODE_PASSTHROUGH /* Non-VCS stream; bytes forwarded as-is. */ +} replDecodeMode; + +/* Decoder for the single primary link. Accumulates the leading bytes (they may + * span several reads), classifies the stream, then either feeds the codec or + * forwards plaintext untouched. */ +typedef struct replDecompressor { + streamDecompressor decompressor; /* Valid once mode == COMPRESSED. */ + replDecodeMode mode; + uint8_t envelope[VCS_ENVELOPE_SIZE]; /* Leading bytes gathered during PROBE. */ + size_t envelope_len; + sds decode_buf; /* Most recent decoded bytes. */ +} replDecompressor; + +/* Decoder API. Create returns NULL on error; Destroy is NULL-safe. Decode + * feeds len transport bytes and drains decoded output into decode_buf + * (cleared first). On REPL_DECODE_OK, *out_len is the decoded byte count + * (0 means a partial frame was buffered; resume next tick). output_max caps + * decoded output as a decompression-bomb guard. */ +replDecompressor *replDecompressorCreate(void); +void replDecompressorDestroy(replDecompressor *rd); +replDecodeResult replDecompressorDecode(replDecompressor *rd, + const void *src, + size_t len, + size_t output_max, + size_t *out_len); +/* Decode scratch buffer (valid bytes = sdslen) after a decode. */ +sds replDecompressorBuf(replDecompressor *rd); +/* True once the probe classified the stream as plaintext (non-VCS). */ +bool replDecompressorIsPassthrough(const replDecompressor *rd); + +#endif /* COMPRESSION_REPL_H */ diff --git a/src/compression_stream.c b/src/compression_stream.c index 16e19c680ba..2c935b6427c 100644 --- a/src/compression_stream.c +++ b/src/compression_stream.c @@ -5,6 +5,7 @@ */ #include "compression_stream.h" +#include "serverassert.h" #include "zmalloc.h" #include #include @@ -24,9 +25,8 @@ static bool vcsHasMagicPrefix(const uint8_t *buf, size_t len) { return memcmp(buf, VCS_MAGIC, n) == 0; } -static int writeVcsEnvelope(streamWriterWriteFn write_cb, - void *ctx, - compressionAlgo algo) { +/* Fill the 7-byte VCS envelope. Returns -1 if algo has no wire codec id. */ +static int buildVcsEnvelope(uint8_t *out, compressionAlgo algo, uint8_t stream_kind) { uint8_t codec; switch (algo) { case ALGO_LZ4: @@ -43,14 +43,24 @@ static int writeVcsEnvelope(streamWriterWriteFn write_cb, [VCS_OFFSET_VERSION] = VCS_VERSION, [VCS_OFFSET_CODEC] = codec, [VCS_OFFSET_RESERVED] = 0, - [VCS_OFFSET_STREAM_KIND] = VCS_STREAM_RDB, + [VCS_OFFSET_STREAM_KIND] = stream_kind, }; + memcpy(out, envelope, VCS_ENVELOPE_SIZE); + return 0; +} + +static int writeVcsEnvelope(streamWriterWriteFn write_cb, + void *ctx, + compressionAlgo algo, + uint8_t stream_kind) { + uint8_t envelope[VCS_ENVELOPE_SIZE]; + if (buildVcsEnvelope(envelope, algo, stream_kind) != 0) return -1; return write_cb(ctx, envelope, VCS_ENVELOPE_SIZE) == 0 ? 0 : -1; } /* Reject a nonzero reserved byte so a future envelope extension fails loudly * rather than being silently misinterpreted. */ -static int readVcsEnvelope(const uint8_t *buf, compressionAlgo *algo) { +static int readVcsEnvelope(const uint8_t *buf, uint8_t expected_stream_kind, compressionAlgo *algo) { if (buf[VCS_OFFSET_VERSION] != VCS_VERSION) return -1; switch (buf[VCS_OFFSET_CODEC]) { @@ -61,10 +71,15 @@ static int readVcsEnvelope(const uint8_t *buf, compressionAlgo *algo) { return -1; } if (buf[VCS_OFFSET_RESERVED] != 0) return -1; - if (buf[VCS_OFFSET_STREAM_KIND] != VCS_STREAM_RDB) return -1; + if (buf[VCS_OFFSET_STREAM_KIND] != expected_stream_kind) return -1; return 0; } +int streamParseVcsEnvelope(const uint8_t *buf, size_t len, uint8_t expected_stream_kind, compressionAlgo *algo) { + if (len < VCS_ENVELOPE_SIZE || !vcsHasMagicPrefix(buf, VCS_MAGIC_SIZE)) return -1; + return readVcsEnvelope(buf, expected_stream_kind, algo); +} + /* ===== Streaming Writer ===== */ #define STREAM_WRITER_INPUT_CHUNK_SIZE (1024 * 1024) @@ -73,6 +88,7 @@ int streamWriterInit(streamWriter *writer, const streamWriterConfig *cfg, stream memset(writer, 0, sizeof(*writer)); writer->write_cb = write_cb; writer->write_ctx = write_ctx; + writer->stream_kind = cfg->stream_kind == 0 ? VCS_STREAM_RDB : cfg->stream_kind; if (streamCompressorInit(&writer->compressor, cfg->algo, cfg->level, cfg->codec_checksum_enabled) != 0) { @@ -82,12 +98,27 @@ int streamWriterInit(streamWriter *writer, const streamWriterConfig *cfg, stream return 0; } +void streamWriterSetSink(streamWriter *writer, sds *sink) { + /* Sink must be installed before the frame starts, so all output lands in + * one destination. */ + assert(writer->state == STREAM_WRITER_STATE_INITIAL); + writer->sink = sink; +} + /* Envelope is emitted lazily so a writer that's created but never written * doesn't leave a stub envelope on the sink. */ static int streamWriterEnsureEnvelope(streamWriter *writer) { if (writer->state == STREAM_WRITER_STATE_ACTIVE) return 0; if (writer->state != STREAM_WRITER_STATE_INITIAL) return -1; - if (writeVcsEnvelope(writer->write_cb, writer->write_ctx, writer->compressor.algo) != 0) { + if (writer->sink) { + uint8_t envelope[VCS_ENVELOPE_SIZE]; + if (buildVcsEnvelope(envelope, writer->compressor.algo, writer->stream_kind) != 0) { + writer->state = STREAM_WRITER_STATE_ERROR; + return -1; + } + *writer->sink = sdscatlen(*writer->sink, (char *)envelope, VCS_ENVELOPE_SIZE); + } else if (writeVcsEnvelope(writer->write_cb, writer->write_ctx, writer->compressor.algo, + writer->stream_kind) != 0) { writer->state = STREAM_WRITER_STATE_ERROR; return -1; } @@ -95,10 +126,34 @@ static int streamWriterEnsureEnvelope(streamWriter *writer) { return 0; } +/* Sink path: compress directly into the caller's sds tail. */ +static int streamWriterFeedToSink(streamWriter *writer, + const uint8_t *input, + size_t input_len, + compressFlushMode flush_mode) { + const size_t bound = streamCompressorOutputBound(&writer->compressor, input_len); + if (bound == 0) { + writer->state = STREAM_WRITER_STATE_ERROR; + return -1; + } + *writer->sink = sdsMakeRoomFor(*writer->sink, bound); + const ssize_t compressed = streamCompressorFeed(&writer->compressor, + (uint8_t *)(*writer->sink) + sdslen(*writer->sink), + sdsavail(*writer->sink), input, input_len, flush_mode); + if (compressed < 0) { + writer->state = STREAM_WRITER_STATE_ERROR; + return -1; + } + sdsIncrLen(*writer->sink, (size_t)compressed); + return 0; +} + static int streamWriterFeedAndWrite(streamWriter *writer, const uint8_t *input, size_t input_len, compressFlushMode flush_mode) { + if (writer->sink) return streamWriterFeedToSink(writer, input, input_len, flush_mode); + const size_t needed = streamCompressorOutputBound(&writer->compressor, input_len); if (needed > writer->out_buf_size) { writer->out_buf = zrealloc(writer->out_buf, needed); @@ -204,7 +259,7 @@ int streamReaderInit(streamReader *reader, const streamReaderConfig *cfg, stream } if (reader->probe.header_len == VCS_ENVELOPE_SIZE) { - if (readVcsEnvelope(reader->probe.header, &algo) != 0) { + if (readVcsEnvelope(reader->probe.header, VCS_STREAM_RDB, &algo) != 0) { streamReaderSetError(reader, STREAM_READER_ERROR_INCOMPATIBLE); return -1; } diff --git a/src/compression_stream.h b/src/compression_stream.h index 4da66b9061a..da205261ec8 100644 --- a/src/compression_stream.h +++ b/src/compression_stream.h @@ -8,6 +8,7 @@ #define COMPRESSION_STREAM_H #include "compression.h" +#include "sds.h" /* VCS envelope: * [0..2] magic "VCS" @@ -37,6 +38,8 @@ /* Identifies an RDB payload in the envelope. */ #define VCS_STREAM_RDB 0x01 +/* Identifies a replication stream payload in the envelope. */ +#define VCS_STREAM_REPL 0x02 typedef int (*streamWriterWriteFn)(void *ctx, const uint8_t *data, size_t len); /* Returns >0 bytes read, 0 on EOF, -1 on error. Partial reads allowed. */ @@ -48,6 +51,7 @@ typedef struct { compressionAlgo algo; int level; bool codec_checksum_enabled; + uint8_t stream_kind; /* 0 selects VCS_STREAM_RDB. */ } streamWriterConfig; typedef enum { @@ -63,6 +67,8 @@ typedef struct streamWriter { size_t out_buf_size; streamWriterWriteFn write_cb; void *write_ctx; + sds *sink; /* When set, compress directly into *sink instead of out_buf + write_cb. */ + uint8_t stream_kind; streamWriterState state; } streamWriter; @@ -70,6 +76,9 @@ typedef struct streamWriter { * error. The writer pushes compressed bytes to write_cb. Errors are sticky: * later operations fail without emitting bytes. */ int streamWriterInit(streamWriter *writer, const streamWriterConfig *cfg, streamWriterWriteFn write_cb, void *write_ctx); +/* Redirect compressed output (envelope + frames) straight into *sink, bypassing + * the internal scratch buffer and write callback. */ +void streamWriterSetSink(streamWriter *writer, sds *sink); int streamWriterWrite(streamWriter *writer, const void *buf, size_t len); /* Emits codec-buffered bytes while leaving the frame open. */ int streamWriterFlush(streamWriter *writer); @@ -78,6 +87,10 @@ int streamWriterFinish(streamWriter *writer); /* Releases resources without implicitly finalizing the frame. */ void streamWriterFree(streamWriter *writer); +/* Parse a complete VCS envelope, requiring the given stream kind. Returns 0 + * and sets *algo on success, -1 on any mismatch. */ +int streamParseVcsEnvelope(const uint8_t *buf, size_t len, uint8_t expected_stream_kind, compressionAlgo *algo); + /* ===== Reader ===== */ /* Default reader compressed-input/decompressed-output buffer size. Tiny caller diff --git a/src/config.c b/src/config.c index e31e241d378..77382e6c066 100644 --- a/src/config.c +++ b/src/config.c @@ -181,6 +181,10 @@ configEnum rdb_compression_enum[] = {{"no", RDB_COMPRESSION_NO}, {"lz4-stream", RDB_COMPRESSION_LZ4_STREAM}, {NULL, 0}}; +configEnum repl_compression_enum[] = {{"no", REPL_COMPRESSION_NO}, + {"lz4-stream", REPL_COMPRESSION_LZ4_STREAM}, + {NULL, 0}}; + /* Output buffer limits presets. */ clientBufferLimitsConfig clientBufferLimitsDefaults[CLIENT_TYPE_OBUF_COUNT] = { {0, 0, 0}, /* normal */ @@ -866,6 +870,10 @@ static dict *matchPatternsToConfigs(robj **patterns, int pattern_count) { * CONFIG SET implementation *----------------------------------------------------------------------------*/ +/* Set by the repl-compression apply callback. Reconnects are irreversible, so + * reconciliation is deferred until the whole CONFIG SET commits. */ +static int repl_compression_reconcile_pending = 0; + void configSetCommand(client *c) { const char *errstr = NULL; const char *invalid_arg_name = NULL; @@ -983,6 +991,10 @@ void configSetCommand(client *c) { } } + /* Reset deferred side-effect state before applies run; apply callbacks set + * it, and it is consumed only on a successful commit below. */ + repl_compression_reconcile_pending = 0; + /* Apply all configs after being set */ for (i = 0; i < config_count && apply_fns[i] != NULL; i++) { if (!apply_fns[i](&errstr)) { @@ -1005,6 +1017,11 @@ void configSetCommand(client *c) { ValkeyModuleConfigChangeV1 cc = {.num_changes = config_count, .config_names = config_names}; moduleFireServerEvent(VALKEYMODULE_EVENT_CONFIG, VALKEYMODULE_SUBEVENT_CONFIG_CHANGE, &cc); addReply(c, shared.ok); + /* CONFIG SET committed: now safe to run deferred irreversible side effects. */ + if (repl_compression_reconcile_pending) { + repl_compression_reconcile_pending = 0; + reconcileReplicaCompression(); + } goto end; err: @@ -2658,6 +2675,14 @@ static int updateJemallocBgThread(const char **err) { return 1; } +static int updateReplCompression(const char **err) { + UNUSED(err); + /* Record intent only; configSetCommand reconciles after the command commits, + * so a rolled-back CONFIG SET disconnects nothing. */ + repl_compression_reconcile_pending = 1; + return 1; +} + static int updateReplBacklogSize(const char **err) { UNUSED(err); resizeReplicationBacklog(); @@ -3341,12 +3366,14 @@ static int isValidDbHashSeed(sds val, const char **err) { return 1; } + standardConfig static_configs[] = { /* Bool configs */ createBoolConfig("rdbchecksum", NULL, IMMUTABLE_CONFIG, server.rdb_checksum, 1, NULL, NULL), createBoolConfig("daemonize", NULL, IMMUTABLE_CONFIG, server.daemonize, 0, NULL, NULL), createBoolConfig("always-show-logo", NULL, IMMUTABLE_CONFIG, server.always_show_logo, 0, NULL, NULL), createBoolConfig("protected-mode", NULL, MODIFIABLE_CONFIG, server.protected_mode, 1, NULL, NULL), + createEnumConfig("repl-compression", NULL, MODIFIABLE_CONFIG, repl_compression_enum, server.repl_compression, REPL_COMPRESSION_NO, NULL, updateReplCompression), createBoolConfig("rdb-del-sync-files", NULL, MODIFIABLE_CONFIG, server.rdb_del_sync_files, 0, NULL, NULL), createBoolConfig("activerehashing", NULL, MODIFIABLE_CONFIG, server.activerehashing, 1, NULL, NULL), createBoolConfig("stop-writes-on-bgsave-error", NULL, MODIFIABLE_CONFIG, server.stop_writes_on_bgsave_err, 1, NULL, NULL), diff --git a/src/io_threads.c b/src/io_threads.c index 0563eaddda0..baa7a517df7 100644 --- a/src/io_threads.c +++ b/src/io_threads.c @@ -321,6 +321,9 @@ static void *IOThreadMain(void *myid) { case JOB_REQ_POLL: ioThreadPoll((aeEventLoop *)data); break; + case JOB_REQ_WRITE_CLIENT: + ioThreadWriteToClient((client *)data); + break; default: serverPanic("Invalid SPSC job type: %d", type); } @@ -506,6 +509,9 @@ int trySendReadToIOThreads(client *c) { if (c->io_write_state == CLIENT_PENDING_IO) return C_OK; /* For simplicity, don't offload replica clients reads as read traffic from replica is negligible */ if (getClientType(c) == CLIENT_TYPE_REPLICA) return C_ERR; + /* A live replication decoder must run on the main thread; the IO-thread + * read path does not decode. Destroyed once the probe resolves to plaintext. */ + if (c->flag.primary && server.repl_decompressor) return C_ERR; /* With Lua debug client we may call connWrite directly in the main thread */ if (c->flag.lua_debug) return C_ERR; /* For simplicity let the main-thread handle the blocked clients */ diff --git a/src/networking.c b/src/networking.c index 025110c66de..f3ee5dfa4c1 100644 --- a/src/networking.c +++ b/src/networking.c @@ -37,6 +37,8 @@ #include "fpconv_dtoa.h" #include "fmtargs.h" #include "io_threads.h" +#include "compression_repl.h" +#include "monotonic.h" #include "module.h" #include "connection.h" #include "zmalloc.h" @@ -1785,6 +1787,14 @@ int clientHasPendingReplies(client *c) { /* Replicas use global shared replication buffer instead of * private output buffer. */ serverAssert(c->bufpos == 0 && listLength(c->reply) == 0); + + /* Unsent compressed data counts as pending. Skip while CLIENT_PENDING_IO: + * the IO thread owns out_buf; postWriteToReplica re-checks when the job completes. */ + if (c->repl_data->repl_compressor && c->io_write_state != CLIENT_PENDING_IO && + c->repl_data->repl_compressor->out_buf_pos < sdslen(c->repl_data->repl_compressor->out_buf)) { + return 1; + } + if (c->repl_data->ref_repl_buf_node == NULL) return 0; /* If the last replication buffer block content is totally sent, @@ -2137,6 +2147,8 @@ int freeClient(client *c) { return 0; } + replDestroyCompression(c); + /* For connected clients, call the disconnection event of modules hooks. */ if (c->conn) { moduleFireServerEvent(VALKEYMODULE_EVENT_CLIENT_CHANGE, VALKEYMODULE_SUBEVENT_CLIENT_CHANGE_DISCONNECTED, c); @@ -2433,25 +2445,20 @@ client *lookupClientByID(uint64_t id) { return c; } -static void postWriteToReplica(client *c) { - if (c->nwritten <= 0) return; - - server.stat_net_repl_output_bytes += c->nwritten; - - /* Locate the last node which has leftover data and - * decrement reference counts of all nodes in front of it. - * Set c->ref_repl_buf_node to point to the last node and - * c->ref_block_pos to the offset within that node */ +/* Advance the replica's replication-backlog cursor (ref_repl_buf_node / + * ref_block_pos) past consumed raw bytes, releasing the reference on each + * fully-sent block. Shared by the compressed and plaintext post-write paths. */ +static void advanceReplicaRefBlock(client *c, size_t consumed) { listNode *curr = c->repl_data->ref_repl_buf_node; listNode *next = NULL; - size_t nwritten = c->nwritten + c->repl_data->ref_block_pos; + size_t remaining = consumed + c->repl_data->ref_block_pos; replBufBlock *o = listNodeValue(curr); - while (nwritten >= o->used) { + while (remaining >= o->used) { next = listNextNode(curr); if (!next) break; /* End of list */ - nwritten -= o->used; + remaining -= o->used; o->refcount--; curr = next; @@ -2459,14 +2466,171 @@ static void postWriteToReplica(client *c) { o->refcount++; } - serverAssert(nwritten <= o->used); + serverAssert(remaining <= o->used); c->repl_data->ref_repl_buf_node = curr; - c->repl_data->ref_block_pos = nwritten; + c->repl_data->ref_block_pos = remaining; +} + +static void postWriteToReplica(client *c) { + /* Check compression error first; IO thread may have flagged this. */ + if (atomic_load_explicit(&c->repl_data->compression_error, memory_order_acquire)) { + serverLog(LL_WARNING, + "Compression error on replica %s (algo=%s, raw_bytes=%zu), disconnecting", + replicationGetReplicaName(c), + compressionAlgoName(replCompressorAlgo(c->repl_data->repl_compressor)), + c->repl_data->repl_compressor ? c->repl_data->repl_compressor->raw_bytes : 0); + freeClientAsync(c); + return; + } + + if (c->nwritten <= 0) return; + + server.stat_net_repl_output_bytes += c->nwritten; + + if (c->repl_data->repl_compressor) { + replCompressor *compressor = c->repl_data->repl_compressor; + /* The cursor advances by the batch's raw bytes only once out_buf is + * fully sent; a partial send keeps it pinned so the next cycle sends + * the remainder before compressing more. */ + if (compressor->out_buf_pos == sdslen(compressor->out_buf)) { + size_t raw_bytes = compressor->raw_bytes; + + advanceReplicaRefBlock(c, raw_bytes); + + c->repl_data->repl_uncompressed_bytes_total += raw_bytes; + c->repl_data->repl_compressed_bytes_total += sdslen(compressor->out_buf); + + replCompressorResetBatch(compressor); + + incrementalTrimReplicationBacklog(REPL_BACKLOG_TRIM_BLOCKS_PER_CALL); + } + return; + } + + advanceReplicaRefBlock(c, c->nwritten); incrementalTrimReplicationBacklog(REPL_BACKLOG_TRIM_BLOCKS_PER_CALL); } +/* Compressed write path for replicas on either the IO thread or the main thread. */ +static void writeToReplicaCompressed(client *c) { + replCompressor *compressor = c->repl_data->repl_compressor; + serverAssert(compressor != NULL); + + /* Finish sending the previous batch's leftover first; compressed bytes + * must reach the socket in order. */ + if (compressor->out_buf_pos < sdslen(compressor->out_buf)) { + size_t avail = sdslen(compressor->out_buf) - compressor->out_buf_pos; + c->nwritten = connWrite(c->conn, + compressor->out_buf + compressor->out_buf_pos, + avail); + if (c->nwritten <= 0) { + c->write_flags |= WRITE_FLAGS_WRITE_ERROR; + return; + } + compressor->out_buf_pos += c->nwritten; + /* Skip trim here; postWriteToReplica trims only after the batch fully + * drains. Worst-case trim delay is one batch (REPL_COMPRESSION_BATCH_LIMIT). */ + return; + } + + /* postWriteToReplica resets the batch once it fully drains, so a fresh + * batch always starts from an empty buffer. */ + serverAssert(sdslen(compressor->out_buf) == 0 && compressor->out_buf_pos == 0 && + compressor->raw_bytes == 0); + + listNode *last_node; + size_t bufpos; + if (inMainThread()) { + last_node = listLast(server.repl_buffer_blocks); + if (!last_node) return; + bufpos = ((replBufBlock *)listNodeValue(last_node))->used; + } else { + last_node = c->io_last_reply_block; + serverAssert(last_node != NULL); + bufpos = c->io_last_bufpos; + } + listNode *first_node = c->repl_data->ref_repl_buf_node; + + /* Compress new replication-backlog bytes, capped at + * REPL_COMPRESSION_BATCH_LIMIT raw bytes per cycle to bound per-batch + * latency and keep out_buf size predictable. */ + monotime compress_start = getMonotonicUs(); + size_t total_raw = 0; + for (listNode *cur = first_node; cur != NULL; cur = listNextNode(cur)) { + replBufBlock *block = listNodeValue(cur); + size_t start = (cur == first_node) ? c->repl_data->ref_block_pos : 0; + size_t end = (cur == last_node) ? bufpos : block->used; + + if (end <= start) { + serverAssert(end >= start); + if (cur == last_node) break; + continue; + } + + size_t len = end - start; + /* Cap this write at the remaining batch budget; the cursor resumes mid-block next cycle. */ + size_t remaining = REPL_COMPRESSION_BATCH_LIMIT - total_raw; + if (len > remaining) len = remaining; + int rc = replCompressorWrite(compressor, block->buf + start, len); + if (rc < 0) { + atomic_store_explicit(&c->repl_data->compression_error, 1, memory_order_release); + atomic_fetch_add_explicit(&c->repl_data->repl_compression_errors, 1, memory_order_relaxed); + c->write_flags |= WRITE_FLAGS_WRITE_ERROR; + return; + } + total_raw += len; + if (total_raw >= REPL_COMPRESSION_BATCH_LIMIT) break; + if (cur == last_node) break; + } + + if (total_raw == 0) return; + + /* Flush so the batch's compressed bytes land in out_buf. */ + if (replCompressorFlush(compressor) != 0) { + atomic_store_explicit(&c->repl_data->compression_error, 1, memory_order_release); + atomic_fetch_add_explicit(&c->repl_data->repl_compression_errors, 1, memory_order_relaxed); + c->write_flags |= WRITE_FLAGS_WRITE_ERROR; + return; + } + + /* Best-effort: a minor race with INFO reads is acceptable for timing metrics. */ + atomic_fetch_add_explicit(&c->repl_data->repl_compression_cpu_usec, + (long long)(getMonotonicUs() - compress_start), memory_order_relaxed); + + compressor->raw_bytes = total_raw; + + /* Send out_buf. The backlog cursor advances only after a full send + * (postWriteToReplica), so a partial send keeps it pinned to the start of + * the batch. */ + size_t avail = sdslen(compressor->out_buf); + if (avail == 0) { + if (total_raw > 0) { + /* Compressor produced no output despite non-zero input: treat as error + * to prevent infinite re-compression of the same data. */ + atomic_store_explicit(&c->repl_data->compression_error, 1, memory_order_release); + atomic_fetch_add_explicit(&c->repl_data->repl_compression_errors, 1, memory_order_relaxed); + c->write_flags |= WRITE_FLAGS_WRITE_ERROR; + } + return; + } + + c->nwritten = connWrite(c->conn, compressor->out_buf, avail); + if (c->nwritten <= 0) { + c->write_flags |= WRITE_FLAGS_WRITE_ERROR; + return; + } + compressor->out_buf_pos = c->nwritten; +} + static void writeToReplica(client *c) { + /* Compressed replicas use the framed write path; the decision lives here so + * callers do not branch on the per-replica compressor. */ + if (c->repl_data->repl_compressor != NULL) { + writeToReplicaCompressed(c); + return; + } + listNode *last_node; size_t bufpos; @@ -4354,8 +4518,16 @@ void readQueryFromClient(connection *conn) { bool repeat = false; int iter = 0; do { + size_t qblen_before_read = c->querybuf ? sdslen(c->querybuf) : 0; bool full_read = readToQueryBuf(c); if (handleReadResult(c) == C_OK) { + if (c->flag.primary && + server.repl_decompressor && + replDecompressQueryBuf(c, qblen_before_read) == C_ERR) { + serverLog(LL_WARNING, "Disconnecting primary due to replication stream decompression failure"); + freeClientAsync(c); + return; + } if (processInputBuffer(c) == C_ERR) return; trimCommandQueue(c); } @@ -6099,7 +6271,11 @@ size_t getClientOutputBufferMemoryUsage(client *c) { repl_buf_size = last->repl_offset + last->size - cur->repl_offset; repl_node_num = last->id - cur->id + 1; } - return repl_buf_size + (repl_node_size * repl_node_num); + size_t compressor_size = 0; + if (c->repl_data->repl_compressor) { + compressor_size = replCompressorMemUsage(c->repl_data->repl_compressor); + } + return repl_buf_size + (repl_node_size * repl_node_num) + compressor_size; } size_t list_item_size = sizeof(listNode) + sizeof(clientReplyBlock); diff --git a/src/replication.c b/src/replication.c index 36ccfa768f8..ee6d333cdb0 100644 --- a/src/replication.c +++ b/src/replication.c @@ -41,6 +41,8 @@ #include "connection.h" #include "module.h" #include "cluster_migrateslots.h" +#include "io_threads.h" +#include "compression_repl.h" #include #include @@ -84,6 +86,193 @@ ConnectionType *connTypeOfReplication(void) { * pair. Mostly useful for logging, since we want to log a replica using its * IP address and its listening port which is more clear for the user, for * example: "Closing connection with replica 10.1.2.3:6380". */ + +/* Reconcile replica transports after a runtime repl-compression change; a live + * link cannot switch mid-stream. Disabled: drop replicas on a compressed stream. + * Enabled: drop online replicas that are capable but still plaintext. Async + * disconnect keeps the iteration safe; still-syncing replicas are untouched. */ +void reconcileReplicaCompression(void) { + listIter li; + listNode *ln; + int disconnected = 0; + + listRewind(server.replicas, &li); + while ((ln = listNext(&li))) { + client *replica = ln->value; + if (!replica->repl_data) continue; + + int has_compressor = replica->repl_data->repl_compressor != NULL; + int mismatch; + if (server.repl_compression) { + mismatch = (replica->repl_data->repl_state == REPLICA_STATE_ONLINE && + (replica->repl_data->replica_capa & REPLICA_CAPA_COMPRESSION) && !has_compressor); + } else { + mismatch = has_compressor; + } + if (!mismatch) continue; + + serverLog(LL_NOTICE, "Disconnecting replica %s to renegotiate replication compression (now %s)", + replicationGetReplicaName(replica), server.repl_compression ? "enabled" : "disabled"); + freeClientAsync(replica); + disconnected++; + } + + if (disconnected > 0) { + serverLog(LL_NOTICE, "Disconnected %d replicas to reconcile replication compression", disconnected); + } + + /* Capability with the upstream primary was negotiated at handshake; drop the + * link so the reconnect re-advertises with the new setting. */ + if (server.primary_host) { + if (server.primary) { + serverLog(LL_NOTICE, "Disconnecting from primary to renegotiate replication compression (now %s)", + server.repl_compression ? "enabled" : "disabled"); + freeClientAsync(server.primary); + } else if (cancelReplicationHandshake(1)) { + serverLog(LL_NOTICE, "Restarting sync with primary to renegotiate replication compression (now %s)", + server.repl_compression ? "enabled" : "disabled"); + } + } +} + +static bool shouldEnableReplicaCompression(client *c) { + if (!server.repl_compression || !c || !c->repl_data) return false; + return (c->repl_data->replica_capa & REPLICA_CAPA_COMPRESSION) != 0; +} + +static int replInitCompression(client *c, compressionAlgo algo, int level) { + if (!c || !c->repl_data) return C_ERR; + + replDestroyCompression(c); + + c->repl_data->repl_compressor = replCompressorCreate(algo, level); + if (!c->repl_data->repl_compressor) return C_ERR; + + atomic_store_explicit(&c->repl_data->compression_error, 0, memory_order_relaxed); + + return C_OK; +} + +/* Destroy a replica's compression state and stats. Waits for any in-flight + * IO write job first; the IO thread owns the compressor during a write. */ +void replDestroyCompression(client *c) { + if (!c || !c->repl_data) return; + + if (c->repl_data->repl_compressor) { + waitForClientIO(c); + + replCompressorDestroy(c->repl_data->repl_compressor); + c->repl_data->repl_compressor = NULL; + } + + atomic_store_explicit(&c->repl_data->compression_error, 0, memory_order_relaxed); + c->repl_data->repl_compressed_bytes_total = 0; + c->repl_data->repl_uncompressed_bytes_total = 0; + atomic_store_explicit(&c->repl_data->repl_compression_errors, 0, memory_order_relaxed); + atomic_store_explicit(&c->repl_data->repl_compression_cpu_usec, 0, memory_order_relaxed); +} + +/* Initialize framed transport compression for a replica at PSYNC completion. + * The stream writer emits the VCS envelope lazily on its first write. */ +static int replicaInitCompressionOnPsync(client *c) { + /* Derive the codec from the configured mode, mirroring the RDB path. */ + compressionAlgo algo = server.repl_compression == REPL_COMPRESSION_LZ4_STREAM ? ALGO_LZ4 : ALGO_NONE; + int level = 0; /* Codec default. */ + + serverAssert(c->io_write_state == CLIENT_IDLE); + if (replInitCompression(c, algo, level) != C_OK) { + serverLog(LL_WARNING, "Failed to initialize compression for replica %s", + replicationGetReplicaName(c)); + return C_ERR; + } + + serverLog(LL_NOTICE, "Replication compression enabled for replica %s (algo=%s, level=%d)", + replicationGetReplicaName(c), compressionAlgoName(algo), level); + return C_OK; +} + +static void replDestroyDecompression(void) { + if (server.repl_decompressor) { + replDecompressorDestroy(server.repl_decompressor); + server.repl_decompressor = NULL; + } +} + +/* (Re)create the replica-side decompressor for a fresh stream. */ +static void replRefreshDecompression(void) { + replDestroyDecompression(); + server.repl_decompressor = replDecompressorCreate(); +} + +/* Cap on decompressed output per replication batch. Compressor batches at + * 1 MB raw input; 256 MB tolerates realistic ratios. Excess indicates a + * malicious or buggy primary; disconnect rather than allocate unbounded. */ +#define REPL_STREAM_DECODER_OUTPUT_MAX (256 * 1024 * 1024) + +/* Decompress newly-read replication stream data in the query buffer. + * Replaces the raw compressed bytes (from new_data_start onward) with + * decompressed output and adjusts the replica's read_reploff accordingly. */ +int replDecompressQueryBuf(client *c, size_t new_data_start) { + size_t raw_input_len, decompressed_len; + + serverAssert(server.repl_decompressor != NULL); + serverAssert(c != NULL); + serverAssert(c->querybuf != NULL); + serverAssert(new_data_start <= sdslen(c->querybuf)); + + raw_input_len = sdslen(c->querybuf) - new_data_start; + if (raw_input_len == 0) return C_OK; + + monotime decompress_start = getMonotonicUs(); + + /* Feed transport bytes and drain decoded output through the adapter. */ + replDecodeResult dr = replDecompressorDecode(server.repl_decompressor, + c->querybuf + new_data_start, raw_input_len, + REPL_STREAM_DECODER_OUTPUT_MAX, &decompressed_len); + if (dr != REPL_DECODE_OK) { + if (dr == REPL_DECODE_FRAME_DONE) + serverLog(LL_WARNING, "Primary closed compressed replication frame unexpectedly"); + server.repl_decompression_errors++; + return C_ERR; + } + + sds decode_buf = replDecompressorBuf(server.repl_decompressor); + + if (new_data_start == 0) { + sdsclear(c->querybuf); + } else { + sdsrange(c->querybuf, 0, new_data_start - 1); + } + + /* sdscatlen may reallocate c->querybuf. If this client is still on the + * thread-local shared query buffer, give it ownership of the current buffer + * first so the shared pointer is not left dangling for later readers. */ + if (c->querybuf == thread_shared_qb && sdsavail(c->querybuf) < decompressed_len) { + initSharedQueryBuf(); + } + c->querybuf = sdscatlen(c->querybuf, decode_buf, decompressed_len); + if (c->querybuf_peak < sdslen(c->querybuf)) c->querybuf_peak = sdslen(c->querybuf); + + /* Convert the transport bytes already counted in handleReadResult() into + * logical replication bytes before processInputBuffer() observes the stream. + * If the reader buffered a partial compressed frame and emitted 0 bytes, + * the logical offset stays unchanged until a later read produces output. */ + c->repl_data->read_reploff -= (long long)raw_input_len; + c->repl_data->read_reploff += (long long)decompressed_len; + + server.repl_decompression_cpu_usec += getMonotonicUs() - decompress_start; + server.repl_decompressed_bytes_total += decompressed_len; + + /* Plaintext stream confirmed: the decoder is pure overhead from here on. + * Drop it so later reads skip this path entirely (callers gate on + * server.repl_decompressor) and primary-link reads can use IO threads. */ + if (replDecompressorIsPassthrough(server.repl_decompressor)) { + replDecompressorDestroy(server.repl_decompressor); + server.repl_decompressor = NULL; + } + return C_OK; +} + char *replicationGetReplicaName(client *c) { static char buf[NET_HOST_PORT_STR_LEN]; char ip[NET_IP_STR_LEN]; @@ -971,6 +1160,16 @@ int primaryTryPartialResynchronization(client *c, long long psync_offset) { freeClientAsync(c); return C_OK; } + + /* Initialize compression after +CONTINUE (plaintext) and before + * addReplyReplicationBacklog so backlog data goes through the compressed path. */ + if (shouldEnableReplicaCompression(c)) { + if (replicaInitCompressionOnPsync(c) != C_OK) { + freeClientAsync(c); + return C_OK; + } + } + psync_len = addReplyReplicationBacklog(c, psync_offset); serverLog( LL_NOTICE, @@ -1477,6 +1676,12 @@ void replconfCommand(client *c) { } } else if (!strcasecmp(objectGetVal(c->argv[j + 1]), REPLICA_CAPA_SKIP_RDB_CHECKSUM_STR)) c->repl_data->replica_capa |= REPLICA_CAPA_SKIP_RDB_CHECKSUM; + /* "compression": the replica can decode a compressed incremental + * replication stream. The primary compresses only when both sides + * enable repl-compression; a primary that does not understand the + * capability ignores it. */ + else if (!strcasecmp(objectGetVal(c->argv[j + 1]), REPLICA_CAPA_COMPRESSION_STR)) + c->repl_data->replica_capa |= REPLICA_CAPA_COMPRESSION; } else if (!strcasecmp(objectGetVal(c->argv[j]), "ack")) { /* REPLCONF ACK is used by replica to inform the primary the amount * of replication stream that it processed so far. It is an @@ -1625,6 +1830,14 @@ int replicaPutOnline(client *replica) { replica->repl_data->repl_state = REPLICA_STATE_ONLINE; replica->repl_data->repl_ack_time = server.unixtime; /* Prevent false timeout. */ + /* Initialize compression for full-sync replicas going online. */ + if (shouldEnableReplicaCompression(replica)) { + if (replicaInitCompressionOnPsync(replica) != C_OK) { + freeClientAsync(replica); + return 0; + } + } + refreshGoodReplicasCount(); /* Fire the replica change modules event. */ moduleFireServerEvent(VALKEYMODULE_EVENT_REPLICA_CHANGE, VALKEYMODULE_SUBEVENT_REPLICA_CHANGE_ONLINE, NULL); @@ -2418,6 +2631,7 @@ void replicaAfterLoadPrimaryRDB(connection *conn, rdbSaveInfo *rsi, int disk_bas server.repl_down_since = 0; /* Send the initial ACK immediately to put this replica in online state. */ replicationSendAck(); + if (server.repl_provisional_compression) replRefreshDecompression(); } /* Fire the primary link modules event. */ @@ -3426,8 +3640,15 @@ int streamReplDataBufToDb(client *c) { /* Read and process repl data block */ replDataBufBlock *o = listNodeValue(cur); used = o->used; + size_t qblen_before = sdslen(c->querybuf); c->querybuf = sdscatlen(c->querybuf, o->buf, used); c->repl_data->read_reploff += used; + if (server.repl_decompressor && + replDecompressQueryBuf(c, qblen_before) == C_ERR) { + serverLog(LL_WARNING, "Dual-channel replication stream decompression failure"); + blockingOperationEnds(); + return C_ERR; + } processInputBuffer(c); server.pending_repl_data.mem -= (used + sizeof(replDataBufBlock) + sizeof(listNode)); server.pending_repl_data.len -= used; @@ -3785,6 +4006,7 @@ int dualChannelReplMainConnRecvPsyncReply(connection *conn, sds *err) { serverCommunicateSystemd("STATUS=PRIMARY <-> REPLICA sync: Partial Resynchronization accepted. Ready to " "accept connections in read-write mode.\n"); } + if (server.repl_provisional_compression) replRefreshDecompression(); dualChannelSyncHandlePsync(); return C_OK; } @@ -3913,9 +4135,10 @@ int syncWithPrimaryHandleSendHandshakeState(connection *conn) { * The primary will ignore capabilities it does not understand. */ // we can ignore primary's conditions when sending capa (is_primary_stream_verified=1) - int send_skip_rdb_checksum_capa = replicationSupportSkipRDBChecksum(conn, useDisklessLoad(), 1); - char *argv[9] = {"REPLCONF", "capa", "eof", "capa", "psync2", NULL, NULL, NULL, NULL}; - size_t lens[9] = {8, 4, 3, 4, 6, 0, 0, 0, 0}; + int use_diskless_load = useDisklessLoad(); + int send_skip_rdb_checksum_capa = replicationSupportSkipRDBChecksum(conn, use_diskless_load, 1); + char *argv[11] = {"REPLCONF", "capa", "eof", "capa", "psync2", NULL, NULL, NULL, NULL, NULL, NULL}; + size_t lens[11] = {8, 4, 3, 4, 6, 0, 0, 0, 0, 0, 0}; int argc = 5; if (send_skip_rdb_checksum_capa) { argv[argc] = "capa"; @@ -3933,6 +4156,17 @@ int syncWithPrimaryHandleSendHandshakeState(connection *conn) { lens[argc] = strlen("dual-channel"); argc++; } + /* Capture the compression decision for this handshake so the decoder is + * set up from what was advertised, not a config that may change mid-stream. */ + server.repl_provisional_compression = server.repl_compression; + if (server.repl_provisional_compression) { + argv[argc] = "capa"; + lens[argc] = strlen("capa"); + argc++; + argv[argc] = REPLICA_CAPA_COMPRESSION_STR; + lens[argc] = strlen(REPLICA_CAPA_COMPRESSION_STR); + argc++; + } err = sendCommandArgv(conn, argc, argv, lens); if (err) goto err; @@ -4302,6 +4536,7 @@ void syncWithPrimary(connection *conn) { serverCommunicateSystemd("STATUS=PRIMARY <-> REPLICA sync: Partial Resynchronization accepted. Ready to " "accept connections in read-write mode.\n"); } + if (server.repl_provisional_compression) replRefreshDecompression(); return; } @@ -4579,6 +4814,7 @@ void replicationUnsetPrimary(void) { * the replicas will be able to partially resync with us, so it will be * a very fast reconnection. */ disconnectReplicas(); + replDestroyDecompression(); server.repl_state = REPL_STATE_NONE; /* We need to make sure the new primary will start the replication stream @@ -4638,6 +4874,11 @@ void replicationHandlePrimaryDisconnection(void) { /* Any other repl_state means the state machine already moved on * (e.g. REPL_STATE_CONNECT, CONNECTING, NONE) — leave it untouched. */ + /* Tear down the compressed replication decoder so a later (possibly + * uncompressed) primary stream isn't fed into stale frame state. + * Idempotent when no decoder exists. */ + replDestroyDecompression(); + /* We lost connection with our primary, don't disconnect replicas yet, * maybe we'll be able to PSYNC with our primary later. We'll disconnect * the replicas only if we'll have to do a full resync with our primary. */ diff --git a/src/server.c b/src/server.c index df2e91f9a7c..30163cf33ce 100644 --- a/src/server.c +++ b/src/server.c @@ -47,6 +47,8 @@ #include "threads_mngr.h" #include "fmtargs.h" #include "io_threads.h" +#include "compression.h" +#include "compression_repl.h" #include "tls.h" #include "sds.h" #include "module.h" @@ -2831,6 +2833,9 @@ void resetServerStats(void) { server.stat_sync_full = 0; server.stat_sync_partial_ok = 0; server.stat_sync_partial_err = 0; + server.repl_decompression_errors = 0; + server.repl_decompression_cpu_usec = 0; + server.repl_decompressed_bytes_total = 0; server.stat_io_reads_processed = 0; server.stat_total_reads_processed = 0; server.stat_io_writes_processed = 0; @@ -6634,6 +6639,17 @@ sds genValkeyInfoString(dict *section_dict, int all_sections, int everything) { "slave_priority:%d\r\n", server.replica_priority, "slave_read_only:%d\r\n", server.repl_replica_ro, "replica_announced:%d\r\n", server.replica_announced)); + if (server.repl_decompression_errors > 0) { + info = sdscatfmt(info, "repl_decompression_errors:%U\r\n", + (unsigned long long)server.repl_decompression_errors); + } + if (server.repl_decompressed_bytes_total > 0) { + info = sdscatfmt(info, + "repl_decompression_cpu_usec:%I\r\n" + "repl_decompressed_bytes_total:%U\r\n", + (long long)server.repl_decompression_cpu_usec, + (unsigned long long)server.repl_decompressed_bytes_total); + } } info = sdscatprintf(info, "connected_slaves:%lu\r\n", listLength(server.replicas)); @@ -6666,12 +6682,33 @@ sds genValkeyInfoString(dict *section_dict, int all_sections, int everything) { info = sdscatprintf(info, "slave%d:ip=%s,port=%d,state=%s," - "offset=%lld,lag=%ld,type=%s\r\n", + "offset=%lld,lag=%ld,type=%s", replica_id, replica_ip, replica->repl_data->replica_listening_port, state, replica->repl_data->repl_ack_off, lag, replica->flag.repl_rdb_channel ? "rdb-channel" : replica->repl_data->repl_state == REPLICA_STATE_BG_RDB_LOAD ? "main-channel" : "replica"); + if (replica->repl_data->repl_compressor) { + info = sdscatprintf(info, + ",compression=%s" + ",compressed_bytes=%lld" + ",uncompressed_bytes=%lld" + ",compression_ratio=%.2f" + ",compression_errors=%lld" + ",compression_cpu_usec=%lld", + compressionAlgoName(replCompressorAlgo(replica->repl_data->repl_compressor)), + replica->repl_data->repl_compressed_bytes_total, + replica->repl_data->repl_uncompressed_bytes_total, + replica->repl_data->repl_uncompressed_bytes_total > 0 + ? (double)replica->repl_data->repl_compressed_bytes_total / + (double)replica->repl_data->repl_uncompressed_bytes_total + : 0.0, + atomic_load_explicit(&replica->repl_data->repl_compression_errors, + memory_order_relaxed), + atomic_load_explicit(&replica->repl_data->repl_compression_cpu_usec, + memory_order_relaxed)); + } + info = sdscat(info, "\r\n"); replica_id++; } } diff --git a/src/server.h b/src/server.h index c786b2adbbe..8b28e7ff3c9 100644 --- a/src/server.h +++ b/src/server.h @@ -451,9 +451,11 @@ typedef enum { #define REPLICA_CAPA_PSYNC2 (1 << 1) /* Supports PSYNC2 protocol. */ #define REPLICA_CAPA_DUAL_CHANNEL (1 << 2) /* Supports dual channel replication sync */ #define REPLICA_CAPA_SKIP_RDB_CHECKSUM (1 << 3) /* Supports skipping RDB checksum for sync requests. */ +#define REPLICA_CAPA_COMPRESSION (1 << 4) /* Supports replication compression. */ /* Replica capability strings */ #define REPLICA_CAPA_SKIP_RDB_CHECKSUM_STR "skip-rdb-checksum" /* Supports skipping RDB checksum for sync requests. */ +#define REPLICA_CAPA_COMPRESSION_STR "compression" /* Supports replication compression. */ /* Replica requirements */ #define REPLICA_REQ_NONE 0 @@ -602,6 +604,9 @@ typedef enum { RDB_COMPRESSION_NO = 0, RDB_COMPRESSION_YES, RDB_COMPRESSION_LZ4_STREAM } rdb_compression_mode; +typedef enum { REPL_COMPRESSION_NO = 0, + REPL_COMPRESSION_LZ4_STREAM } repl_compression_mode; + /* Structure representing a non-owning view of a buffer. * A stringRef struct does not manage the underlying memory, so its destruction * will not free the buffer. */ @@ -1218,6 +1223,9 @@ typedef struct ClientPubSubData { context of client side caching. */ } ClientPubSubData; +typedef struct replCompressor replCompressor; +typedef struct replDecompressor replDecompressor; + typedef struct ClientReplicationData { int repl_state; /* Replication state if this is a replica. */ int repl_start_cmd_stream_on_ack; /* Install replica write handler on first ACK. */ @@ -1249,6 +1257,15 @@ typedef struct ClientReplicationData { size_t ref_block_pos; /* Access position of referenced buffer block, i.e. the next offset to send. */ sds replica_nodeid; /* Node id in cluster mode. */ + /* Incremental replication compression state (primary side, per replica). */ + replCompressor *repl_compressor; /* Per-replica replication compressor (NULL if uncompressed). */ + _Atomic(int) compression_error; /* Set by the IO thread on compression failure, + * read by the main thread in postWriteToReplica. */ + /* Compression metrics (primary side, per replica). */ + long long repl_compressed_bytes_total; /* Total compressed bytes sent (main thread only). */ + long long repl_uncompressed_bytes_total; /* Total raw bytes before compression (main thread only). */ + _Atomic(long long) repl_compression_errors; /* Compression failure count. */ + _Atomic(long long) repl_compression_cpu_usec; /* Cumulative CPU time in compression (microseconds). */ } ClientReplicationData; typedef struct ClientModuleData { @@ -2060,6 +2077,10 @@ struct valkeyServer { int saveparamslen; /* Number of saving points */ char *rdb_filename; /* Name of RDB file */ int rdb_compression; /* RDB compression mode */ + int repl_compression; /* Replication compression mode */ + int repl_provisional_compression; /* Replica: compression advertised on the current upstream + * handshake. Gates decompressor setup (not the live config), + * so a mid-handshake config flip cannot desync the link. */ int rdb_checksum; /* Use RDB checksum? */ int rdb_del_sync_files; /* Remove RDB files used only for SYNC if the instance does not use persistence. */ @@ -2172,33 +2193,37 @@ struct valkeyServer { long long read_reploff; int dbid; } repl_provisional_primary; - client *cached_primary; /* Cached primary to be reused for PSYNC. */ - rio *loading_rio; /* Pointer to the rio object currently used for loading data. */ - int repl_syncio_timeout; /* Timeout for synchronous I/O calls */ - int repl_state; /* Replication status if the instance is a replica */ - int repl_rdb_channel_state; /* State of the replica's rdb channel during dual-channel-replication */ - off_t repl_transfer_size; /* Size of RDB to read from primary during sync. */ - off_t repl_transfer_read; /* Amount of RDB read from primary during sync. */ - off_t repl_transfer_last_fsync_off; /* Offset when we fsync-ed last time. */ - connection *repl_transfer_s; /* Replica -> Primary SYNC connection */ - connection *repl_rdb_transfer_s; /* Primary FULL SYNC connection (RDB download) */ - int repl_transfer_fd; /* Replica -> Primary SYNC temp file descriptor */ - char *repl_transfer_tmpfile; /* Replica-> Primary SYNC temp file name */ - _Atomic(time_t) repl_transfer_lastio; /* Unix time of the latest read, for timeout */ - int repl_serve_stale_data; /* Serve stale data when link is down? */ - int repl_replica_ro; /* Replica is read only? */ - int repl_replica_ignore_maxmemory; /* If true replicas do not evict. */ - time_t repl_down_since; /* Unix time at which link with primary went down */ - int repl_disable_tcp_nodelay; /* Disable TCP_NODELAY after SYNC? */ - int repl_mptcp; /* Use Multipath TCP for replica on client side */ - int replica_priority; /* Reported in INFO and used by Sentinel. */ - int replica_announced; /* If true, replica is announced by Sentinel */ - int replica_announce_port; /* Give the primary this listening port. */ - char *replica_announce_ip; /* Give the primary this ip address. */ - int propagation_error_behavior; /* Configures the behavior of the replica - * when it receives an error on the replication stream */ - int repl_ignore_disk_write_error; /* Configures whether replicas panic when unable to - * persist writes to AOF. */ + client *cached_primary; /* Cached primary to be reused for PSYNC. */ + rio *loading_rio; /* Pointer to the rio object currently used for loading data. */ + int repl_syncio_timeout; /* Timeout for synchronous I/O calls */ + int repl_state; /* Replication status if the instance is a replica */ + int repl_rdb_channel_state; /* State of the replica's rdb channel during dual-channel-replication */ + off_t repl_transfer_size; /* Size of RDB to read from primary during sync. */ + off_t repl_transfer_read; /* Amount of RDB read from primary during sync. */ + off_t repl_transfer_last_fsync_off; /* Offset when we fsync-ed last time. */ + connection *repl_transfer_s; /* Replica -> Primary SYNC connection */ + connection *repl_rdb_transfer_s; /* Primary FULL SYNC connection (RDB download) */ + int repl_transfer_fd; /* Replica -> Primary SYNC temp file descriptor */ + char *repl_transfer_tmpfile; /* Replica-> Primary SYNC temp file name */ + _Atomic(time_t) repl_transfer_lastio; /* Unix time of the latest read, for timeout */ + int repl_serve_stale_data; /* Serve stale data when link is down? */ + int repl_replica_ro; /* Replica is read only? */ + int repl_replica_ignore_maxmemory; /* If true replicas do not evict. */ + time_t repl_down_since; /* Unix time at which link with primary went down */ + int repl_disable_tcp_nodelay; /* Disable TCP_NODELAY after SYNC? */ + int repl_mptcp; /* Use Multipath TCP for replica on client side */ + int replica_priority; /* Reported in INFO and used by Sentinel. */ + int replica_announced; /* If true, replica is announced by Sentinel */ + int replica_announce_port; /* Give the primary this listening port. */ + char *replica_announce_ip; /* Give the primary this ip address. */ + int propagation_error_behavior; /* Configures the behavior of the replica + * when it receives an error on the replication stream */ + int repl_ignore_disk_write_error; /* Configures whether replicas panic when unable to + * persist writes to AOF. */ + replDecompressor *repl_decompressor; /* Replica-side replication decoder (NULL when inactive). */ + size_t repl_decompression_errors; /* Decompression failures (replica side). */ + long long repl_decompression_cpu_usec; /* Cumulative CPU time in decompression (microseconds). */ + size_t repl_decompressed_bytes_total; /* Total decompressed bytes processed (replica side). */ /* The following two fields is where we store primary PSYNC replid/offset * while the PSYNC is in progress. At the end we'll copy the fields into @@ -3019,6 +3044,9 @@ int getClientTypeByName(char *name); char *getClientTypeName(int client_class); void flushReplicasOutputBuffers(void); void disconnectReplicas(void); +void reconcileReplicaCompression(void); +void replDestroyCompression(client *c); +int replDecompressQueryBuf(client *c, size_t new_data_start); void evictClients(void); int listenToPort(connListener *fds); void pauseActions(pause_purpose purpose, mstime_t end, uint32_t actions); @@ -3047,6 +3075,9 @@ void protectClient(client *c); void unprotectClient(client *c); void initSharedQueryBuf(void); void freeSharedQueryBuf(void); +#ifndef __cplusplus +extern _Thread_local sds thread_shared_qb; /* Per-thread shared query buffer (see networking.c). */ +#endif client *lookupClientByID(uint64_t id); int authRequired(client *c); void clientSetUser(client *c, user *u, int authenticated); diff --git a/src/unit/test_repl_compression.cpp b/src/unit/test_repl_compression.cpp new file mode 100644 index 00000000000..c737f5eb83d --- /dev/null +++ b/src/unit/test_repl_compression.cpp @@ -0,0 +1,87 @@ +/* + * Copyright (c) Valkey Contributors + * All rights reserved. + * SPDX-License-Identifier: BSD-3-Clause + */ + +/* Unit tests for replication compression configuration and capability constants. */ + +#include "generated_wrappers.hpp" + +extern "C" { +#include "compression.h" +#include "compression_repl.h" +#include "server.h" +} + +/* zmalloc.h defines helper macros that collide with libstdc++ internals. */ +#ifdef __xstr +#undef __xstr +#endif +#ifdef __str +#undef __str +#endif + +TEST(replCompression, capaCompressionBitNoConflict) { + /* Each REPLICA_CAPA_* must occupy a unique bit position. */ + int all_capas[] = { + REPLICA_CAPA_EOF, + REPLICA_CAPA_PSYNC2, + REPLICA_CAPA_DUAL_CHANNEL, + REPLICA_CAPA_SKIP_RDB_CHECKSUM, + REPLICA_CAPA_COMPRESSION, + }; + int count = sizeof(all_capas) / sizeof(all_capas[0]); + + for (int i = 0; i < count; i++) { + /* Each value must be a power of two (single bit set). */ + EXPECT_NE(all_capas[i], 0) << "capability " << i << " must be non-zero"; + EXPECT_EQ(all_capas[i] & (all_capas[i] - 1), 0) + << "capability " << i << " must be a power of two"; + + for (int j = i + 1; j < count; j++) { + EXPECT_EQ(all_capas[i] & all_capas[j], 0) + << "capabilities " << i << " and " << j << " must not share bits"; + } + } + + /* Verify the specific value. */ + EXPECT_EQ(REPLICA_CAPA_COMPRESSION, (1 << 4)); +} + +TEST(replCompression, capaCompressionStr) { + EXPECT_STREQ(REPLICA_CAPA_COMPRESSION_STR, "compression"); +} + +TEST(replCompression, algoConstants) { + /* ALGO_LZ4 must be non-zero so it's distinguishable from zero-init. */ + EXPECT_NE(ALGO_LZ4, 0); +} + +TEST(replCompression, resetBatchRetainsAllocationForIncompressibleBatch) { + replCompressor *rc = replCompressorCreate(ALGO_LZ4, 0); + ASSERT_TRUE(rc != NULL); + /* ~1 MiB of incompressible (xorshift) bytes: compressed payload ~= input, + * exercising the greedy-SDS over-allocation path. Stays at/under the 1 MiB + * batch cap so the payload is within the retention bound. */ + const size_t n = 1000000; + unsigned char *buf = (unsigned char *)zmalloc(n); + uint32_t x = 0x9e3779b9u; + for (size_t i = 0; i < n; i++) { + x ^= x << 13; + x ^= x >> 17; + x ^= x << 5; + buf[i] = (unsigned char)(x >> 24); + } + ASSERT_EQ(replCompressorWrite(rc, buf, n), 0); + ASSERT_EQ(replCompressorFlush(rc), 0); + size_t alloc_before = sdsalloc(rc->out_buf); + size_t len_before = sdslen(rc->out_buf); + EXPECT_GT(len_before, (size_t)(900 * 1024)); /* ratio ~1: payload is large */ + EXPECT_GT(alloc_before, (size_t)(1024 * 1024)); /* greedy SDS over-allocates past 1 MiB */ + replCompressorResetBatch(rc); + EXPECT_EQ(sdslen(rc->out_buf), (size_t)0); /* cleared */ + EXPECT_EQ(sdsalloc(rc->out_buf), alloc_before); /* retained, not freed to empty */ + zfree(buf); + replCompressorDestroy(rc); +} diff --git a/tests/integration/block-repl.tcl b/tests/integration/block-repl.tcl index 52b4a53ead8..4358d11db55 100644 --- a/tests/integration/block-repl.tcl +++ b/tests/integration/block-repl.tcl @@ -11,7 +11,7 @@ proc stop_bg_block_op {handle} { catch {exec /bin/kill -9 $handle} } -start_server {tags {"repl" "external:skip"}} { +start_server {tags {"repl" "external:skip" repl-compression}} { start_server {overrides {save {}}} { set master [srv -1 client] set master_host [srv -1 host] diff --git a/tests/integration/dual-channel-replication.tcl b/tests/integration/dual-channel-replication.tcl index 8f5fd56c5b8..1abdddf46ee 100644 --- a/tests/integration/dual-channel-replication.tcl +++ b/tests/integration/dual-channel-replication.tcl @@ -22,7 +22,7 @@ proc wait_and_resume_process idx { resume_process $pid } -start_server {tags {"dual-channel-replication external:skip"}} { +start_server {tags {"dual-channel-replication external:skip" repl-compression}} { set replica [srv 0 client] set replica_host [srv 0 host] set replica_port [srv 0 port] @@ -88,7 +88,7 @@ start_server {tags {"dual-channel-replication external:skip"}} { } } -start_server {tags {"dual-channel-replication external:skip"}} { +start_server {tags {"dual-channel-replication external:skip" repl-compression}} { set replica [srv 0 client] set replica_host [srv 0 host] set replica_port [srv 0 port] @@ -122,7 +122,7 @@ start_server {tags {"dual-channel-replication external:skip"}} { } } -start_server {tags {"dual-channel-replication external:skip"}} { +start_server {tags {"dual-channel-replication external:skip" repl-compression}} { set replica [srv 0 client] set replica_host [srv 0 host] set replica_port [srv 0 port] @@ -210,7 +210,7 @@ start_server {tags {"dual-channel-replication external:skip"}} { } } -start_server {tags {"dual-channel-replication external:skip"}} { +start_server {tags {"dual-channel-replication external:skip" repl-compression}} { set replica [srv 0 client] set replica_host [srv 0 host] set replica_port [srv 0 port] @@ -345,7 +345,7 @@ start_server {tags {"dual-channel-replication external:skip"}} { } } -start_server {tags {"dual-channel-replication external:skip"}} { +start_server {tags {"dual-channel-replication external:skip" repl-compression}} { set replica1 [srv 0 client] set replica1_host [srv 0 host] set replica1_port [srv 0 port] @@ -479,7 +479,7 @@ start_server {tags {"dual-channel-replication external:skip"}} { } } -start_server {tags {"dual-channel-replication external:skip"}} { +start_server {tags {"dual-channel-replication external:skip" repl-compression}} { set replica [srv 0 client] set replica_host [srv 0 host] set replica_port [srv 0 port] @@ -516,7 +516,7 @@ start_server {tags {"dual-channel-replication external:skip"}} { } } -start_server {tags {"dual-channel-replication external:skip"}} { +start_server {tags {"dual-channel-replication external:skip" repl-compression}} { set replica [srv 0 client] set replica_host [srv 0 host] set replica_port [srv 0 port] @@ -581,7 +581,7 @@ start_server {tags {"dual-channel-replication external:skip"}} { } } -start_server {tags {"dual-channel-replication external:skip"}} { +start_server {tags {"dual-channel-replication external:skip" repl-compression}} { set replica1 [srv 0 client] set replica1_host [srv 0 host] set replica1_port [srv 0 port] @@ -655,7 +655,7 @@ start_server {tags {"dual-channel-replication external:skip"}} { } } -start_server {tags {"dual-channel-replication external:skip"}} { +start_server {tags {"dual-channel-replication external:skip" repl-compression}} { set primary [srv 0 client] set primary_host [srv 0 host] set primary_port [srv 0 port] @@ -717,7 +717,7 @@ start_server {tags {"dual-channel-replication external:skip"}} { } } -start_server {tags {"dual-channel-replication external:skip"}} { +start_server {tags {"dual-channel-replication external:skip" repl-compression}} { set primary [srv 0 client] set primary_host [srv 0 host] set primary_port [srv 0 port] @@ -778,7 +778,7 @@ start_server {tags {"dual-channel-replication external:skip"}} { } } -start_server {tags {"dual-channel-replication external:skip"}} { +start_server {tags {"dual-channel-replication external:skip" repl-compression}} { set primary [srv 0 client] set primary_host [srv 0 host] set primary_port [srv 0 port] @@ -858,7 +858,7 @@ start_server {tags {"dual-channel-replication external:skip"}} { $primary config set shutdown-timeout 0 } -start_server {tags {"dual-channel-replication external:skip"}} { +start_server {tags {"dual-channel-replication external:skip" repl-compression}} { set primary [srv 0 client] set primary_host [srv 0 host] set primary_port [srv 0 port] @@ -919,7 +919,7 @@ start_server {tags {"dual-channel-replication external:skip"}} { } } -start_server {tags {"dual-channel-replication external:skip"}} { +start_server {tags {"dual-channel-replication external:skip" repl-compression}} { set primary [srv 0 client] set primary_host [srv 0 host] set primary_port [srv 0 port] @@ -1009,7 +1009,7 @@ start_server {tags {"dual-channel-replication external:skip"}} { } -start_server {tags {"dual-channel-replication external:skip"}} { +start_server {tags {"dual-channel-replication external:skip" repl-compression}} { set primary [srv 0 client] set primary_host [srv 0 host] set primary_port [srv 0 port] @@ -1134,7 +1134,7 @@ start_server {tags {"dual-channel-replication external:skip"}} { } } -start_server {tags {"dual-channel-replication external:skip"}} { +start_server {tags {"dual-channel-replication external:skip" repl-compression}} { set primary [srv 0 client] set primary_host [srv 0 host] set primary_port [srv 0 port] @@ -1185,7 +1185,7 @@ start_server {tags {"dual-channel-replication external:skip"}} { } } -start_server {tags {"dual-channel-replication external:skip"}} { +start_server {tags {"dual-channel-replication external:skip" repl-compression}} { set primary [srv 0 client] set primary_host [srv 0 host] set primary_port [srv 0 port] @@ -1249,7 +1249,7 @@ start_server {tags {"dual-channel-replication external:skip"}} { } } -start_server {tags {"dual-channel-replication external:skip"}} { +start_server {tags {"dual-channel-replication external:skip" repl-compression}} { set primary [srv 0 client] set primary_host [srv 0 host] set primary_port [srv 0 port] diff --git a/tests/integration/repl-compression.tcl b/tests/integration/repl-compression.tcl new file mode 100644 index 00000000000..06458432432 --- /dev/null +++ b/tests/integration/repl-compression.tcl @@ -0,0 +1,1001 @@ +tags {"repl repl-compression external:skip"} { + +# ============================================================ +# Config CRUD — single-server tests, no replication needed +# ============================================================ + +start_server {overrides {save "" repl-compression no}} { + + test {Repl compression config defaults are correct} { + assert_equal "no" [lindex [r config get repl-compression] 1] + } + + test {repl-compression can be toggled on and off} { + r config set repl-compression lz4-stream + assert_equal "lz4-stream" [lindex [r config get repl-compression] 1] + r config set repl-compression no + assert_equal "no" [lindex [r config get repl-compression] 1] + } + + test {Repl compression configs survive CONFIG REWRITE and restart} { + r config set repl-compression lz4-stream + r config rewrite + + restart_server 0 true false + + assert_equal "lz4-stream" [lindex [r config get repl-compression] 1] + + # Restore default + r config set repl-compression no + } +} + +# ============================================================ +# Replication handshake behavior — primary + replica tests +# ============================================================ + +start_server {tags {"repl"} overrides {save ""}} { + set primary [srv 0 client] + set primary_host [srv 0 host] + set primary_port [srv 0 port] + + test {Replica with repl-compression no does NOT send capa compression} { + start_server {overrides {save "" repl-compression 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 repl-compression lz4-stream and diskless load sends capa compression} { + start_server {overrides {save "" repl-compression lz4-stream 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 repl-compression lz4-stream and disk-backed load also negotiates compression} { + $primary config set repl-compression lz4-stream + set _code [catch { + start_server {overrides {save "" repl-compression lz4-stream 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" + } + + # Disk-backed full sync completes, then compression activates for the + # post-sync incremental stream (the full-sync RDB itself is never + # compressed by this capability). + wait_for_condition 50 100 { + [regexp -all "compression=lz4" [$primary info replication]] >= 1 + } else { + fail "Compression not negotiated for disk-backed replica" + } + + # Exercise the compressed incremental stream over a disk-backed link. + for {set i 0} {$i < 100} {incr i} { + $primary set "diskbacked:$i" [string repeat "v" 50] + } + wait_for_condition 50 100 { + [$replica get "diskbacked:99"] eq [string repeat "v" 50] + } else { + fail "Disk-backed replica did not receive compressed incremental stream" + } + assert_equal [$primary debug digest] [$replica debug digest] + + $replica replicaof no one + } + } _res _opts] + $primary config set repl-compression no + return -options $_opts $_res + } + + test {Primary receiving capa compression still completes full sync correctly (no-op)} { + $primary flushall + for {set i 0} {$i < 100} {incr i} { + $primary set "noop:$i" [string repeat "value$i " 10] + } + + start_server {overrides {save "" repl-compression lz4-stream repl-diskless-load swapdb}} { + set replica [srv 0 client] + $replica replicaof $primary_host $primary_port + wait_for_sync $replica + + wait_for_condition 50 100 { + [status $replica master_link_status] eq "up" + } else { + fail "Replica did not complete full sync" + } + + # Data integrity check — primary records capa but takes no action + assert_equal [string repeat "value42 " 10] [$replica get noop:42] + assert_equal 100 [$replica dbsize] + + $replica replicaof no one + } + } + + test {Backward compatibility - older replica without capa compression connects successfully} { + start_server {overrides {save ""}} { + 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" + } + + assert_equal {up} [s 0 master_link_status] + + $replica replicaof no one + } + } + + test {Toggling repl-compression mid-runtime affects the next handshake} { + # First sync with compression disabled + start_server {overrides {save "" repl-compression no 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 with repl-compression no" + } + + assert_equal {up} [s 0 master_link_status] + + # Toggle compression on at runtime + $replica config set repl-compression lz4-stream + + # Disconnect and reconnect to trigger a new handshake + $replica replicaof no one + $replica replicaof $primary_host $primary_port + + wait_for_condition 50 100 { + [s 0 master_link_status] eq {up} + } else { + fail "Replication not started after toggling repl-compression lz4-stream" + } + + assert_equal {up} [s 0 master_link_status] + + $replica replicaof no one + } + } + + test {Compressed incremental replication delivers correct data} { + $primary config set repl-compression lz4-stream + $primary flushall + + start_server {overrides {save "" repl-compression lz4-stream 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" + } + + # Write data on primary AFTER full sync (goes through incremental stream) + for {set i 0} {$i < 50} {incr i} { + $primary set "compressed:$i" [string repeat "payload$i " 20] + } + + # Wait for replica to catch up + wait_for_condition 50 100 { + [$replica dbsize] == [$primary dbsize] + } else { + fail "Replica did not catch up: replica=[$replica dbsize] primary=[$primary dbsize]" + } + + # Verify data integrity + for {set i 0} {$i < 50} {incr i} { + assert_equal [string repeat "payload$i " 20] [$replica get "compressed:$i"] + } + + $replica replicaof no one + } + + $primary config set repl-compression no + } + + test {Compressed incremental replication handles values larger than the batch limit} { + # A value past REPL_COMPRESSION_BATCH_LIMIT (1 MB) is compressed across + # multiple dispatches; verify it round-trips intact. + $primary config set repl-compression lz4-stream + $primary flushall + + start_server {overrides {save "" repl-compression lz4-stream 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" + } + + # ~4 MB value (well past the 1 MB batch limit) written after full sync. + set bigval [string repeat "abcdefghij0123456789" 209715] + $primary set bigkey $bigval + + wait_for_condition 50 200 { + [$replica get bigkey] eq $bigval + } else { + fail "Large value did not replicate intact under compression" + } + assert_equal [string length $bigval] [string length [$replica get bigkey]] + + $replica replicaof no one + } + + $primary config set repl-compression no + } + + test {Compressed incremental replication handles incompressible values larger than the batch limit} { + # A pseudo-random (incompressible) value past REPL_COMPRESSION_BATCH_LIMIT + # (1 MB) exercises the codec's ratio~1 expansion path across multiple + # dispatches; verify it round-trips intact with no compression errors. + $primary config set repl-compression lz4-stream + $primary flushall + + start_server {overrides {save "" repl-compression lz4-stream 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" + } + + # Deterministic pseudo-random binary payload >= 1.5 MiB, built in + # 4096-byte chunks from a seeded PRNG so failures are reproducible. + expr {srand(424242)} + set payload "" + while {[string length $payload] < 1572864} { + set chunk "" + for {set i 0} {$i < 4096} {incr i} { + append chunk [format %c [expr {int(rand()*256)}]] + } + append payload $chunk + } + $primary set incompressible_key $payload + + wait_for_condition 50 200 { + [$replica get incompressible_key] eq $payload + } else { + fail "Incompressible value did not replicate intact under compression" + } + assert_equal {up} [s 0 master_link_status] + + set info [$primary info replication] + assert_match "*compression=lz4*" $info + assert {[regexp {compression_errors=([0-9]+)} $info -> comp_errors]} + assert_equal 0 $comp_errors + + $replica replicaof no one + } + + $primary config set repl-compression no + } + + test {Partial resync with compression delivers correct data} { + $primary config set repl-compression lz4-stream + $primary flushall + + start_server {overrides {save "" repl-compression lz4-stream 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" + } + + # Write initial data + $primary set key1 value1 + + wait_for_condition 50 100 { + [$replica get key1] eq {value1} + } else { + fail "Initial replication failed" + } + + # Break the replication link from the primary side. The replica keeps + # its cached primary + offset and auto-reconnects, which exercises the + # compressed *partial* resync path (REPLICAOF NO ONE would force a full + # resync instead). + set partial_before [status $primary sync_partial_ok] + $primary client kill type replica + + wait_for_condition 50 100 { + [s 0 master_link_status] eq {up} + } else { + fail "Reconnection failed" + } + + # Confirm a partial resync actually happened (not a silent full sync) + wait_for_condition 50 100 { + [status $primary sync_partial_ok] > $partial_before + } else { + fail "Expected a partial resync after reconnect, but none occurred" + } + + # Write more data after partial resync + $primary set key2 value2 + $primary set key3 [string repeat "x" 1000] + + wait_for_condition 50 100 { + [$replica get key3] eq [string repeat "x" 1000] + } else { + fail "Post-partial-resync replication failed" + } + + assert_equal value1 [$replica get key1] + assert_equal value2 [$replica get key2] + + $replica replicaof no one + } + + $primary config set repl-compression no + } + + test {Replica with repl-compression lz4-stream handles a plaintext primary (passthrough)} { + # Primary has compression OFF, replica ON: the replica advertises the + # capability but the primary sends plaintext, so the replica must pass + # the stream through untouched rather than expecting a VCS envelope. + $primary config set repl-compression no + $primary flushall + + start_server {overrides {save "" repl-compression lz4-stream 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 (primary plaintext, replica compression on)" + } + + # Incremental writes arrive as plaintext; passthrough must deliver them. + for {set i 0} {$i < 50} {incr i} { + $primary set "pt:$i" [string repeat "payload$i " 20] + } + wait_for_condition 50 100 { + [$replica get pt:49] eq [string repeat "payload49 " 20] + } else { + fail "Replica did not receive plaintext data via passthrough" + } + assert_equal [$primary dbsize] [$replica dbsize] + + # Primary never compressed (its config is off). + assert_equal 0 [string match {*compression=lz4*} [$primary info replication]] + + $replica replicaof no one + } + } + + test {Replica repl-compression flip renegotiates upstream without manual reconnect} { + $primary config set repl-compression lz4-stream + + start_server {overrides {save "" repl-compression lz4-stream repl-diskless-load swapdb}} { + set replica [srv 0 client] + $replica replicaof $primary_host $primary_port + + wait_for_condition 50 200 { + [s 0 master_link_status] eq {up} && + [string match {*compression=lz4*} [$primary info replication]] + } else { + fail "Compressed replication not established" + } + + # Flipping compression OFF on the replica must, on its own, drop and + # reconnect the upstream link so it re-advertises capa without + # compression. No manual replicaof is issued. + $replica config set repl-compression no + + wait_for_condition 50 200 { + [s 0 master_link_status] eq {up} && + ![string match {*compression=lz4*} [$primary info replication]] + } else { + fail "Replica did not renegotiate to plaintext after flip" + } + + # Data still flows after the renegotiation. + $primary set flipkey flipval + wait_for_condition 50 100 { + [$replica get flipkey] eq {flipval} + } else { + fail "Data not replicated after replica-side flip" + } + + $replica replicaof no one + } + } + + test {CONFIG SET repl-compression no disconnects compressed replicas} { + $primary config set repl-compression lz4-stream + + start_server {overrides {save "" repl-compression lz4-stream 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" + } + + # Wait for compression to be active (replica must be state=online) + wait_for_condition 50 200 { + [string match {*state=online*compression=lz4*} [$primary info replication]] + } else { + fail "Compression not active on replica" + } + + # Disable compression on primary — should disconnect compressed replicas + $primary config set repl-compression no + + # Replica should disconnect and reconnect + wait_for_condition 50 200 { + [s 0 master_link_status] eq {up} + } else { + fail "Replica did not reconnect after repl-compression disabled" + } + + # After reconnect, compression should NOT be active + set info [$primary info replication] + if {[string match "*compression=lz4*" $info]} { + fail "Compression still active after disable" + } + + $replica replicaof no one + } + } + + test {Multiple compressed replicas receive replication correctly} { + # Verifies that multiple compressed replicas can connect to the same + # primary and all receive replication data. With IO threads enabled, + # writes are distributed across the shared inbox. + $primary config set repl-compression lz4-stream + + start_server {overrides {save "" repl-compression lz4-stream repl-diskless-load swapdb}} { + set replica1 [srv 0 client] + $replica1 replicaof $primary_host $primary_port + + wait_for_condition 50 100 { + [s 0 master_link_status] eq {up} + } else { + fail "Replica 1 not started" + } + + start_server {overrides {save "" repl-compression lz4-stream repl-diskless-load swapdb}} { + set replica2 [srv 0 client] + $replica2 replicaof $primary_host $primary_port + + wait_for_condition 50 100 { + [s 0 master_link_status] eq {up} + } else { + fail "Replica 2 not started" + } + + # Write data and verify both replicas receive it + for {set i 0} {$i < 30} {incr i} { + $primary set "multi_repl:$i" "value_$i" + } + + wait_for_condition 50 100 { + [$replica1 get "multi_repl:29"] eq {value_29} && + [$replica2 get "multi_repl:29"] eq {value_29} + } else { + fail "Not all replicas caught up" + } + + # Verify both have compression active + set info [$primary info replication] + set matches [regexp -all "compression=lz4" $info] + assert {$matches >= 2} + + $replica2 replicaof no one + } + $replica1 replicaof no one + } + $primary config set repl-compression no + } + + # ============================================================ + # Comprehensive data-type coverage with compression enabled + # ============================================================ + + if {[lsearch $::denytags "repl-compression-suite"] == -1} { + + $primary config set repl-compression lz4-stream + $primary flushall + + start_server {overrides {save "" repl-compression lz4-stream 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" + } + + # Wait for replica to be fully online with compression active + wait_for_condition 50 200 { + [string match {*state=online*compression=lz4*} [$primary info replication]] + } else { + fail "Compression not active on replica" + } + + test {Compressed replication handles strings correctly} { + $primary set str_key [string repeat "hello world " 100] + wait_for_condition 50 100 { + [$replica get str_key] eq [string repeat "hello world " 100] + } else { + fail "String replication failed" + } + } + + test {Compressed replication handles lists correctly} { + for {set i 0} {$i < 100} {incr i} { + $primary rpush mylist "element_$i" + } + wait_for_condition 50 100 { + [$replica llen mylist] == 100 + } else { + fail "List replication failed: got [$replica llen mylist]" + } + assert_equal "element_0" [$replica lindex mylist 0] + assert_equal "element_99" [$replica lindex mylist 99] + } + + test {Compressed replication handles hashes correctly} { + for {set i 0} {$i < 50} {incr i} { + $primary hset myhash field_$i [string repeat "value$i" 20] + } + wait_for_condition 50 100 { + [$replica hlen myhash] == 50 + } else { + fail "Hash replication failed" + } + assert_equal [string repeat "value25" 20] [$replica hget myhash field_25] + } + + test {Compressed replication handles sets correctly} { + for {set i 0} {$i < 100} {incr i} { + $primary sadd myset "member_$i" + } + wait_for_condition 50 100 { + [$replica scard myset] == 100 + } else { + fail "Set replication failed" + } + assert_equal 1 [$replica sismember myset "member_50"] + } + + test {Compressed replication handles sorted sets correctly} { + for {set i 0} {$i < 100} {incr i} { + $primary zadd myzset $i "member_$i" + } + wait_for_condition 50 100 { + [$replica zcard myzset] == 100 + } else { + fail "Sorted set replication failed" + } + assert_equal "member_0" [lindex [$replica zrange myzset 0 0] 0] + } + + test {Compressed replication handles large pipeline correctly} { + for {set i 0} {$i < 1000} {incr i} { + $primary set "pipeline:$i" [string repeat "x" 100] + } + wait_for_condition 50 200 { + [$replica get "pipeline:999"] eq [string repeat "x" 100] + } else { + fail "Pipeline replication failed" + } + assert_equal [string repeat "x" 100] [$replica get "pipeline:500"] + } + + test {Compressed replication handles MULTI/EXEC correctly} { + $primary multi + $primary set tx_key1 tx_val1 + $primary set tx_key2 tx_val2 + $primary incr tx_counter + $primary exec + wait_for_condition 50 100 { + [$replica get tx_key2] eq {tx_val2} + } else { + fail "Transaction replication failed" + } + assert_equal "1" [$replica get tx_counter] + } + + test {Compressed replication handles DEL and expiry correctly} { + $primary set expire_key "will_expire" + $primary pexpire expire_key 100 + $primary set del_key "will_delete" + $primary del del_key + after 200 + wait_for_condition 50 100 { + [$replica exists expire_key] == 0 + } else { + fail "Expiry replication failed" + } + assert_equal 0 [$replica exists del_key] + } + + $replica replicaof no one + } + + test {Compression handles io-threads change gracefully} { + start_server {overrides {save "" repl-compression lz4-stream 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" + } + + # Write some data + $primary set io_test_key "io_test_value" + wait_for_condition 50 100 { + [$replica get io_test_key] eq {io_test_value} + } else { + fail "Initial replication failed" + } + + # Verify data still replicates after more writes + for {set i 0} {$i < 20} {incr i} { + $primary set "after_io_change:$i" "value_$i" + } + wait_for_condition 50 100 { + [$replica get "after_io_change:19"] eq {value_19} + } else { + fail "Replication after io-threads context failed" + } + + $replica replicaof no one + } + } + + $primary config set repl-compression no + + } ;# end repl-compression-suite +} + +# ============================================================ +# Multi-replica compressed replication tests +# ============================================================ + +# Disabling repl-compression at runtime disconnects compressed replicas. The +# disconnect is deferred until after CONFIG SET commits (so a rolled-back +# multi-option CONFIG SET drops nothing), then the replica reconnects plaintext. +start_server {tags {"repl"} overrides {save "" repl-compression lz4-stream}} { + set primary [srv 0 client] + set primary_host [srv 0 host] + set primary_port [srv 0 port] + + test {Disabling repl-compression disconnects compressed replicas} { + start_server {overrides {save "" repl-compression lz4-stream repl-diskless-load swapdb}} { + set replica [srv 0 client] + $replica replicaof $primary_host $primary_port + wait_for_sync $replica + + wait_for_condition 50 100 { + [regexp -all "compression=lz4" [$primary info replication]] >= 1 + } else { + fail "Compression not active before disable" + } + + $primary config set repl-compression no + + # The compressed replica is dropped, then reconnects as plaintext, so + # the link no longer reports compression=lz4. + wait_for_condition 50 100 { + [regexp -all "compression=lz4" [$primary info replication]] == 0 + } else { + fail "Compressed replica was not disconnected after repl-compression no" + } + wait_for_condition 50 100 { + [status $replica master_link_status] eq "up" + } else { + fail "Replica did not reconnect after compression disabled" + } + + $replica replicaof no one + } + $primary config set repl-compression lz4-stream + } +} + +# Enabling repl-compression at runtime disconnects capable-but-plaintext replicas +# so they reconnect compressed (symmetric with the disable case; deferred to +# CONFIG SET commit so a rolled-back command reconnects nothing). +start_server {tags {"repl"} overrides {save "" repl-compression no}} { + set primary [srv 0 client] + set primary_host [srv 0 host] + set primary_port [srv 0 port] + + test {Enabling repl-compression reconnects capable replicas compressed} { + start_server {overrides {save "" repl-compression lz4-stream repl-diskless-load swapdb}} { + set replica [srv 0 client] + $replica replicaof $primary_host $primary_port + wait_for_sync $replica + + # Primary compression is off, so although the replica advertised the + # capability the negotiated link is plaintext. + wait_for_condition 50 100 { + [status $replica master_link_status] eq "up" + } else { + fail "Replica did not sync" + } + assert_equal 0 [regexp -all "compression=lz4" [$primary info replication]] + + # Enable on the primary: the capable replica is dropped and reconnects + # over a compressed stream. + $primary config set repl-compression lz4-stream + wait_for_condition 50 100 { + [regexp -all "compression=lz4" [$primary info replication]] >= 1 + } else { + fail "Capable replica did not reconnect compressed after enable" + } + + # Data flows correctly over the new compressed link. + $primary set enabled_key enabled_val + wait_for_condition 50 100 { + [$replica get enabled_key] eq "enabled_val" + } else { + fail "Compressed replication did not deliver after enable" + } + assert_equal [$primary debug digest] [$replica debug digest] + + $replica replicaof no one + } + } +} + +# Test 4: Multiple replicas distribute across threads and stay in sync +start_server {tags {"repl"} overrides {save "" io-threads 4 io-threads-always-active yes repl-compression lz4-stream}} { + set primary [srv 0 client] + set primary_host [srv 0 host] + set primary_port [srv 0 port] + + test {Multiple replicas all stay in sync under load} { + start_server {overrides {save "" repl-compression lz4-stream repl-diskless-load swapdb}} { + set replica1 [srv 0 client] + $replica1 replicaof $primary_host $primary_port + wait_for_sync $replica1 + + start_server {overrides {save "" repl-compression lz4-stream repl-diskless-load swapdb}} { + set replica2 [srv 0 client] + $replica2 replicaof $primary_host $primary_port + wait_for_sync $replica2 + + start_server {overrides {save "" repl-compression lz4-stream repl-diskless-load swapdb}} { + set replica3 [srv 0 client] + $replica3 replicaof $primary_host $primary_port + wait_for_sync $replica3 + + # Wait for all replicas to have compression active + wait_for_condition 50 200 { + [regexp -all "compression=lz4" [$primary info replication]] >= 3 + } else { + fail "Not all replicas have compression active" + } + + # Generate sustained load + for {set i 0} {$i < 500} {incr i} { + $primary set "multi_repl:$i" [string repeat "x" 100] + } + + # Wait for all replicas to catch up + wait_for_condition 100 200 { + [$replica1 dbsize] == [$primary dbsize] && + [$replica2 dbsize] == [$primary dbsize] && + [$replica3 dbsize] == [$primary dbsize] + } else { + fail "Not all replicas caught up: r1=[$replica1 dbsize] r2=[$replica2 dbsize] r3=[$replica3 dbsize] primary=[$primary dbsize]" + } + + # Verify data integrity + set primary_digest [$primary debug digest] + assert_equal $primary_digest [$replica1 debug digest] + assert_equal $primary_digest [$replica2 debug digest] + assert_equal $primary_digest [$replica3 debug digest] + + $replica3 replicaof no one + } + $replica2 replicaof no one + } + $replica1 replicaof no one + } + } +} + +# Test 5: Compressed replication survives replica disconnect/reconnect +start_server {tags {"repl"} overrides {save "" io-threads 4 io-threads-always-active yes repl-compression lz4-stream}} { + set primary [srv 0 client] + set primary_host [srv 0 host] + set primary_port [srv 0 port] + + test {Compressed replication survives replica disconnect and reconnect} { + start_server {overrides {save "" repl-compression lz4-stream repl-diskless-load swapdb}} { + set replica1 [srv 0 client] + $replica1 replicaof $primary_host $primary_port + wait_for_sync $replica1 + + start_server {overrides {save "" repl-compression lz4-stream repl-diskless-load swapdb}} { + set replica2 [srv 0 client] + $replica2 replicaof $primary_host $primary_port + wait_for_sync $replica2 + + # Drive traffic, verify both in sync + for {set i 0} {$i < 200} {incr i} { + $primary set "pre_disconnect:$i" [string repeat "x" 50] + } + + wait_for_condition 50 200 { + [$replica1 dbsize] == [$primary dbsize] && + [$replica2 dbsize] == [$primary dbsize] + } else { + fail "Replicas not in sync before disconnect" + } + + # Disconnect replica1 + $replica1 replicaof no one + + # Drive more traffic — replica2 should stay connected and in sync + for {set i 0} {$i < 200} {incr i} { + $primary set "post_disconnect:$i" [string repeat "x" 50] + } + + wait_for_condition 50 200 { + [$replica2 get "post_disconnect:199"] eq [string repeat "x" 50] + } else { + fail "Replica2 did not stay in sync after replica1 disconnect" + } + + # Reconnect replica1 + $replica1 replicaof $primary_host $primary_port + + wait_for_condition 50 200 { + [status $replica1 master_link_status] eq "up" + } else { + fail "Replica1 did not reconnect" + } + + # Wait for replica1 to catch up + wait_for_condition 50 200 { + [$replica1 dbsize] == [$primary dbsize] + } else { + fail "Replica1 did not re-sync: replica1=[$replica1 dbsize] primary=[$primary dbsize]" + } + + # Verify compression is re-negotiated on replica1 + wait_for_condition 50 200 { + [regexp -all "compression=lz4" [$primary info replication]] >= 2 + } else { + fail "Compression not re-negotiated after reconnect" + } + + # Verify data integrity + assert_equal [$primary debug digest] [$replica1 debug digest] + assert_equal [$primary debug digest] [$replica2 debug digest] + + $replica2 replicaof no one + } + $replica1 replicaof no one + } + } +} + +# Chained replication: each hop negotiates compression independently, and the +# middle node simultaneously decodes its primary link on the main thread while +# encoding for its own replica on IO threads. +start_server {tags {"repl"} overrides {save "" repl-compression lz4-stream}} { + set primary [srv 0 client] + set primary_host [srv 0 host] + set primary_port [srv 0 port] + + test {Chained replication compresses each hop independently} { + start_server {overrides {save "" repl-compression lz4-stream repl-diskless-load swapdb io-threads 4 io-threads-always-active yes}} { + set middle [srv 0 client] + set middle_host [srv 0 host] + set middle_port [srv 0 port] + + start_server {overrides {save "" repl-compression lz4-stream repl-diskless-load swapdb}} { + set leaf [srv 0 client] + + $middle replicaof $primary_host $primary_port + wait_for_sync $middle + $leaf replicaof $middle_host $middle_port + wait_for_sync $leaf + + # Both hops negotiated compression. + wait_for_condition 50 200 { + [regexp -all "compression=lz4" [$primary info replication]] >= 1 && + [regexp -all "compression=lz4" [$middle info replication]] >= 1 + } else { + fail "Compression not active on both hops" + } + + # Writes flow primary -> middle -> leaf across two compressed hops. + for {set i 0} {$i < 200} {incr i} { + $primary set "chain:$i" "chain_value_$i" + } + wait_for_condition 50 200 { + [$leaf dbsize] == [$primary dbsize] + } else { + fail "Leaf did not catch up: leaf=[$leaf dbsize] primary=[$primary dbsize]" + } + assert_equal "chain_value_0" [$leaf get chain:0] + assert_equal "chain_value_99" [$leaf get chain:99] + assert_equal "chain_value_199" [$leaf get chain:199] + + # Flip compression off on the leaf only and force a clean + # reconnect: hop2 renegotiates plaintext, hop1 stays compressed. + $leaf config set repl-compression no + $leaf replicaof no one + $leaf replicaof $middle_host $middle_port + wait_for_sync $leaf + + wait_for_condition 50 200 { + [regexp -all "compression=lz4" [$middle info replication]] == 0 + } else { + fail "Hop2 still compressed after leaf disabled repl-compression" + } + assert {[regexp -all "compression=lz4" [$primary info replication]] >= 1} + + # Data still flows end-to-end over mixed hops. + $primary set chain:final final_val + wait_for_condition 50 200 { + [$leaf get chain:final] eq {final_val} + } else { + fail "Write did not reach leaf after hop2 renegotiated plaintext" + } + + $leaf replicaof no one + } + $middle replicaof no one + } + } +} + + +} diff --git a/tests/integration/replication-2.tcl b/tests/integration/replication-2.tcl index c18ff24fc43..6215349da50 100644 --- a/tests/integration/replication-2.tcl +++ b/tests/integration/replication-2.tcl @@ -1,4 +1,4 @@ -start_server {tags {"repl external:skip"}} { +start_server {tags {"repl external:skip" repl-compression}} { start_server {} { test {First server should have role slave after SLAVEOF} { r -1 slaveof [srv 0 host] [srv 0 port] diff --git a/tests/integration/replication-3.tcl b/tests/integration/replication-3.tcl index f53a05abce1..e1bcb3914f0 100644 --- a/tests/integration/replication-3.tcl +++ b/tests/integration/replication-3.tcl @@ -1,4 +1,4 @@ -start_server {tags {"repl external:skip"}} { +start_server {tags {"repl external:skip" repl-compression}} { start_server {} { test {First server should have role slave after SLAVEOF} { r -1 slaveof [srv 0 host] [srv 0 port] diff --git a/tests/integration/replication-4.tcl b/tests/integration/replication-4.tcl index d5b55b3b570..76880d2a42a 100644 --- a/tests/integration/replication-4.tcl +++ b/tests/integration/replication-4.tcl @@ -1,4 +1,4 @@ -start_server {tags {"repl network external:skip singledb:skip"} overrides {save {}}} { +start_server {tags {"repl network external:skip singledb:skip" repl-compression} overrides {save {}}} { start_server { overrides {save {}}} { set master [srv -1 client] @@ -52,7 +52,7 @@ start_server {tags {"repl network external:skip singledb:skip"} overrides {save } } -start_server {tags {"repl external:skip"}} { +start_server {tags {"repl external:skip" repl-compression}} { start_server {} { set master [srv -1 client] set master_host [srv -1 host] @@ -118,7 +118,7 @@ start_server {tags {"repl external:skip"}} { } } -start_server {tags {"repl external:skip"}} { +start_server {tags {"repl external:skip" repl-compression}} { start_server {} { set master [srv -1 client] set master_host [srv -1 host] @@ -168,7 +168,7 @@ start_server {tags {"repl external:skip"}} { } } -start_server {tags {"repl external:skip"}} { +start_server {tags {"repl external:skip" repl-compression}} { start_server {} { set master [srv -1 client] set master_host [srv -1 host] @@ -249,7 +249,7 @@ start_server {tags {"repl external:skip"}} { } } -start_server {tags {"repl external:skip"}} { +start_server {tags {"repl external:skip" repl-compression}} { start_server {} { set master [srv -1 client] set master_host [srv -1 host] diff --git a/tests/integration/replication-aof-sync.tcl b/tests/integration/replication-aof-sync.tcl index 14113bd0708..893fa976aa6 100644 --- a/tests/integration/replication-aof-sync.tcl +++ b/tests/integration/replication-aof-sync.tcl @@ -17,7 +17,7 @@ proc get_aof_manifest_path {r} { return [file join $dir $appenddirname $appendfilename$::manifest_suffix] } -tags {"repl external:skip"} { +tags {"repl external:skip" repl-compression} { # Test 1: Disk-based full sync with aof-use-rdb-preamble yes should # reuse the RDB file as AOF base file diff --git a/tests/integration/replication.tcl b/tests/integration/replication.tcl index 235a43652ca..eb2a0912b3c 100644 --- a/tests/integration/replication.tcl +++ b/tests/integration/replication.tcl @@ -5,7 +5,7 @@ proc log_file_matches {log pattern} { string match $pattern $content } -start_server {tags {"repl network external:skip"}} { +start_server {tags {"repl network external:skip" repl-compression}} { set slave [srv 0 client] set slave_host [srv 0 host] set slave_port [srv 0 port] @@ -59,7 +59,7 @@ start_server {tags {"repl network external:skip"}} { } } -start_server {tags {"repl external:skip"}} { +start_server {tags {"repl external:skip" repl-compression}} { set A [srv 0 client] set A_host [srv 0 host] set A_port [srv 0 port] @@ -226,7 +226,7 @@ start_server {tags {"repl external:skip"}} { } } -start_server {tags {"repl external:skip"}} { +start_server {tags {"repl external:skip" repl-compression}} { r set mykey foo start_server {} { @@ -427,7 +427,7 @@ foreach mdl {no yes} dualchannel {no yes} { } } -start_server {tags {"repl external:skip"} overrides {save {}}} { +start_server {tags {"repl external:skip" repl-compression} overrides {save {}}} { set master [srv 0 client] set master_host [srv 0 host] set master_port [srv 0 port] @@ -890,7 +890,7 @@ proc compute_cpu_usage {start end} { # test diskless rdb pipe with multiple replicas, which may drop half way -start_server {tags {"repl external:skip"} overrides {save ""}} { +start_server {tags {"repl external:skip" repl-compression} overrides {save ""}} { set master [srv 0 client] $master config set repl-diskless-sync yes $master config set repl-diskless-sync-delay 5 @@ -1470,7 +1470,7 @@ test {replica can handle EINTR if use diskless load} { } } {} {external:skip} -start_server {tags {"repl" "external:skip"}} { +start_server {tags {"repl" "external:skip" repl-compression}} { test "replica do not write the reply to the replication link - SYNC (_addReplyToBufferOrList)" { set rd [valkey_deferring_client] set lines [count_log_lines 0] @@ -1552,7 +1552,7 @@ start_server {tags {"repl" "external:skip"}} { } } -start_server {tags {"repl external:skip"}} { +start_server {tags {"repl external:skip" repl-compression}} { set master [srv 0 client] set master_host [srv 0 host] set master_port [srv 0 port] @@ -1640,7 +1640,7 @@ foreach dualchannel {yes no} { } {} {external:skip} } -start_server {tags {"repl external:skip"}} { +start_server {tags {"repl external:skip" repl-compression}} { set replica [srv 0 client] $replica set replica_key replica_value diff --git a/tests/integration/skip-rdb-checksum.tcl b/tests/integration/skip-rdb-checksum.tcl index 324096c5a20..2b01451a541 100644 --- a/tests/integration/skip-rdb-checksum.tcl +++ b/tests/integration/skip-rdb-checksum.tcl @@ -26,7 +26,7 @@ proc test_skip_rdb_checksum {primary primary_host primary_port primary_skipped_r } } -start_server {tags {"repl tls cluster:skip external:skip"} overrides {save {}}} { +start_server {tags {"repl tls cluster:skip external:skip" repl-compression} overrides {save {}}} { set primary [srv 0 client] set primary_host [srv 0 host] set primary_port [srv 0 port] diff --git a/tests/support/server.tcl b/tests/support/server.tcl index 65afff98907..1a257313be1 100644 --- a/tests/support/server.tcl +++ b/tests/support/server.tcl @@ -2,7 +2,7 @@ set ::global_overrides {} set ::tags {} set ::valgrind_errors {} # Tags that are only allowed at the top level (not in nested blocks) -set ::toplevel_only_tags {large-memory needs:other-server compatible-redis network} +set ::toplevel_only_tags {large-memory needs:other-server compatible-redis network repl-compression} proc start_server_error {executable config_file error} { set err {} diff --git a/tests/test_helper.tcl b/tests/test_helper.tcl index a54c374d16d..9bc9baa53f1 100644 --- a/tests/test_helper.tcl +++ b/tests/test_helper.tcl @@ -808,7 +808,7 @@ for {set j 0} {$j < [llength $argv]} {incr j} { } else { # Validate that allowtags only use top-level tags if {[lsearch -exact $::toplevel_only_tags $tag] < 0} { - puts "Error: --tags allowlist can only use top-level-only tags: large-memory, needs:other-server, compatible-redis, network" + puts "Error: --tags allowlist can only use top-level-only tags: $::toplevel_only_tags" puts "Invalid tag: $tag" exit 1 } diff --git a/valkey.conf b/valkey.conf index 2553d31fb67..01e82dbc211 100644 --- a/valkey.conf +++ b/valkey.conf @@ -832,6 +832,20 @@ repl-diskless-sync-max-replicas 0 # lose all your data. repl-diskless-load disabled +# Compression for the replication transport between primary and replica. +# no - no compression (default) +# lz4-stream - compress the incremental replication stream with LZ4 +# When set to a compressing mode on a replica, the replica advertises the +# "compression" capability during the PSYNC handshake. The primary records this +# capability and compresses the incremental replication stream sent to that +# replica (only if the primary itself has repl-compression enabled). +# +# Changing this at runtime reconnects only the replicas that negotiated the +# compression capability, so they renegotiate the transport (a live link cannot +# switch between plaintext and compressed mid-stream). Replicas that never +# advertised the capability are left connected and unaffected. +repl-compression no + # This dual channel replication sync feature optimizes the full synchronization process # between a primary and its replicas. When enabled, it reduces both memory and CPU load # on the primary server.