[backport] Backport sweep for 9.0

src/cluster_migrateslots.c

@@ -386,16 +386,25 @@ 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(job_name->ptr) != 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;
         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;
@@ -1581,6 +1590,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,15 +2067,19 @@ 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 "
                               "due to %s.",
                               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;
             }

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:

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);
 

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(job_name->ptr) != 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;

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

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]