Persistence - Remove Unowned Keys

For cluster, after startup loading, remove keys
that shouldn't be served by this server based on slot
assignment of a cluster.

Signed-off-by: Liang Tang <[email protected]>

src/cluster.c

@@ -108,6 +108,22 @@ ConnectionType *connTypeOfCluster(void) {
     return connectionTypeTcp();
 }
 
+/* delete unowned keys from database at server start up after loading persistence files. */
+void delUnownedKeys(void) {
+    if (!server.cluster_enabled) {
+        return;
+    }
+    clusterNode *primary = clusterNodeGetPrimary(server.cluster->myself);
+    for (unsigned int i = 0; i < CLUSTER_SLOTS; i++) {
+        if (server.cluster->slots[i] != primary) {
+            unsigned int deleted = delKeysInSlot(i);
+            if (deleted > 0) {
+                serverLog(LL_VERBOSE, "Deleted %u keys from unowned slot: %d after loading RDB", deleted, i);
+            }
+        }
+    }
+}
+
 /* -----------------------------------------------------------------------------
  * DUMP, RESTORE and MIGRATE commands
  * -------------------------------------------------------------------------- */

src/cluster.h

@@ -128,4 +128,5 @@ int isNodeAvailable(clusterNode *node);
 long long getNodeReplicationOffset(clusterNode *node);
 sds aggregateClientOutputBuffer(client *c);
 void resetClusterStats(void);
+unsigned int delKeysInSlot(unsigned int);
 #endif /* __CLUSTER_H */

src/server.c

@@ -6930,7 +6930,7 @@ int main(int argc, char **argv) {
         if (server.cluster_enabled) {
             serverAssert(verifyClusterConfigWithData() == C_OK);
         }
-
+        delUnownedKeys();
         for (j = 0; j < CONN_TYPE_MAX; j++) {
             connListener *listener = &server.listeners[j];
             if (listener->ct == NULL) continue;

src/server.h

@@ -3678,6 +3678,8 @@ unsigned long LFUDecrAndReturn(robj *o);
 int performEvictions(void);
 void startEvictionTimeProc(void);
 
+void delUnownedKeys(void);
+
 /* Keys hashing / comparison functions for dict.c hash tables. */
 uint64_t dictSdsHash(const void *key);
 uint64_t dictSdsCaseHash(const void *key);