1 files changed, 292 insertions, 0 deletions

diff --git a/examples/redis-unstable/tests/unit/cluster/links.tcl b/examples/redis-unstable/tests/unit/cluster/links.tcl
new file mode 100644
index 0000000..a202c37
--- /dev/null
+++ b/examples/redis-unstable/tests/unit/cluster/links.tcl
@@ -0,0 +1,292 @@
+proc get_links_with_peer {this_instance_id peer_nodename} {
+    set links [R $this_instance_id cluster links]
+    set links_with_peer {}
+    foreach l $links {
+        if {[dict get $l node] eq $peer_nodename} {
+            lappend links_with_peer $l
+        }
+    }
+    return $links_with_peer
+}
+
+# Return the entry in CLUSTER LINKS output by instance identified by `this_instance_id` that
+# corresponds to the link established toward a peer identified by `peer_nodename`
+proc get_link_to_peer {this_instance_id peer_nodename} {
+    set links_with_peer [get_links_with_peer $this_instance_id $peer_nodename]
+    foreach l $links_with_peer {
+        if {[dict get $l direction] eq "to"} {
+            return $l
+        }
+    }
+    return {}
+}
+
+# Return the entry in CLUSTER LINKS output by instance identified by `this_instance_id` that
+# corresponds to the link accepted from a peer identified by `peer_nodename`
+proc get_link_from_peer {this_instance_id peer_nodename} {
+    set links_with_peer [get_links_with_peer $this_instance_id $peer_nodename]
+    foreach l $links_with_peer {
+        if {[dict get $l direction] eq "from"} {
+            return $l
+        }
+    }
+    return {}
+}
+
+# Reset cluster links to their original state
+proc reset_links {id} {
+    set limit [lindex [R $id CONFIG get cluster-link-sendbuf-limit] 1]
+
+    # Set a 1 byte limit and wait for cluster cron to run
+    # (executes every 100ms) and terminate links
+    R $id CONFIG SET cluster-link-sendbuf-limit 1
+    after 150
+
+    # Reset limit
+    R $id CONFIG SET cluster-link-sendbuf-limit $limit
+
+    # Wait until the cluster links come back up for each node
+    wait_for_condition 50 100 {
+        [number_of_links $id] == [expr [number_of_peers $id] * 2]
+    } else {
+        fail "Cluster links did not come back up"
+    }
+}
+
+proc number_of_peers {id} {
+    expr [llength $::servers] - 1
+}
+
+proc number_of_links {id} {
+    llength [R $id cluster links]
+}
+
+proc publish_messages {server num_msgs msg_size} {
+    for {set i 0} {$i < $num_msgs} {incr i} {
+        $server PUBLISH channel [string repeat "x" $msg_size]
+    }
+}
+
+start_cluster 1 2 {tags {external:skip cluster}} {
+    set primary_id 0
+    set replica1_id 1
+
+    set primary [Rn $primary_id]
+    set replica1 [Rn $replica1_id]
+
+    test "Broadcast message across a cluster shard while a cluster link is down" {
+        set replica1_node_id [$replica1 CLUSTER MYID]
+
+        set channelname ch3
+
+        # subscribe on replica1
+        set subscribeclient1 [redis_deferring_client -1]
+        $subscribeclient1 deferred 1
+        $subscribeclient1 SSUBSCRIBE $channelname
+        $subscribeclient1 read
+
+        # subscribe on replica2
+        set subscribeclient2 [redis_deferring_client -2]
+        $subscribeclient2 deferred 1
+        $subscribeclient2 SSUBSCRIBE $channelname
+        $subscribeclient2 read
+
+        # Verify number of links with cluster stable state
+        assert_equal [expr [number_of_peers $primary_id]*2] [number_of_links $primary_id]
+
+        # Disconnect the cluster between primary and replica1 and publish a message.
+        $primary MULTI
+        $primary DEBUG CLUSTERLINK KILL TO $replica1_node_id
+        $primary SPUBLISH $channelname hello
+        set res [$primary EXEC]
+
+        # Verify no client exists on the primary to receive the published message.
+        assert_equal $res {OK 0}
+
+        # Wait for all the cluster links are healthy
+        wait_for_condition 50 100 {
+            [number_of_peers $primary_id]*2 == [number_of_links $primary_id]
+        } else {
+            fail "All peer links couldn't be established"
+        }
+
+        # Publish a message afterwards.
+        $primary SPUBLISH $channelname world
+
+        # Verify replica1 has received only (world) / hello is lost.
+        assert_equal "smessage ch3 world" [$subscribeclient1 read]
+
+        # Verify replica2 has received both messages (hello/world)
+        assert_equal "smessage ch3 hello" [$subscribeclient2 read]
+        assert_equal "smessage ch3 world" [$subscribeclient2 read]
+    } {} {needs:debug}
+}
+
+start_cluster 3 0 {tags {external:skip cluster}} {
+    test "Each node has two links with each peer" {
+        for {set id 0} {$id < [llength $::servers]} {incr id} {
+            # Assert that from point of view of each node, there are two links for
+            # each peer. It might take a while for cluster to stabilize so wait up
+            # to 5 seconds.
+            wait_for_condition 50 100 {
+                [number_of_peers $id]*2 == [number_of_links $id]
+            } else {
+                assert_equal [expr [number_of_peers $id]*2] [number_of_links $id]
+            }
+
+            set nodes [get_cluster_nodes $id]
+            set links [R $id cluster links]
+
+            # For each peer there should be exactly one
+            # link "to" it and one link "from" it.
+            foreach n $nodes {
+                if {[cluster_has_flag $n myself]} continue
+                set peer [dict get $n id]
+                set to 0
+                set from 0
+                foreach l $links {
+                    if {[dict get $l node] eq $peer} {
+                        if {[dict get $l direction] eq "to"} {
+                            incr to
+                        } elseif {[dict get $l direction] eq "from"} {
+                            incr from
+                        }
+                    }
+                }
+                assert {$to eq 1}
+                assert {$from eq 1}
+            }
+        }
+    }
+
+    test {Validate cluster links format} {
+        set lines [R 0 cluster links]
+        foreach l $lines {
+            if {$l eq {}} continue
+            assert_equal [llength $l] 12
+            assert_equal 1 [dict exists $l "direction"]
+            assert_equal 1 [dict exists $l "node"]
+            assert_equal 1 [dict exists $l "create-time"]
+            assert_equal 1 [dict exists $l "events"]
+            assert_equal 1 [dict exists $l "send-buffer-allocated"]
+            assert_equal 1 [dict exists $l "send-buffer-used"]
+        }
+    }
+
+    set primary1_id 0
+    set primary2_id 1
+
+    set primary1 [Rn $primary1_id]
+    set primary2 [Rn $primary2_id]
+
+    test "Disconnect link when send buffer limit reached" {
+        # On primary1, set timeout to 1 hour so links won't get disconnected due to timeouts
+        set oldtimeout [lindex [$primary1 CONFIG get cluster-node-timeout] 1]
+        $primary1 CONFIG set cluster-node-timeout [expr 60*60*1000]
+
+        # Get primary1's links with primary2
+        set primary2_name [dict get [cluster_get_myself $primary2_id] id]
+        set orig_link_p1_to_p2 [get_link_to_peer $primary1_id $primary2_name]
+        set orig_link_p1_from_p2 [get_link_from_peer $primary1_id $primary2_name]
+
+        # On primary1, set cluster link send buffer limit to 256KB, which is large enough to not be
+        # overflowed by regular gossip messages but also small enough that it doesn't take too much
+        # memory to overflow it. If it is set too high, Redis may get OOM killed by kernel before this
+        # limit is overflowed in some RAM-limited test environments.
+        set oldlimit [lindex [$primary1 CONFIG get cluster-link-sendbuf-limit] 1]
+        $primary1 CONFIG set cluster-link-sendbuf-limit [expr 256*1024]
+        assert {[CI $primary1_id total_cluster_links_buffer_limit_exceeded] eq 0}
+
+        # To manufacture an ever-growing send buffer from primary1 to primary2,
+        # make primary2 unresponsive.
+        set primary2_pid [srv [expr -1*$primary2_id] pid]
+        pause_process $primary2_pid
+
+        # On primary1, send 128KB Pubsub messages in a loop until the send buffer of the link from
+        # primary1 to primary2 exceeds buffer limit therefore be dropped.
+        # For the send buffer to grow, we need to first exhaust TCP send buffer of primary1 and TCP
+        # receive buffer of primary2 first. The sizes of these two buffers vary by OS, but 100 128KB
+        # messages should be sufficient.
+        set i 0
+        wait_for_condition 100 0 {
+            [catch {incr i} e] == 0 &&
+            [catch {$primary1 publish channel [prepare_value [expr 128*1024]]} e] == 0 &&
+            [catch {after 500} e] == 0 &&
+            [CI $primary1_id total_cluster_links_buffer_limit_exceeded] >= 1
+        } else {
+            fail "Cluster link not freed as expected"
+        }
+
+        # A new link to primary2 should have been recreated
+        set new_link_p1_to_p2 [get_link_to_peer $primary1_id $primary2_name]
+        assert {[dict get $new_link_p1_to_p2 create-time] > [dict get $orig_link_p1_to_p2 create-time]}
+
+        # Link from primary2 should not be affected
+        set same_link_p1_from_p2 [get_link_from_peer $primary1_id $primary2_name]
+        assert {[dict get $same_link_p1_from_p2 create-time] eq [dict get $orig_link_p1_from_p2 create-time]}
+
+        # Revive primary2
+        resume_process $primary2_pid
+
+        # Reset configs on primary1 so config changes don't leak out to other tests
+        $primary1 CONFIG set cluster-node-timeout $oldtimeout
+        $primary1 CONFIG set cluster-link-sendbuf-limit $oldlimit
+
+        reset_links $primary1_id
+    }
+
+    test "Link memory increases with publishes" {
+        set server_id 0
+        set server [Rn $server_id]
+        set msg_size 10000
+        set num_msgs 10
+
+        # Remove any sendbuf limit
+        $primary1 CONFIG set cluster-link-sendbuf-limit 0
+
+        # Publish ~100KB to one of the servers
+        $server MULTI
+        $server INFO memory
+        publish_messages $server $num_msgs $msg_size
+        $server INFO memory
+        set res [$server EXEC]
+
+        set link_mem_before_pubs [getInfoProperty $res mem_cluster_links]
+
+        # Remove the first half of the response string which contains the
+        # first "INFO memory" results and search for the property again
+        set res [string range $res [expr [string length $res] / 2] end]
+        set link_mem_after_pubs [getInfoProperty $res mem_cluster_links]
+        
+        # We expect the memory to have increased by more than
+        # the culmulative size of the publish messages
+        set mem_diff_floor [expr $msg_size * $num_msgs]
+        set mem_diff [expr $link_mem_after_pubs - $link_mem_before_pubs]
+        assert {$mem_diff > $mem_diff_floor}
+
+        # Reset links to ensure no leftover data for the next test
+        reset_links $server_id
+    }
+
+    test "Link memory resets after publish messages flush" {
+        set server [Rn 0]
+        set msg_size 100000
+        set num_msgs 10
+
+        set link_mem_before [status $server mem_cluster_links]
+
+        # Publish ~1MB to one of the servers
+        $server MULTI
+        publish_messages $server $num_msgs $msg_size
+        $server EXEC
+
+        # Wait until the cluster link memory has returned to below the pre-publish value.
+        # We can't guarantee it returns to the exact same value since gossip messages
+        # can cause the values to fluctuate.
+        wait_for_condition 1000 500 {
+            [status $server mem_cluster_links] <= $link_mem_before
+        } else {
+            fail "Cluster link memory did not settle back to expected range"
+        }
+    }
+}