From 413b51c515e869995f39f1ee5f4a30a533135ecd Mon Sep 17 00:00:00 2001 From: Satheesha CH Gowda Date: Mon, 20 Jul 2026 15:37:35 -0700 Subject: [PATCH 1/4] Avoid offloading writes to IO threads for the slot migration export job while snapshotting. (#4104) **Problem Description** Atomic slot migration is failing when io-threads enabled and pipeline requests are ongoing ```== CRITICAL == This slot-import-target is sending an error to its slot-import-source: 'Protocol error: invalid CRLF in request' after processing the command 'set'``` **Proposed Fix** Disable I/O threads offloading for slot migration jobs when snapshotting to prevent query buffer desynchronization during pipelining Signed-off-by: Satheesha Gowda --- src/cluster_migrateslots.c | 17 +++++++++++++++-- src/io_threads.c | 6 ++++++ tests/unit/cluster/cluster-migrateslots.tcl | 10 +++++----- 3 files changed, 26 insertions(+), 7 deletions(-) diff --git a/src/cluster_migrateslots.c b/src/cluster_migrateslots.c index 3f817429b97..eb5cbcb5d37 100644 --- a/src/cluster_migrateslots.c +++ b/src/cluster_migrateslots.c @@ -1581,6 +1581,15 @@ int childSnapshotForSyncSlot(rio *aof, slotMigrationJob *job) { void killSlotMigrationChild(void) { /* No slot migration child? return. */ if (server.child_type != CHILD_TYPE_SLOT_MIGRATION) return; + + /* If we already closed the exit pipe, the child is already exiting. + * Sending SIGUSR1 now might cause a race condition/deadlock in the child, + * especially when compiled with coverage. */ + if (server.slot_migration_child_exit_pipe == -1) { + serverLog(LL_NOTICE, "Slot migration child %ld is already exiting, not killing.", (long)server.child_pid); + return; + } + serverLog(LL_NOTICE, "Killing running slot migration child: %ld", (long)server.child_pid); /* Because we are not using here waitpid (like we have in killAppendOnlyChild @@ -2049,7 +2058,9 @@ void proceedWithSlotMigration(slotMigrationJob *job) { * resulting in premature flush of the output buffer and data * consistency issues. To prevent this, we defer snapshot until * there are no pending writes. */ - if (hasActiveChildProcess() || job->client->flag.pending_write) { + if (hasActiveChildProcess() || job->client->flag.pending_write || + job->client->io_write_state != CLIENT_IDLE || + job->client->io_read_state != CLIENT_IDLE) { run_with_period(5000) { serverLog(LL_NOTICE, "Slot migration %s waiting before snapshotting " @@ -2057,7 +2068,9 @@ void proceedWithSlotMigration(slotMigrationJob *job) { job->description, hasActiveChildProcess() ? "active child process" - : "pending writes in output buffer"); + : (job->client->flag.pending_write + ? "pending writes in output buffer" + : "pending IO operations")); } return; } diff --git a/src/io_threads.c b/src/io_threads.c index 3905ef2c0bd..e73fcb16b70 100644 --- a/src/io_threads.c +++ b/src/io_threads.c @@ -5,6 +5,7 @@ */ #include "io_threads.h" +#include "cluster_migrateslots.h" static _Thread_local int thread_id = 0; /* Thread local var */ static pthread_t io_threads[IO_THREADS_MAX_NUM] = {0}; @@ -422,6 +423,11 @@ int trySendWriteToIOThreads(client *c) { if (getClientType(c) == CLIENT_TYPE_REPLICA && c->repl_data->repl_state != REPLICA_STATE_ONLINE) return C_ERR; /* We can't offload debugged clients as the main-thread may read at the same time */ if (c->flag.lua_debug) return C_ERR; + /* Avoid offloading writes to IO thread for the slot migration export job while snapshotting. + * During this phase, replies accumulate in the output buffer but must not be flushed + * as concurrent IO thread writes would race with the main thread processing incoming + * ACKs on the same client's query buffer. */ + if (c->slot_migration_job && !clusterSlotMigrationShouldInstallWriteHandler(c)) return C_ERR; size_t tid = (c->id % (server.active_io_threads_num - 1)) + 1; /* Handle case where client has a pending IO read job on a different thread: diff --git a/tests/unit/cluster/cluster-migrateslots.tcl b/tests/unit/cluster/cluster-migrateslots.tcl index f68ecf07825..2223c52df99 100644 --- a/tests/unit/cluster/cluster-migrateslots.tcl +++ b/tests/unit/cluster/cluster-migrateslots.tcl @@ -173,7 +173,7 @@ proc do_node_restart {idx} { } # Disable replica migration to prevent empty nodes from joining other shards. -start_cluster 3 3 {tags {logreqres:skip external:skip cluster} overrides {cluster-allow-replica-migration no cluster-node-timeout 15000 cluster-databases 16}} { +start_cluster 3 3 {tags {logreqres:skip external:skip cluster network} overrides {cluster-allow-replica-migration no cluster-node-timeout 15000 cluster-databases 16}} { set node0_id [R 0 CLUSTER MYID] set node1_id [R 1 CLUSTER MYID] @@ -2123,7 +2123,7 @@ start_cluster 3 3 {tags {logreqres:skip external:skip cluster} overrides {cluste } } -start_cluster 3 0 {tags {logreqres:skip external:skip cluster}} { +start_cluster 3 0 {tags {logreqres:skip external:skip cluster network}} { set 16383_slot_tag "{6ZJ}" set node0_id [R 0 CLUSTER MYID] @@ -2159,7 +2159,7 @@ start_cluster 3 0 {tags {logreqres:skip external:skip cluster}} { } } -start_cluster 3 6 {tags {logreqres:skip external:skip cluster}} { +start_cluster 3 6 {tags {logreqres:skip external:skip cluster network}} { set node0_id [R 0 CLUSTER MYID] set node1_id [R 1 CLUSTER MYID] set node2_id [R 2 CLUSTER MYID] @@ -2215,7 +2215,7 @@ start_cluster 3 6 {tags {logreqres:skip external:skip cluster}} { } } -start_cluster 3 0 {tags {logreqres:skip external:skip cluster}} { +start_cluster 3 0 {tags {logreqres:skip external:skip cluster network}} { set node0_id [R 0 CLUSTER MYID] set node1_id [R 1 CLUSTER MYID] @@ -2280,7 +2280,7 @@ start_cluster 3 0 {tags {logreqres:skip external:skip cluster}} { } -start_cluster 3 3 {tags {logreqres:skip external:skip cluster aofrw} overrides {appendonly yes auto-aof-rewrite-percentage 0}} { +start_cluster 3 3 {tags {logreqres:skip external:skip cluster aofrw network} overrides {appendonly yes auto-aof-rewrite-percentage 0}} { set node0_id [R 0 CLUSTER MYID] set node1_id [R 1 CLUSTER MYID] set node2_id [R 2 CLUSTER MYID] From 4bb2a0288ac345f90c074db13d8ff23ebd5f8edb Mon Sep 17 00:00:00 2001 From: Binbin Date: Wed, 22 Jul 2026 04:23:05 +0800 Subject: [PATCH 2/4] Validate slot import ranges when loading from RDB (#4229) When loading slot migration import jobs from the RDB, the start and end slots of each range were not validated. A corrupted or truncated RDB could produce out-of-range slot values or a reversed range (start > end), which would then be used to build migration jobs and slot ranges. Reject any slot import range whose start or end slot is greater than or equal to CLUSTER_SLOTS, or whose start slot is greater than the end slot, logging a warning and aborting the load. Fixes #4222. Signed-off-by: Binbin --- src/cluster_migrateslots.c | 5 +++++ src/kvstore.c | 4 ++-- 2 files changed, 7 insertions(+), 2 deletions(-) diff --git a/src/cluster_migrateslots.c b/src/cluster_migrateslots.c index eb5cbcb5d37..35f31488303 100644 --- a/src/cluster_migrateslots.c +++ b/src/cluster_migrateslots.c @@ -396,6 +396,11 @@ int clusterRDBLoadSlotImport(rio *rdb) { uint64_t end_slot; if ((start_slot = rdbLoadLen(rdb, NULL)) == RDB_LENERR) goto err; if ((end_slot = rdbLoadLen(rdb, NULL)) == RDB_LENERR) goto err; + if (start_slot >= CLUSTER_SLOTS || end_slot >= CLUSTER_SLOTS || start_slot > end_slot) { + serverLog(LL_WARNING, "Invalid slot import range in RDB: start=%llu end=%llu", + (unsigned long long)start_slot, (unsigned long long)end_slot); + goto err; + } slotRange *slot_range = zmalloc(sizeof(slotRange)); slot_range->start_slot = start_slot; diff --git a/src/kvstore.c b/src/kvstore.c index bd1574727e5..1d055932e2c 100644 --- a/src/kvstore.c +++ b/src/kvstore.c @@ -147,7 +147,7 @@ static int getAndClearHashtableIndexFromCursor(kvstore *kvs, unsigned long long } int kvstoreIsImporting(kvstore *kvs, int didx) { - assert(didx < kvs->num_hashtables); + assert(didx >= 0 && didx < kvs->num_hashtables); return hashtableFind(kvs->importing, (void *)(intptr_t)didx, NULL); } @@ -938,7 +938,7 @@ bool kvstoreHashtableDelete(kvstore *kvs, int didx, const void *key) { * are not included in hashtable metrics and are excluded from scanning and * random key lookup. */ void kvstoreSetIsImporting(kvstore *kvs, int didx, int is_importing) { - assert(didx < kvs->num_hashtables); + assert(didx >= 0 && didx < kvs->num_hashtables); hashtable *ht = kvstoreGetHashtable(kvs, didx); From 61e424c86609a01c7e25d238942d16de43688039 Mon Sep 17 00:00:00 2001 From: Quanye Yang Date: Wed, 22 Jul 2026 04:46:47 +0800 Subject: [PATCH 3/4] Check and reject invalid slot import job names during RDB load (#4210) ## Summary - `clusterRDBLoadSlotImport()` accepted a variable-length `job_name` from RDB, but `createSlotImportJob()` always `memcpy`'s `CLUSTER_NAMELEN` (40) bytes. A crafted RDB with a shorter name caused a heap out-of-bounds read during startup. - Reject slot-import records whose `job_name` length is not exactly `CLUSTER_NAMELEN`, matching the existing command-path check. - Add an integration test that crafts a short `job_name` RDB and asserts the server fails closed. Fixes #4207 ## Test plan - [x] `./runtest --single tests/integration/rdb-slot-import.tcl` --------- Signed-off-by: quanyeyang --- src/cluster_migrateslots.c | 6 +- src/valkey-check-rdb.c | 6 ++ tests/integration/rdb-slot-import.tcl | 106 ++++++++++++++++++++++++++ 3 files changed, 117 insertions(+), 1 deletion(-) create mode 100644 tests/integration/rdb-slot-import.tcl diff --git a/src/cluster_migrateslots.c b/src/cluster_migrateslots.c index 35f31488303..a8e8a81c9da 100644 --- a/src/cluster_migrateslots.c +++ b/src/cluster_migrateslots.c @@ -386,10 +386,14 @@ int clusterRDBSaveSlotImports(rio *rdb) { /* Load a single slot import from the RDB. */ int clusterRDBLoadSlotImport(rio *rdb) { - robj *job_name; + robj *job_name = NULL; list *slot_ranges = createSlotRangeList(); uint64_t num_slot_ranges; if ((job_name = rdbLoadStringObject(rdb)) == NULL) goto err; + if (sdslen(objectGetVal(job_name)) != CLUSTER_NAMELEN) { + serverLog(LL_WARNING, "Invalid slot import job name length in RDB"); + goto err; + } if ((num_slot_ranges = rdbLoadLen(rdb, NULL)) == RDB_LENERR) goto err; for (uint64_t i = 0; i < num_slot_ranges; i++) { uint64_t start_slot; diff --git a/src/valkey-check-rdb.c b/src/valkey-check-rdb.c index 10aa12d3130..eb2ef235b19 100644 --- a/src/valkey-check-rdb.c +++ b/src/valkey-check-rdb.c @@ -31,6 +31,7 @@ #include "server.h" #include "rdb.h" #include "module.h" +#include "cluster.h" #include "hdr_histogram.h" #include "fpconv_dtoa.h" @@ -703,6 +704,11 @@ int redis_check_rdb(char *rdbfilename, FILE *fp) { } else if (type == RDB_OPCODE_SLOT_IMPORT) { robj *job_name; if ((job_name = rdbLoadStringObject(&rdb)) == NULL) goto eoferr; + if (sdslen(objectGetVal(job_name)) != CLUSTER_NAMELEN) { + rdbCheckError("Invalid slot import job name length in RDB"); + decrRefCount(job_name); + goto err; + } decrRefCount(job_name); uint64_t num_slot_ranges; if ((num_slot_ranges = rdbLoadLen(&rdb, NULL)) == RDB_LENERR) goto eoferr; diff --git a/tests/integration/rdb-slot-import.tcl b/tests/integration/rdb-slot-import.tcl new file mode 100644 index 00000000000..bfc2194cb9e --- /dev/null +++ b/tests/integration/rdb-slot-import.tcl @@ -0,0 +1,106 @@ +tags {"rdb cluster external:skip"} { + +# Helper: start a server that is expected to fail during RDB load, then inspect logs. +proc start_server_and_kill_it {overrides code} { + upvar srv srv + set ::slot_import_short_name_asan 0 + set srv [start_server [list overrides $overrides keep_persistence true]] + uplevel 1 $code + # Server may already be dead after a failed RDB load; skip leak checks. + dict set srv skipleaks 1 + # Pre-fix ASan OOB is already reported by the test assertion; clear stderr + # so kill_server does not emit a duplicate sanitizer failure. + if {$::slot_import_short_name_asan} { + close [open [dict get $srv stderr] w] + } + kill_server $srv +} + +# Craft an RDB with RDB_OPCODE_SLOT_IMPORT (0xF3) whose job_name is 1 byte. +# A correct save path always writes CLUSTER_NAMELEN (40) bytes; the load path +# must reject shorter names before createSlotImportJob() memcpy's 40 bytes. +# +# Trailing checksum is written as 8 zero bytes. The loader treats cksum==0 as +# "checksum disabled", so we do not need an external CRC64 helper. +proc craft_slot_import_short_job_name_rdb {src_rdb dst_rdb} { + set fd [open $src_rdb rb] + fconfigure $fd -translation binary + set data [read $fd] + close $fd + + # Empty RDB ends with: EOF(0xFF) + 8-byte CRC64. + binary scan [string index $data end-8] c eof_byte + set eof_byte [expr {$eof_byte & 0xff}] + if {$eof_byte != 0xff} { + error "Expected RDB EOF opcode 0xFF before checksum, got $eof_byte" + } + + set prefix [string range $data 0 end-9] + # F3 | len=1 | 'A' | num_ranges=1 | start=0 | end=0 | FF | cksum=0 + set record [binary format H* f30141010000ff0000000000000000] + set fd [open $dst_rdb wb] + fconfigure $fd -translation binary + puts -nonewline $fd $prefix + puts -nonewline $fd $record + close $fd +} + +set gen_path [tmpdir "rdb-slot-import-gen"] +start_server [list overrides [list "dir" $gen_path "save" "" "dbfilename" "empty.rdb"] keep_persistence true] { + test {Generate baseline empty RDB for slot-import craft} { + r save + assert_equal 1 [file exists [file join $gen_path empty.rdb]] + } +} + +set victim_path [tmpdir "rdb-slot-import-short-name"] +set victim_rdb [file join $victim_path dump.rdb] +craft_slot_import_short_job_name_rdb \ + [file join $gen_path empty.rdb] \ + $victim_rdb + +test {valkey-check-rdb rejects short slot-import job_name} { + catch { + exec $::VALKEY_CHECK_RDB_BIN $victim_rdb + } result + assert_match {*--- RDB ERROR DETECTED ---*} $result + assert_match {*Invalid slot import job name length*} $result + assert_no_match {*RDB looks OK*} $result +} + +start_server_and_kill_it [list \ + "dir" $victim_path \ + "dbfilename" "dump.rdb" \ + "save" "" \ + "appendonly" "no" \ + "cluster-enabled" "yes" \ + "cluster-config-file" "nodes.conf" \ +] { + test {Server rejects RDB slot-import job_name shorter than CLUSTER_NAMELEN} { + # Before the fix (see #4207): ASan aborts in createSlotImportJob()'s + # memcpy(..., CLUSTER_NAMELEN) on a 1-byte job_name, or a non-ASan build + # may perform an OOB read. After the fix the loader must fail closed. + wait_for_condition 50 100 { + [string match {*Invalid slot import job name*} \ + [exec cat [dict get $srv stdout]]] || + [string match {*Short read or OOM loading DB*} \ + [exec cat [dict get $srv stdout]]] || + [string match {*Fatal error loading the DB*} \ + [exec cat [dict get $srv stdout]]] + } else { + set stdout [exec cat [dict get $srv stdout]] + set stderr [exec cat [dict get $srv stderr]] + if {[string match {*heap-buffer-overflow*} $stderr] || + [string match {*AddressSanitizer*} $stderr]} { + set ::slot_import_short_name_asan 1 + fail "Bug reproduced: short slot-import job_name caused sanitizer OOB read during RDB load (expected until fixed)." + } + fail "Server did not reject short slot-import job_name RDB.\nSTDOUT:\n$stdout\nSTDERR:\n$stderr" + } + + # Must not become ready for clients. + assert_equal 0 [count_message_lines [dict get $srv stdout] "Ready to accept"] + } +} + +} ;# tags From c1744a1afa6ed41239519ebc283d89250a303391 Mon Sep 17 00:00:00 2001 From: "valkeyrie-bot[bot]" <3692572+valkeyrie-bot[bot]@users.noreply.github.com> Date: Wed, 22 Jul 2026 09:52:14 +0000 Subject: [PATCH 4/4] Repair backport validation failure --- src/cluster_migrateslots.c | 2 +- src/valkey-check-rdb.c | 2 +- 2 files changed, 2 insertions(+), 2 deletions(-) diff --git a/src/cluster_migrateslots.c b/src/cluster_migrateslots.c index a8e8a81c9da..42023828827 100644 --- a/src/cluster_migrateslots.c +++ b/src/cluster_migrateslots.c @@ -390,7 +390,7 @@ int clusterRDBLoadSlotImport(rio *rdb) { list *slot_ranges = createSlotRangeList(); uint64_t num_slot_ranges; if ((job_name = rdbLoadStringObject(rdb)) == NULL) goto err; - if (sdslen(objectGetVal(job_name)) != CLUSTER_NAMELEN) { + if (sdslen(job_name->ptr) != CLUSTER_NAMELEN) { serverLog(LL_WARNING, "Invalid slot import job name length in RDB"); goto err; } diff --git a/src/valkey-check-rdb.c b/src/valkey-check-rdb.c index eb2ef235b19..f7e110bcd63 100644 --- a/src/valkey-check-rdb.c +++ b/src/valkey-check-rdb.c @@ -704,7 +704,7 @@ int redis_check_rdb(char *rdbfilename, FILE *fp) { } else if (type == RDB_OPCODE_SLOT_IMPORT) { robj *job_name; if ((job_name = rdbLoadStringObject(&rdb)) == NULL) goto eoferr; - if (sdslen(objectGetVal(job_name)) != CLUSTER_NAMELEN) { + if (sdslen(job_name->ptr) != CLUSTER_NAMELEN) { rdbCheckError("Invalid slot import job name length in RDB"); decrRefCount(job_name); goto err;