Fix deferred freeClient clobbering replication state after replicaof (#3719)

## Summary

This PR addresses 2 race conditions where deferred `freeClient`
(introduced by #3324) clobbers replication state set by `REPLICAOF`
commands.

Found while triaging the recurring `test-ubuntu-tls-io-threads` daily CI
failure in `tests/unit/wait.tcl`.

## Root Cause

Since PR #3324 ("Redesign IO threading communication model"),
`freeClient()` on a primary client with pending IO is deferred via
`freeClientAsync` (gated on `clientHasPendingIO`). When the deferred
free eventually executes, it chains through `replicationCachePrimary()`
-> `replicationHandlePrimaryDisconnection()`, which unconditionally sets
`server.repl_state = REPL_STATE_CONNECT`.

This causes two bugs:

### Bug 1: REPLICAOF NO ONE (SIGSEGV)

`replicationUnsetPrimary()` sets `primary_host = NULL` before calling
`freeClient`. The deferred free runs later, sets `repl_state =
REPL_STATE_CONNECT` while `primary_host` is still NULL.
`replicationCron` then calls `connectWithPrimary()` which passes NULL to
`connTLSConnect()` -> `inet_pton(AF_INET, NULL, ...)` -> SIGSEGV.

### Bug 2: REPLICAOF newhost newport (connection leak)

`replicationSetPrimary()` calls `freeClient(old_primary)` (deferred),
then sets `primary_host` to the new IP and progresses `repl_state` to
`REPL_STATE_CONNECTING` with a new connection handle in
`server.repl_transfer_s`. The deferred free runs later, clobbers
`repl_state` back to `REPL_STATE_CONNECT`. `replicationCron` then calls
`connectWithPrimary()` again, overwriting `server.repl_transfer_s`
without closing the previous connection -- an FD leak.

## Fix

Make `replicationHandlePrimaryDisconnection()` only transition to
`REPL_STATE_CONNECT` when `repl_state` is still `REPL_STATE_CONNECTED`
and `primary_host` is set. This means the disconnection is genuine and
no other state transition has already occurred. If `repl_state` has
already moved on (CONNECT, CONNECTING, NONE, etc.), the deferred free is
stale and the function leaves the state untouched.

Additionally:
- `connTLSConnect()`: Return `C_ERR` if `addr` is NULL (defense in
depth).
- `tests/unit/wait.tcl`: Add 10s timeout to the blocking WAITAOF test,
and add dedicated tests for the repoint scenario.

## Reproduction

Reproduced locally by establishing TLS replication and executing
`REPLICAOF NO ONE`:
- Without fix: server crashes with signal 11, accessing address 0x0
- With fix: server continues operating normally

## Testing

- Full `unit/wait` test suite passes (51/51) with IO threads enabled
- New tests "Repoint replica between primaries does not leak connections
or crash" and "Rapid repoint does not crash or leak" pass
- Crash reproduced locally over TLS (SIGSEGV without fix, graceful
handling with fix)

---------

Signed-off-by: Yaron Sananes <yaron.sananes@gmail.com>
Signed-off-by: Ran Shidlansik <ranshid@amazon.com>
Co-authored-by: Ran Shidlansik <ranshid@amazon.com>

Author: yaronsananes

src/debug.c

@@ -1066,6 +1066,9 @@ void debugCommand(client *c) {
     } else if (!strcasecmp(objectGetVal(c->argv[1]), "client-enforce-reply-list") && c->argc == 3) {
         server.debug_client_enforce_reply_list = atoi(objectGetVal(c->argv[2]));
         addReply(c, shared.ok);
+    } else if (!strcasecmp(objectGetVal(c->argv[1]), "force-free-primary-async") && c->argc == 3) {
+        server.debug_force_free_primary_async = atoi(objectGetVal(c->argv[2]));
+        addReply(c, shared.ok);
     } else if (!handleDebugClusterCommand(c)) {
         addReplySubcommandSyntaxError(c);
         return;

src/networking.c

@@ -2102,6 +2102,14 @@ int freeClient(client *c) {
         return 0;
     }
 
+    /* Debug: force async free for the primary client to deterministically
+     * reproduce the deferred-free replication state clobber race. */
+    if (server.debug_force_free_primary_async && c->flag.primary) {
+        server.debug_force_free_primary_async = 0;
+        freeClientAsync(c);
+        return 0;
+    }
+
     /* For connected clients, call the disconnection event of modules hooks. */
     if (c->conn) {
         moduleFireServerEvent(VALKEYMODULE_EVENT_CLIENT_CHANGE, VALKEYMODULE_SUBEVENT_CLIENT_CHANGE_DISCONNECTED, c);

src/replication.c

@@ -4560,15 +4560,32 @@ void replicationHandlePrimaryDisconnection(void) {
         moduleFireServerEvent(VALKEYMODULE_EVENT_PRIMARY_LINK_CHANGE, VALKEYMODULE_SUBEVENT_PRIMARY_LINK_DOWN, NULL);
 
     server.primary = NULL;
-    server.repl_state = REPL_STATE_CONNECT;
-    server.repl_down_since = server.unixtime;
+
+    /* freeClient(primary) can be deferred via freeClientAsync when the client
+     * has pending IO. By the time we run in that deferred context,
+     * replicationUnsetPrimary()/replicationSetPrimary() may have already
+     * finalized replication state. Only transition to REPL_STATE_CONNECT if
+     * we were genuinely connected (REPL_STATE_CONNECTED) and primary_host is
+     * still set. Otherwise this is a stale deferred free and we must not
+     * clobber the current state. */
+    if (server.repl_state == REPL_STATE_CONNECTED && server.primary_host) {
+        server.repl_state = REPL_STATE_CONNECT;
+        server.repl_down_since = server.unixtime;
+    } else if (server.repl_state == REPL_STATE_CONNECTED) {
+        /* primary_host is NULL: deliberate unset in progress. */
+        server.repl_state = REPL_STATE_NONE;
+        server.repl_down_since = server.unixtime;
+    }
+    /* Any other repl_state means the state machine already moved on
+     * (e.g. REPL_STATE_CONNECT, CONNECTING, NONE) — leave it untouched. */
+
     /* 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. */
 
     /* Try to re-connect immediately rather than wait for replicationCron
      * waiting 1 second may risk backlog being recycled. */
-    if (server.primary_host) {
+    if (server.repl_state == REPL_STATE_CONNECT && server.primary_host) {
         serverLog(LL_NOTICE, "Reconnecting to PRIMARY %s:%d", server.primary_host, server.primary_port);
         connectWithPrimary();
     }

src/server.c

@@ -2967,6 +2967,7 @@ void initServer(void) {
     server.reply_buffer_resizing_enabled = 1;
     server.client_mem_usage_buckets = NULL;
     server.debug_client_enforce_reply_list = 0;
+    server.debug_force_free_primary_async = 0;
     resetReplicationBuffer();
 
     /* Make sure the locale is set on startup based on the config file. */

src/server.h

@@ -1853,6 +1853,7 @@ struct valkeyServer {
     int enable_module_cmd;                    /* Enable MODULE commands, see PROTECTED_ACTION_ALLOWED_* */
     int enable_debug_assert;                  /* Enable debug asserts */
     int debug_client_enforce_reply_list;      /* Force client to always use the reply list */
+    int debug_force_free_primary_async;       /* Force freeClient on primary to use async path */
     /* Reply construction copy avoidance */
     int min_io_threads_copy_avoid;           /* Minimum number of IO threads for copy avoidance in reply construction */
     int min_string_size_copy_avoid_threaded; /* Minimum bulk string size for copy avoidance in reply construction when IO threads enabled */

src/tls.c

@@ -1603,6 +1603,7 @@ static int connTLSConnect(connection *conn_,
     unsigned char addr_buf[sizeof(struct in6_addr)];
 
     if (conn->c.state != CONN_STATE_NONE) return C_ERR;
+    if (addr == NULL) return C_ERR;
     ERR_clear_error();
 
     /* Check whether addr is an IP address, if not, use the value for Server Name Indication */

tests/unit/wait.tcl

@@ -343,7 +343,7 @@ tags {"wait aof network external:skip"} {
                 set rd [valkey_deferring_client -1]
                 $rd incr foo
                 $rd read
-                $rd waitaof 0 1 0
+                $rd waitaof 0 1 10000
                 wait_for_blocked_client -1
                 $replica replicaof $master_host $master_port
                 assert_equal [$rd read] {1 1}
@@ -532,3 +532,49 @@ start_server {} {
 }
 }
 }
+
+start_server {tags {"wait network external:skip"}} {
+start_server {} {
+start_server {} {
+    set master1 [srv -2 client]
+    set master1_host [srv -2 host]
+    set master1_port [srv -2 port]
+
+    set master2 [srv -1 client]
+    set master2_host [srv -1 host]
+    set master2_port [srv -1 port]
+
+    set replica [srv 0 client]
+
+    test {Repoint replica with deferred freeClient does not double-connect} {
+        $replica replicaof $master1_host $master1_port
+        wait_for_condition 50 100 {
+            [s 0 master_link_status] eq {up}
+        } else {
+            fail "Replication to master1 not established"
+        }
+
+        $replica debug force-free-primary-async 1
+
+        set log_lines [count_log_lines 0]
+        $replica replicaof $master2_host $master2_port
+        wait_for_condition 50 200 {
+            [s 0 master_link_status] eq {up}
+        } else {
+            fail "Replication to master2 not established after repoint"
+        }
+        after 2000
+
+        set logfile [srv 0 stdout]
+        set lines [split [exec tail -n +[expr {$log_lines + 1}] < $logfile] "\n"]
+        set sync_count 0
+        foreach line $lines {
+            if {[string match "*Connecting to PRIMARY $master2_host:$master2_port*" $line]} {
+                incr sync_count
+            }
+        }
+        assert_equal 1 $sync_count
+    }
+}
+}
+}