diff --git a/src/cluster_migrateslots.c b/src/cluster_migrateslots.c index 50ba3270fb2..67fe628df74 100644 --- a/src/cluster_migrateslots.c +++ b/src/cluster_migrateslots.c @@ -390,16 +390,25 @@ int clusterRDBSaveSlotImports(rio *rdb, int rdbver) { /* 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; 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; @@ -1583,6 +1592,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 @@ -2051,7 +2069,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 " @@ -2059,7 +2079,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 d6368f5524e..fa864cc1ef6 100644 --- a/src/io_threads.c +++ b/src/io_threads.c @@ -5,6 +5,7 @@ */ #include "io_threads.h" +#include "cluster_migrateslots.h" #include "queues.h" #include @@ -545,6 +546,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; int is_replica = getClientType(c) == CLIENT_TYPE_REPLICA; clientReplyBlock *block = NULL; diff --git a/src/kvstore.c b/src/kvstore.c index 846b58705eb..71d1068cd71 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); } @@ -949,7 +949,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); diff --git a/src/valkey-check-rdb.c b/src/valkey-check-rdb.c index 3243b9da590..9ef59bcbc4e 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" @@ -704,6 +705,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 diff --git a/tests/unit/cluster/cluster-migrateslots.tcl b/tests/unit/cluster/cluster-migrateslots.tcl index ccbbbfcc2dd..bbbadaa2743 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] @@ -2138,7 +2138,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] @@ -2174,7 +2174,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] @@ -2230,7 +2230,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] @@ -2295,7 +2295,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] @@ -2445,7 +2445,7 @@ start_cluster 3 3 {tags {logreqres:skip external:skip cluster aofrw} overrides { } } -start_cluster 3 0 {tags {logreqres:skip external:skip cluster} overrides {cluster-require-full-coverage no slot-migration-max-failover-repl-bytes 0}} { +start_cluster 3 0 {tags {logreqres:skip external:skip cluster network} overrides {cluster-require-full-coverage no slot-migration-max-failover-repl-bytes 0}} { test "Slot migration remaining_repl_size on the source node" { set 16383_slot_tag "{6ZJ}" set_debug_prevent_pause 1 @@ -2488,7 +2488,7 @@ start_cluster 3 0 {tags {logreqres:skip external:skip cluster} overrides {cluste } } -start_cluster 3 0 {tags {logreqres:skip external:skip cluster} overrides {cluster-require-full-coverage no slot-migration-max-failover-repl-bytes -1}} { +start_cluster 3 0 {tags {logreqres:skip external:skip cluster network} overrides {cluster-require-full-coverage no slot-migration-max-failover-repl-bytes -1}} { test "slot-migration-max-failover-repl-bytes -1 disables repl bytes limit" { set 16383_slot_tag "{6ZJ}"