Merge remote-tracking branch 'origin/unstable' into vectors_distance_bin.popcount

src/cluster.h

@@ -107,7 +107,6 @@ clusterNode *getMyClusterNode(void);
 char *getMyClusterId(void);
 int getClusterSize(void);
 int getMyShardSlotCount(void);
-int handleDebugClusterCommand(client *c);
 int clusterNodePending(clusterNode  *node);
 char **getClusterNodesList(size_t *numnodes);
 int clusterNodeIsMaster(clusterNode *n);

src/cluster_legacy.c

@@ -2602,7 +2602,7 @@ uint32_t writePingExt(clusterMsg *hdr, int gossipcount)  {
     totlen += getShardIdPingExtSize();
     extensions++;
 
-    /* Populate insternal secret */
+    /* Populate internal secret */
     if (cursor != NULL) {
         clusterMsgPingExtInternalSecret *ext = preparePingExt(cursor, CLUSTERMSG_EXT_TYPE_INTERNALSECRET, getInternalSecretPingExtSize());
         memcpy(ext->internal_secret, server.cluster->internal_secret, CLUSTER_INTERNALSECRETLEN);

src/cluster_legacy.h

@@ -231,7 +231,7 @@ typedef struct {
     uint16_t ver;       /* Protocol version, currently set to 1. */
     uint16_t port;      /* Primary port number (TCP or TLS). */
     uint16_t type;      /* Message type */
-    uint16_t count;     /* Only used for some kind of messages. */
+    uint16_t count;     /* Only used for some kinds of messages. */
     uint64_t currentEpoch;  /* The epoch accordingly to the sending node. */
     uint64_t configEpoch;   /* The config epoch if it's a master, or the last
                                epoch advertised by its master if it is a
@@ -258,8 +258,8 @@ typedef struct {
  * especially during cluster rolling upgrades.
  *
  * Therefore, fields in this struct should remain at the same offset from
- * release to release. The static asserts below ensures that incompatible
- * changes in clusterMsg be caught at compile time.
+ * release to release. The static asserts below ensure that incompatible
+ * changes in clusterMsg are caught at compile time.
  */
 
 static_assert(offsetof(clusterMsg, sig) == 0, "unexpected field offset");

src/config.c

@@ -2688,6 +2688,7 @@ static int applyTLSPort(const char **err) {
     listener->bindaddr_count = server.bindaddr_count;
     listener->port = server.tls_port;
     listener->ct = connectionByType(CONN_TYPE_TLS);
+    clusterUpdateMyselfAnnouncedPorts();
     if (changeListener(listener) == C_ERR) {
         *err = "Unable to listen on this port. Check server logs.";
         return 0;

src/t_stream.c

@@ -1695,7 +1695,10 @@ size_t streamReplyWithRange(client *c, stream *s, streamID *start, streamID *end
     while(streamIteratorGetID(&si,&id,&numfields)) {
         /* Update the group last_id if needed. */
         if (group && streamCompareID(&id,&group->last_id) > 0) {
-            if (group->entries_read != SCG_INVALID_ENTRIES_READ && !streamRangeHasTombstones(s,&group->last_id,NULL)) {
+            if (group->entries_read != SCG_INVALID_ENTRIES_READ &&
+                streamCompareID(&group->last_id, &s->first_id) >= 0 &&
+                !streamRangeHasTombstones(s,&group->last_id,NULL))
+            {
                 /* A valid counter and no tombstones between the group's last-delivered-id
                  * and the stream's last-generated-id mean we can increment the read counter
                  * to keep tracking the group's progress. */

tests/unit/cluster/announced-endpoints.tcl

@@ -49,12 +49,21 @@ start_cluster 2 2 {tags {external:skip cluster}} {
     }
 
     test "CONFIG SET port updates cluster-announced port" {
+        set count [expr [llength $::servers] + 1]
         # Get the original port and change to new_port
-        set orig_port [lindex [R 0 config get port] 1]
+        if {$::tls} {
+            set orig_port [lindex [R 0 config get tls-port] 1]
+        } else {
+            set orig_port [lindex [R 0 config get port] 1]
+        }
         assert {$orig_port != ""}
+        set new_port [find_available_port $orig_port $count]
 
-        set new_port [find_available_port $baseport $count]
-        R 0 config set port $new_port
+        if {$::tls} {
+            R 0 config set tls-port $new_port
+        } else {
+            R 0 config set port $new_port
+        }
 
         # Verify that the new port appears in the output of cluster slots
         wait_for_condition 50 100 {

tests/unit/type/stream-cgroups.tcl

@@ -1300,6 +1300,19 @@ start_server {
         assert_equal [dict get $group entries-read] 1
         assert_equal [dict get $group lag] 1
 
+        # When all the entries are read, the lag is always 0.
+        r XREADGROUP GROUP mygroup alice STREAMS x >
+        set reply [r XINFO STREAM x FULL]
+        set group [lindex [dict get $reply groups] 0]
+        assert_equal [dict get $group entries-read] 5
+        assert_equal [dict get $group lag] 0
+
+        r XADD x 6-0 data f
+        set reply [r XINFO STREAM x FULL]
+        set group [lindex [dict get $reply groups] 0]
+        assert_equal [dict get $group entries-read] 5
+        assert_equal [dict get $group lag] 1
+
         # When all the entries were deleted, the lag is always 0.
         r XTRIM x MAXLEN 0
         set reply [r XINFO STREAM x FULL]