[backport] Backport sweep for 9.2

.github/workflows/daily.yml

@@ -1879,7 +1879,7 @@ jobs:
     strategy:
       fail-fast: false
       matrix:
-        os: [macos-14]
+        os: [macos-15]
     runs-on: ${{ matrix.os }}
     if: |
       (github.event_name == 'workflow_call' || github.event_name == 'workflow_dispatch' ||

src/bgiteration.c

@@ -26,12 +26,9 @@ static bool isScriptCallWriteCmd(struct serverCommand *cmd) {
  * write commands, blocking them from modifying the format while the background thread might be
  * operating on the item.
  *  - PFCOUNT - modifies the underlying string (and is replicated!)
- *  - LINDEX/LRANGE/LPOS - may modify quicklist by LZF compress/uncompress */
+ *  - SORT_RO may modify the encoding (#4698) */
 static bool isFormatChangingCommand(struct serverCommand *cmd) {
     return ((cmd->proc == pfcountCommand) ||
-            (cmd->proc == lindexCommand) ||
-            (cmd->proc == lrangeCommand) ||
-            (cmd->proc == lposCommand) ||
             (cmd->proc == sortroCommand));
 }
 
@@ -406,6 +403,7 @@ struct bgIterator {
      * deleted key.  keyDelete() removes the dbEntry from early_iterate_entries immediately (step 1)
      * to avoid a dangling pointer.  This flag records if the entry was in early_iterate_entries. */
     bool last_key_delete_was_early; // Used only in main thread during command processing
+    unsigned long added_since_feed; // Items queued outside of feedIterator since its last call
 
     // Variables maintaining runtime statistics
     unsigned long dbentries_queued;         // Updated by main thread
@@ -969,13 +967,16 @@ static sds createEntryString(int dbid, dbEntry *de) {
 
 
 static void feedIterator(bgIterator *it, monotime end_time_us) {
-    unsigned int initial_queue_len = mutexQueueLength(it->items_for_iterator);
-
     /* The queue size dynamically adjusts using an AIMD approach.  If we have left over stuff from
      * the prior call to feedIterator, reduce by half the remaining size.  If the queue ran dry
-     * and we have time left (at the end of this function), additively increase the queue length. */
-    if (initial_queue_len > 2 && it->item_count_target >= initial_queue_len) {
-        it->item_count_target -= initial_queue_len / 2;
+     * and we have time left (at the end of this function), additively increase the queue length.
+     * Items added since the prior call (expedited keys, replication) aren't left over, as the
+     * queue may have run dry before they were added. */
+    unsigned long queue_len = mutexQueueLength(it->items_for_iterator);
+    unsigned long left_over = (queue_len > it->added_since_feed) ? queue_len - it->added_since_feed : 0;
+    it->added_since_feed = 0;
+    if (left_over > 2 && it->item_count_target >= left_over) {
+        it->item_count_target -= left_over / 2;
     }
 
     // Now do some feeding
@@ -1076,7 +1077,7 @@ static void feedIterator(bgIterator *it, monotime end_time_us) {
     }
 
     // Smart logic to dynamically adjust the size of the queue
-    if (initial_queue_len == 0 && have_time && it->item_count_target < BGITER_QUEUE_MAX_LENGTH) {
+    if (left_over == 0 && have_time && it->item_count_target < BGITER_QUEUE_MAX_LENGTH) {
         it->item_count_target += BGITER_QUEUE_INCREASE_INCR;
     }
 }
@@ -1095,6 +1096,7 @@ static bool addEarlyIterationKey(bgIterator *it, dbEntry *earlyEntry, int cur_db
     bgIteratorItem *item = makeDbEntryItem(isClonedEntry ? cloneEntry : earlyEntry, dbid, isClonedEntry);
 
     it->dbentries_queued++;
+    it->added_since_feed++;
     if (isClonedEntry) it->dbentry_clones_queued++;
 
     if (it->barrier_items == 0) {
@@ -1804,6 +1806,7 @@ static void handleSwapdb(int db1, int db2) {
             item->dbid = db1;
             item->u.dbid2 = db2;
             it->swapdb_queued++;
+            it->added_since_feed++;
             it->barrier_items++;
             mutexQueueAdd(it->items_for_iterator, item);
         }
@@ -1850,6 +1853,7 @@ static void handleFlushdb(int dbid) {
             item->type = BGITERATOR_ITEM_FLUSHDB;
             item->dbid = dbid;
             it->flushdb_queued++;
+            it->added_since_feed++;
             it->barrier_items++;
             mutexQueueAdd(it->items_for_iterator, item);
         }
@@ -2036,6 +2040,7 @@ static bgIterator *bgIteratorCreate(const char *name,
     it->terminated = false;
     it->cur_cmd_may_replicate = false;
     it->last_key_delete_was_early = false;
+    it->added_since_feed = 0;
 
     it->dbentries_queued = 0;
     it->dbentries_processed = 0;
@@ -2180,7 +2185,7 @@ bgIteratorItem *bgIteratorRead(bgIterator *it) {
          * Without this, a unit test could get stuck waiting on the completion event because
          * feed won't get invoked.  For production, feed is called regularly from the main thread.
          * Note - this is checking that the exact same thread is used and shouldn't count modules. */
-        if (onServerMainThread()) bgIteration_feedIterators_task(NULL, 0, NULL);
+        if (onServerMainThread()) bgIteration_feedIterators();
     } else {
         it->client_is_active = true;
     }
@@ -2644,6 +2649,7 @@ void bgIteration_handleCommandReplication(int dbid,
             item->u.repl.replication_size = replicationItemSize(item);
             bufferedReplicationBytes += item->u.repl.replication_size;
             it->replication_queued++;
+            it->added_since_feed++;
             mutexQueueAdd(it->items_for_iterator, item);
         }
     } // allIterators loop

src/cluster_legacy.c

@@ -253,6 +253,14 @@ static inline char *clusterLinkGetHumanNodeName(clusterLink *link) {
 static_assert(offsetof(clusterMsg, type) + sizeof(uint16_t) == RCVBUF_MIN_READ_LEN,
               "Incorrect length to read to identify type");
 
+/* Upper bound on the length of an inbound packet, checked from the header so
+ * a peer can't make us buffer a huge packet of a type that is never that big.
+ * A PING carries at most 65535 gossip sections (6.8MB) plus extensions that
+ * are small for any sane human nodename and availability zone. Unknown types
+ * are capped too. Pub/Sub and module messages are exempt: their payload comes
+ * from a client and can legitimately be much larger. */
+#define CLUSTERMSG_MAX_LEN (16 * 1024 * 1024)
+
 /* Cluster nodes hash table, mapping nodes addresses 1.2.3.4:6379 to
  * clusterNode structures. */
 dictType clusterNodesDictType = {
@@ -5043,13 +5051,21 @@ void clusterLinkConnectHandler(connection *conn) {
     serverLog(LL_DEBUG, "Connecting with Node %.40s at %s:%d", node->name, node->ip, node->cport);
 }
 
+static inline int isClusterMsgLengthTooLarge(uint16_t type, uint32_t totlen) {
+    type &= ~CLUSTERMSG_MODIFIER_MASK;
+    if (type == CLUSTERMSG_TYPE_PUBLISH || type == CLUSTERMSG_TYPE_PUBLISHSHARD || type == CLUSTERMSG_TYPE_MODULE)
+        return 0;
+    return totlen > CLUSTERMSG_MAX_LEN;
+}
+
 /* Performs sanity check on the message signature and length depending on the type. */
 static inline int isClusterMsgSignatureAndLengthValid(clusterMsgHeader *hdr) {
     if (memcmp(hdr->sig, "RCmb", 4) != 0) return 0;
     uint16_t type = ntohs(hdr->type);
     uint32_t totlen = ntohl(hdr->totlen);
     uint32_t minlen = IS_LIGHT_MESSAGE(type) ? CLUSTERMSG_LIGHT_MIN_LEN : CLUSTERMSG_MIN_LEN;
     if (totlen < minlen) return 0;
+    if (isClusterMsgLengthTooLarge(type, totlen)) return 0;
     return 1;
 }
 
@@ -5093,7 +5109,7 @@ void clusterFindCompletePackets(char *rcvbuf,
         uint16_t type = ntohs(hdr->type);
         uint32_t minlen = IS_LIGHT_MESSAGE(type) ? CLUSTERMSG_LIGHT_MIN_LEN : CLUSTERMSG_MIN_LEN;
 
-        if (totlen < minlen) {
+        if (totlen < minlen || isClusterMsgLengthTooLarge(type, totlen)) {
             *complete_bytes = offset; /* preserve any valid prefix already scanned */
             *result = CLUSTER_IO_BAD_LENGTH;
             return;

src/commands.def

@@ -9655,7 +9655,7 @@ struct COMMAND_ARG SINTERSTORE_Args[] = {
 #ifndef SKIP_CMD_HISTORY_TABLE
 /* SISMEMBER history */
 commandHistory SISMEMBER_History[] = {
-{"9.1.0","Added the `XX` options."},
+{"9.2.0","Added the `XX` option."},
 };
 #endif
 
@@ -9675,7 +9675,7 @@ keySpec SISMEMBER_Keyspecs[1] = {
 struct COMMAND_ARG SISMEMBER_Args[] = {
 {MAKE_ARG("key",ARG_TYPE_KEY,0,NULL,NULL,NULL,CMD_ARG_NONE,0,NULL)},
 {MAKE_ARG("member",ARG_TYPE_STRING,-1,NULL,NULL,NULL,CMD_ARG_NONE,0,NULL)},
-{MAKE_ARG("xx",ARG_TYPE_PURE_TOKEN,-1,"XX",NULL,"9.1.0",CMD_ARG_OPTIONAL,0,NULL)},
+{MAKE_ARG("xx",ARG_TYPE_PURE_TOKEN,-1,"XX",NULL,"9.2.0",CMD_ARG_OPTIONAL,0,NULL)},
 };
 
 /********** SMEMBERS ********************/

src/commands/cluster-getslotmigrations.json

@@ -19,7 +19,7 @@
                 "type": "object",
                 "additionalProperties": false,
                 "properties": {
-                    "link_name": {
+                    "name": {
                         "type": "string",
                         "description": "A 40-byte randomly generated hex string representing the migration.",
                         "pattern": "^[0-9a-fA-F]{40}$"
@@ -72,6 +72,11 @@
                     "cow_size": {
                         "description": "Copy on write bytes during slot migration fork.",
                         "type": "integer"
+                    },
+                    "remaining_repl_size": {
+                        "description": "Bytes pending in the source slot export client's reply buffer. Zero for imports or when no export client is attached.",
+                        "type": "integer",
+                        "minimum": 0
                     }
                 }
             }

src/commands/sismember.json

@@ -8,8 +8,8 @@
         "function": "sismemberCommand",
         "history": [
             [
-                "9.1.0",
-                "Added the `XX` options."
+                "9.2.0",
+                "Added the `XX` option."
             ]
         ],
         "command_flags": [
@@ -70,7 +70,7 @@
                 "name": "xx",
                 "token": "XX",
                 "type": "pure-token",
-                "since": "9.1.0",
+                "since": "9.2.0",
                 "optional": true
             }
         ]

src/module.c

@@ -4463,8 +4463,19 @@ void VM_CloseKey(ValkeyModuleKey *key) {
     zfree(key);
 }
 
-/* Return the type of the key. If the key pointer is NULL then
- * VALKEYMODULE_KEYTYPE_EMPTY is returned. */
+/* Return the type of the key as one of the following:
+ *
+ * - VALKEYMODULE_KEYTYPE_EMPTY
+ * - VALKEYMODULE_KEYTYPE_STRING
+ * - VALKEYMODULE_KEYTYPE_LIST
+ * - VALKEYMODULE_KEYTYPE_HASH
+ * - VALKEYMODULE_KEYTYPE_SET
+ * - VALKEYMODULE_KEYTYPE_ZSET
+ * - VALKEYMODULE_KEYTYPE_MODULE
+ * - VALKEYMODULE_KEYTYPE_STREAM
+ * - VALKEYMODULE_KEYTYPE_PATH_HASH
+ *
+ * If the key pointer is NULL then VALKEYMODULE_KEYTYPE_EMPTY is returned. */
 int VM_KeyType(ValkeyModuleKey *key) {
     if (key == NULL || key->value == NULL) return VALKEYMODULE_KEYTYPE_EMPTY;
     /* We map between defines so that we are free to change the internal
@@ -9620,14 +9631,16 @@ void moduleReleaseGIL(void) {
  *  - VALKEYMODULE_NOTIFY_EVICTED: Eviction events
  *  - VALKEYMODULE_NOTIFY_STREAM: Stream events
  *  - VALKEYMODULE_NOTIFY_MODULE: Module types events
- *  - VALKEYMODULE_NOTIFY_KEYMISS: Key-miss events
+ *  - VALKEYMODULE_NOTIFY_PATH_HASH: Path hash events
+ *  - VALKEYMODULE_NOTIFY_NEW: New key events
+ *  - VALKEYMODULE_NOTIFY_KEY_MISS: Key-miss events
  *                                Notice, key-miss event is the only type
  *                                of event that is fired from within a read command.
  *                                Performing VM_Call with a write command from within
  *                                this notification is wrong and discourage. It will
  *                                cause the read command that trigger the event to be
  *                                replicated to the AOF/Replica.
- *  - VALKEYMODULE_NOTIFY_ALL: All events (Excluding VALKEYMODULE_NOTIFY_KEYMISS)
+ *  - VALKEYMODULE_NOTIFY_ALL: All events (Excluding VALKEYMODULE_NOTIFY_KEY_MISS and VALKEYMODULE_NOTIFY_NEW)
  *  - VALKEYMODULE_NOTIFY_LOADED: A special notification available only for modules,
  *                               indicates that the key was loaded from persistence.
  *                               Notice, when this event fires, the given key

src/notify.c

@@ -88,8 +88,8 @@ sds keyspaceEventsFlagsToString(int flags) {
         if (flags & NOTIFY_STREAM) res = sdscatlen(res, "t", 1);
         if (flags & NOTIFY_MODULE) res = sdscatlen(res, "d", 1);
         if (flags & NOTIFY_PATH_HASH) res = sdscatlen(res, "p", 1);
-        if (flags & NOTIFY_NEW) res = sdscatlen(res, "n", 1);
     }
+    if (flags & NOTIFY_NEW) res = sdscatlen(res, "n", 1);
     if (flags & NOTIFY_KEYSPACE) res = sdscatlen(res, "K", 1);
     if (flags & NOTIFY_KEYEVENT) res = sdscatlen(res, "E", 1);
     if (flags & NOTIFY_KEY_MISS) res = sdscatlen(res, "m", 1);

src/server.c

@@ -4066,7 +4066,8 @@ static void propagatePendingCommands(void) {
  * Must stay an OR of every condition below - never drop one as an
  * optimization, since that would silently skip real pending work. */
 static inline int hasPostExecutionUnitPendingWork(void) {
-    return moduleHasPostExecUnitJobs() || server.also_propagate.numops || server.busy_module_yield_flags;
+    return moduleHasPostExecUnitJobs() || server.also_propagate.numops || server.busy_module_yield_flags ||
+           bgIteration_iterationActive();
 }
 
 /* Performs operations that should be performed after an execution unit ends.

src/t_stream.c

@@ -477,10 +477,14 @@ int streamAppendItem(stream *s, robj **argv, int64_t numfields, streamID *added_
 
     size_t lp_bytes = 0;      /* Total bytes in the tail listpack. */
     unsigned char *lp = NULL; /* Tail listpack pointer. */
+    /* What the rax holds for the key this call writes, NULL if it holds nothing.
+     * Only used to skip a raxInsert storing a pointer the rax already has. */
+    unsigned char *rax_lp = NULL;
 
     if (!raxEOF(&ri)) {
         /* Get a reference to the tail node listpack. */
         lp = ri.data;
+        rax_lp = ri.data;
         lp_bytes = lpBytes(lp);
     }
     raxStop(&ri);
@@ -542,7 +546,8 @@ int streamAppendItem(stream *s, robj **argv, int64_t numfields, streamID *added_
         if (new_node) {
             /* Shrink extra pre-allocated memory */
             lp = lpShrinkToFit(lp);
-            if (ri.data != lp) raxInsert(s->rax, ri.key, ri.key_len, lp, NULL);
+            if (rax_lp != lp) raxInsert(s->rax, ri.key, ri.key_len, lp, NULL);
+            rax_lp = NULL; /* the rest of this call writes rax_key, a new key */
             lp = NULL;
         }
     }
@@ -570,7 +575,6 @@ int streamAppendItem(stream *s, robj **argv, int64_t numfields, streamID *added_
             lp = lpAppend(lp, (unsigned char *)field, sdslen(field));
         }
         lp = lpAppendInteger(lp, 0); /* primary entry zero terminator. */
-        raxInsert(s->rax, (unsigned char *)&rax_key, sizeof(rax_key), lp, NULL);
         /* The first entry we insert, has obviously the same fields of the
          * primary entry. */
         flags |= STREAM_ITEM_FLAG_SAMEFIELDS;
@@ -651,7 +655,7 @@ int streamAppendItem(stream *s, robj **argv, int64_t numfields, streamID *added_
     lp = lpAppendInteger(lp, lp_count);
 
     /* Insert back into the tree in order to update the listpack pointer. */
-    if (ri.data != lp) raxInsert(s->rax, (unsigned char *)&rax_key, sizeof(rax_key), lp, NULL);
+    if (rax_lp != lp) raxInsert(s->rax, (unsigned char *)&rax_key, sizeof(rax_key), lp, NULL);
     s->length++;
     s->entries_added++;
     s->last_id = id;
@@ -1464,7 +1468,14 @@ void streamReplyWithCGLag(client *c, stream *s, streamCG *cg) {
     } else if (cg->entries_read != SCG_INVALID_ENTRIES_READ && !streamRangeHasTombstones(s, &cg->last_id, NULL)) {
         /* No fragmentation ahead means that the group's logical reads counter
          * is valid for performing the lag calculation. */
-        lag = (long long)s->entries_added - cg->entries_read;
+        if (!s->length || streamCompareID(&cg->last_id, &s->first_id) < 0) {
+            /* The group is positioned before the first entry, so entries it
+             * hasn't read may have been trimmed (XTRIM doesn't update
+             * max_deleted_entry_id). Everything left in the stream is unread. */
+            lag = (long long)s->length;
+        } else {
+            lag = (long long)s->entries_added - cg->entries_read;
+        }
         valid = 1;
     } else {
         /* Attempt to retrieve the group's last ID logical read counter. */

src/unit/test_bgiteration.cpp

@@ -3220,3 +3220,37 @@ TEST_F(BgIterationTest, checkNoKeysWriteIsReplicated) {
     expectReadKeySequence(it, 3, LAST_ITEM);
     expectReadComplete(it);
 }
+
+
+// During a save, a write to an unsaved key queues that key ahead of the scan.  Expedited keys
+//  waiting at a feed must not stop the queue target from growing.
+TEST_F(BgIterationTest, queueTargetGrowsWithExpeditedKeysWaiting) {
+    // Start a consistent save, as BGSAVE does.  Capture its feed timer so the test runs the feed.
+    aeTimeProc *feed = NULL;
+    EXPECT_CALL(mock, aeCreateTimeEvent(_, _, _, _, _)).WillOnce(DoAll(SaveArg<2>(&feed), Return(0)));
+    bgIterator *it = bgIteratorCreateFullScanIter("iter", BGITERATOR_CONSISTENCY_START, NULL,
+                                                  iteratorCleanupFn, PRIVDATA);
+    ASSERT_NE(feed, nullptr);
+
+    // Before anything is fed, write 3 keys.  All are expedited into the queue.
+    for (int item = 0; item < 3; item++) {
+        client *writer = getWriteClient(item, "xxx");
+        simulateBlockedWrite(writer);
+        freeTestClient(writer);
+    }
+
+    // The client reads one of them.  (The first read doesn't run a feed.)
+    bgIteratorRead(it);
+
+    bgIteratorStatus status;
+    bgIteratorGetStatus(it, &status);
+    ASSERT_EQ(status.queue_length, 2u);
+    unsigned long target = status.queue_length_target;
+
+    // Run one feed.  Everything in the queue was added since the last feed, so the target must grow.
+    feed(NULL, 0, NULL);
+    bgIteratorGetStatus(it, &status);
+    EXPECT_GT(status.queue_length_target, target);
+
+    expectAnythingCleanup(it);
+}

src/valkey-cli.c

@@ -10002,6 +10002,7 @@ void testHint(const char *input) {
 
     sds hint = getHintForInput(input);
     printf("%s\n", hint);
+    sdsfree(hint);
     exit(0);
 }