From d3b4f4843c77f113c2d7d4a26b36906dddb08125 Mon Sep 17 00:00:00 2001 From: Warrick <1016weicheng@gmail.com> Date: Fri, 11 Sep 2026 18:24:34 +0800 Subject: [PATCH] [fix] fix bug of ratelimit for masterhost client --- src/ctrip_swap_evict.c | 8 +- src/ctrip_swap_server.h | 1 + src/networking.c | 25 ++ src/server.h | 18 +- tests/swap/integration/repl_self_connect.tcl | 265 +++++++++++++++++++ tests/test_helper.tcl | 1 + 6 files changed, 307 insertions(+), 11 deletions(-) create mode 100644 tests/swap/integration/repl_self_connect.tcl diff --git a/src/ctrip_swap_evict.c b/src/ctrip_swap_evict.c index 4578c3787c5..cdf1b2444e8 100644 --- a/src/ctrip_swap_evict.c +++ b/src/ctrip_swap_evict.c @@ -454,11 +454,11 @@ int swapRateLimitReject(swapRatelimitCtx *rlctx, client *c) { return 0; } -static int unprotectClientdProc( +static int ratelimitResumeClientProc( struct aeEventLoop *el, long long id, void *clientData) { client *c = clientData; UNUSED(el), UNUSED(id); - unprotectClient(c); + ratelimitResumeClient(c); c->rate_limit_event_id = -1; return AE_NOMORE; } @@ -471,8 +471,8 @@ void swapRateLimitPause(swapRatelimitCtx *rlctx, client *c) { if (swapRatelimitNeeded(rlctx,server.swap_ratelimit_policy,&pause_ms) && pause_ms > 0 && c->rate_limit_event_id == -1) { - protectClient(c); - c->rate_limit_event_id = aeCreateTimeEvent(server.el,pause_ms,unprotectClientdProc,c,NULL); + ratelimitPauseClient(c); + c->rate_limit_event_id = aeCreateTimeEvent(server.el,pause_ms,ratelimitResumeClientProc,c,NULL); server.stat_swap_ratelimit_client_pause_count++; server.stat_swap_ratelimit_client_pause_ms += pause_ms; } diff --git a/src/ctrip_swap_server.h b/src/ctrip_swap_server.h index 30aeb290d9d..ec8c6a5d36e 100644 --- a/src/ctrip_swap_server.h +++ b/src/ctrip_swap_server.h @@ -47,6 +47,7 @@ #define CLIENT_SWAP_DISCARD_CACHED_MASTER (1ULL<<47) /* The client will not be saved as cached_master. */ #define CLIENT_SWAP_SHIFT_REPL_ID (1ULL<<48) /* shift repl id when this client (drainning master) drained. */ #define CLIENT_SWAP_DONT_RECONNECT_MASTER (1ULL<<49) /* shift repl id when this client (drainning master) drained. */ +#define CLIENT_SWAP_RATELIMIT_PAUSED (1ULL<<50) /* CHECK: SLAVE_CAPA_PSYNC2 is the last SLAVE_CAPA_xx flag */ #define SLAVE_CAPA_RORDB (1<<2) /* Can parse RORDB format. */ diff --git a/src/networking.c b/src/networking.c index 6573b00b9bd..1a146754f10 100644 --- a/src/networking.c +++ b/src/networking.c @@ -1373,6 +1373,7 @@ void unlinkClient(client *c) { aeDeleteTimeEvent(server.el, c->rate_limit_event_id); c->rate_limit_event_id = -1; } + c->flags &= ~CLIENT_SWAP_RATELIMIT_PAUSED; #endif } @@ -1820,7 +1821,11 @@ int handleClientsWithPendingWrites(void) { /* If a client is protected, don't do anything, * that may trigger write error or recreate handler. */ +#ifdef ENABLE_SWAP + if (c->flags & (CLIENT_PROTECTED|CLIENT_SWAP_RATELIMIT_PAUSED)) continue; +#else if (c->flags & CLIENT_PROTECTED) continue; +#endif /* Don't write to clients that are going to be closed anyway. */ if (c->flags & CLIENT_CLOSE_ASAP) continue; @@ -1922,6 +1927,26 @@ void unprotectClient(client *c) { } } +#ifdef ENABLE_SWAP +void ratelimitPauseClient(client *c) { + c->flags |= CLIENT_SWAP_RATELIMIT_PAUSED; + if (c->conn) { + connSetReadHandler(c->conn,NULL); + connSetWriteHandler(c->conn,NULL); + } +} + +void ratelimitResumeClient(client *c) { + if (c->flags & CLIENT_SWAP_RATELIMIT_PAUSED) { + c->flags &= ~CLIENT_SWAP_RATELIMIT_PAUSED; + if (c->conn) { + connSetReadHandler(c->conn,readQueryFromClient); + if (clientHasPendingReplies(c)) clientInstallWriteHandler(c); + } + } +} +#endif + /* Like processMultibulkBuffer(), but for the inline protocol instead of RESP, * this function consumes the client query buffer and creates a command ready * to be executed inside the client structure. Returns C_OK if the command diff --git a/src/server.h b/src/server.h index 954df547905..2f7055fa987 100644 --- a/src/server.h +++ b/src/server.h @@ -299,15 +299,15 @@ extern int configOOMScoreAdjValuesDefaults[CONFIG_OOM_COUNT]; and AOF client */ #define CLIENT_REPL_RDBONLY (1ULL<<42) /* This client is a replica that only wants RDB without replication buffer. */ -/* 1ULL<<43 ~ 1ULL<<49 CLIENT_SWAP_xx flag */ +/* 1ULL<<43 ~ 1ULL<<50 CLIENT_SWAP_xx flag */ -#define CLIENT_TRACKING_HEARTBEAT_SYSTIME (1ULL<<50) /* Heartbeat with systime. */ -#define CLIENT_TRACKING_HEARTBEAT_MKPS (1ULL<<51) /* Heartbeat with mkps(modified keys per second). */ -#define CLIENT_TRACKING_SUBKEY (1ULL<<52) /* Tracking in subkey mode. */ -#define CLIENT_NO_EVICT (1ULL<<53) /* This client is protected against client +#define CLIENT_TRACKING_HEARTBEAT_SYSTIME (1ULL<<51) /* Heartbeat with systime. */ +#define CLIENT_TRACKING_HEARTBEAT_MKPS (1ULL<<52) /* Heartbeat with mkps(modified keys per second). */ +#define CLIENT_TRACKING_SUBKEY (1ULL<<53) /* Tracking in subkey mode. */ +#define CLIENT_NO_EVICT (1ULL<<54) /* This client is protected against client memory eviction. */ -#define CLIENT_TRACKING_INVALIDATEOFF (1ULL<<54) -#define CLIENT_TRACKING_PREFIXRESET (1ULL<<55) +#define CLIENT_TRACKING_INVALIDATEOFF (1ULL<<55) +#define CLIENT_TRACKING_PREFIXRESET (1ULL<<56) /* Client block type (btype field in client structure) * if CLIENT_BLOCKED flag is set. */ @@ -2047,6 +2047,10 @@ int writeToClient(client *c, int handler_installed); void linkClient(client *c); void protectClient(client *c); void unprotectClient(client *c); +#ifdef ENABLE_SWAP +void ratelimitPauseClient(client *c); +void ratelimitResumeClient(client *c); +#endif void initThreadedIO(void); client *lookupClientByID(uint64_t id); int authRequired(client *c); diff --git a/tests/swap/integration/repl_self_connect.tcl b/tests/swap/integration/repl_self_connect.tcl new file mode 100644 index 00000000000..91cff5e7d39 --- /dev/null +++ b/tests/swap/integration/repl_self_connect.tcl @@ -0,0 +1,265 @@ +proc log_file_matches {log pattern} { + set fp [open $log r] + set content [read $fp] + close $fp + string match $pattern $content +} + +proc assert_no_master_link {r own_port peer where} { + assert_equal "master" [status $r role] + set masterlinks {} + foreach line [split [$r client list] "\n"] { + if {[string match "*flags=M*" $line]} { lappend masterlinks $line } + } + if {[llength $masterlinks] != 0} { + fail "$where: role is master but a master link still exists: $masterlinks" + } + # A ghost self replica advertises our own listening port (this is the shape + # seen in production, where master and replica both listen on 6379). + foreach line [split [$r info replication] "\n"] { + if {[string match "slave*port=$own_port,*" $line]} { + fail "$where: we registered ourselves as our own replica: [string trim $line]" + } + } + # A ghost link to the old master shows up as an extra replica on the peer. + if {$peer ne ""} { + set peer_slaves [status $peer connected_slaves] + if {$peer_slaves != 0} { + fail "$where: peer still counts $peer_slaves replica(s), we are replicating from it while claiming to be a master" + } + } +} + +start_server {tags {"repl swap"} overrides {}} { + start_server {} { + set master [srv -1 client] + set master_host [srv -1 host] + set master_port [srv -1 port] + + set slave [srv 0 client] + set slave_port [srv 0 port] + set slave_log [srv 0 stdout] + + set keycount 300 + + test {ratelimit repro: setup cold keyspace} { + $slave slaveof $master_host $master_port + wait_for_sync $slave + for {set i 0} {$i < $keycount} {incr i} { + $master set key_$i val_$i + } + wait_keyspace_cold $master + wait_keyspace_cold $slave + assert_equal [status $slave role] slave + } + + test {SLAVEOF NO ONE while master client is ratelimit paused must not replicate from itself} { + # Arm the swap ratelimiter so that replicated commands pause the + # master client for SWAP_RATELIMIT_PAUSE_MAX_MS (200ms). + # maxmemory must be set in one step from 0, otherwise + # maxmemory_scale_from only scales down gradually and never arms. + # Note that the ratelimiter compares against ctrip_getUsedMemory(), + # which is well below INFO's used_memory (cold filters excluded). + $slave config set swap-ratelimit-policy pause + $slave config set swap-ratelimit-maxmemory-pause-growth-rate 1 + $slave config set maxmemory 0 + $slave config set maxmemory 3mb + + set master_rd [redis_deferring_client -1] + set pause_observed_rounds 0 + set rounds 10 + + for {set round 0} {$round < $rounds} {incr round} { + if {[status $slave role] eq "master"} { + $slave slaveof $master_host $master_port + } + wait_for_sync $slave + + set pauses_before [getInfoProperty [$slave info swap] \ + swap_ratelimit_client_pause_count] + + # Keep the replica's master client swapping so that the + # ratelimiter pauses it. + for {set i 0} {$i < $keycount} {incr i} { + $master_rd append key_$i x + } + + # Deterministic precondition: don't guess with a fixed sleep, + # wait until the ratelimiter actually paused a client. + set pause_observed 0 + for {set i 0} {$i < 100} {incr i} { + if {[getInfoProperty [$slave info swap] \ + swap_ratelimit_client_pause_count] > $pauses_before} { + set pause_observed 1 + incr pause_observed_rounds + break + } + after 20 + } + + $slave slaveof no one + for {set i 0} {$i < $keycount} {incr i} { $master_rd read } + + if {!$pause_observed} continue + + # replicationCron dials once per second: give it time to expose + # a broken repl_state, then assert on the state itself. + after 1500 + assert_no_master_link $slave $slave_port $master "round $round" + if {[log_file_matches $slave_log "*Connecting to MASTER (null):*"]} { + fail "round $round: replicationCron dialed MASTER (null): repl_state == REPL_STATE_CONNECT while masterhost == NULL" + } + } + + $master_rd close + + # Guard against a vacuous pass: if the ratelimiter never paused a + # client, SLAVEOF NO ONE never ran inside a pause window and the race + # was not exercised at all. That would happen for instance if the + # ratelimiter thresholds, config names or arming rules change. + if {$pause_observed_rounds == 0} { + fail "swap ratelimiter never paused any client in $rounds rounds, the test did not exercise the race" + } + } + + test {swap ratelimit pause must not defer freeClient} { + $master config set swap-ratelimit-policy pause + $master config set swap-ratelimit-maxmemory-pause-growth-rate 1 + $master config set maxmemory 0 + $master config set maxmemory 3mb + wait_keyspace_cold $master + + set victim [redis_deferring_client -1] + $victim client setname ratelimit_victim + $victim read + + set pauses_before [getInfoProperty [$master info swap] \ + swap_ratelimit_client_pause_count] + + # A cold read submits key requests and then gets ratelimit paused. + $victim get key_0 + + set pause_observed 0 + for {set i 0} {$i < 100} {incr i} { + if {[getInfoProperty [$master info swap] \ + swap_ratelimit_client_pause_count] > $pauses_before} { + set pause_observed 1 + break + } + after 20 + } + if {!$pause_observed} { + catch {$victim close} + fail "ratelimiter did not pause the victim client, nothing was exercised" + } + # Let the swap itself finish (sub-millisecond) so that the client has + # no in flight key requests: those are deferred by a different + # mechanism (deferFreeClient) which we are not testing here. + after 20 + + set victim_line "" + foreach line [split [$master client list] "\n"] { + if {[string match "*name=ratelimit_victim*" $line]} { set victim_line $line } + } + assert {$victim_line ne ""} + regexp {id=(\d+)} $victim_line _ victim_id + + assert_equal 1 [$master client kill id $victim_id] + + # No sleep, no timing threshold: a synchronous free already happened + # while CLIENT KILL was being executed. + set still_there 0 + foreach line [split [$master client list] "\n"] { + if {[string match "*name=ratelimit_victim*" $line]} { set still_there 1 } + } + catch {$victim close} + if {$still_there} { + fail "a ratelimit paused client survived CLIENT KILL: freeClient() was deferred, so pausing must not be implemented with protectClient()" + } + + $master config set maxmemory 0 + } + + # Since pausing no longer defers the free, a paused master client can be + # unlinked while paused: unlinkClient() cancels the resume time event. + # When the link is merely broken (not SLAVEOF NO ONE) the client object + # is cached (replicationCacheMaster) and resurrected after a successful + # PSYNC (replicationResurrectCachedMaster). If the pause state were not + # dropped together with the time event, the resurrected master client + # would keep CLIENT_SWAP_RATELIMIT_PAUSED forever, and + # handleClientsWithPendingWrites() would skip it: REPLCONF ACKs would + # never reach the master, so the lag the master sees keeps growing + # (breaking WAIT and min-replicas-max-lag), until the next ratelimit + # pause/resume cycle happens to clear the flag. + test {ratelimit paused master client resurrected after PSYNC keeps sending ACKs} { + if {[status $slave role] eq "master"} { + $slave slaveof $master_host $master_port + } + wait_for_sync $slave + + set master_rd [redis_deferring_client -1] + set rounds 3 + + for {set round 0} {$round < $rounds} {incr round} { + # Arm the ratelimiter on the replica. + $slave config set swap-ratelimit-policy pause + $slave config set swap-ratelimit-maxmemory-pause-growth-rate 1 + $slave config set maxmemory 0 + $slave config set maxmemory 3mb + + set pauses_before [getInfoProperty [$slave info swap] \ + swap_ratelimit_client_pause_count] + set partial_before [status $master sync_partial_ok] + + for {set i 0} {$i < $keycount} {incr i} { + $master_rd append key_$i z + } + wait_for_condition 200 5 { + [getInfoProperty [$slave info swap] \ + swap_ratelimit_client_pause_count] > $pauses_before + } else { + fail "round $round: ratelimiter never paused the master client, nothing was exercised" + } + + # Break the link while the master client is paused, then disarm + # the ratelimiter so that no later pause/resume cycle can clear + # a stale flag behind our back. + assert_equal 1 [$slave client kill type master] + $slave config set maxmemory 0 + for {set i 0} {$i < $keycount} {incr i} { $master_rd read } + + wait_for_condition 100 100 { + [status $slave master_link_status] eq "up" + } else { + fail "round $round: replication link did not come back" + } + # The cached master client must have been resurrected, otherwise + # this round did not exercise the code path at all. + wait_for_condition 50 100 { + [status $master sync_partial_ok] > $partial_before + } else { + fail "round $round: reconnection was not a partial resync, the cached master client was not resurrected" + } + + # The replica sends REPLCONF ACK every second, so the lag seen + # by the master normally stays at 0 or 1. A stale pause flag + # makes it grow by one every second. + set max_lag 0 + for {set s 0} {$s < 4} {incr s} { + after 1000 + set line [lsearch -inline [split [$master info replication] "\r\n"] "slave0:*"] + if {![regexp {lag=(\d+)} $line _ lag]} { + fail "round $round: cannot parse replica lag from '$line'" + } + if {$lag > $max_lag} { set max_lag $lag } + } + if {$max_lag > 2} { + fail "round $round: master side lag reached $max_lag seconds after the paused master client was resurrected: REPLCONF ACKs are not sent, the ratelimit pause flag survived unlinkClient()" + } + } + + $master_rd close + $slave config set maxmemory 0 + } + } +} diff --git a/tests/test_helper.tcl b/tests/test_helper.tcl index 31e8bf7d73e..5fb21ec8f06 100644 --- a/tests/test_helper.tcl +++ b/tests/test_helper.tcl @@ -130,6 +130,7 @@ set ::disk_tests { swap/integration/swap_load swap/integration/persist swap/integration/rocksdb_log_rotate + swap/integration/repl_self_connect swap/unit/swap_mode swap/unit/absent_cache swap/unit/dbsize