Cluster: verifyClusterConfigWithData() implemented.

src/cluster.c

@@ -1299,6 +1299,43 @@ void clusterUpdateState(void) {
  * about desynchronizations between the data we have in memory and the
  * cluster configuration. */
 int verifyClusterConfigWithData(void) {
+    int j;
+    int update_config = 0;
+
+    /* Make sure we only have keys in DB0. */
+    for (j = 1; j < server.dbnum; j++) {
+        if (dictSize(server.db[j].dict)) return REDIS_ERR;
+    }
+
+    /* Check that all the slots we see populated memory have a corresponding
+     * entry in the cluster table. Otherwise fix the table. */
+    for (j = 0; j < REDIS_CLUSTER_SLOTS; j++) {
+        if (!countKeysInSlot(j)) continue; /* No keys in this slot. */
+        /* Check if we are assigned to this slot or if we are importing it.
+         * In both cases check the next slot as the configuration makes
+         * sense. */
+        if (server.cluster->slots[j] == server.cluster->myself ||
+            server.cluster->importing_slots_from[j] != NULL) continue;
+
+        /* If we are here data and cluster config don't agree, and we have
+         * slot 'j' populated even if we are not importing it, nor we are
+         * assigned to this slot. Fix this condition. */
+
+        update_config++;
+        /* Case A: slot is unassigned. Take responsability for it. */
+        if (server.cluster->slots[j] == NULL) {
+            redisLog(REDIS_WARNING, "I've keys about slot %d that is "
+                                    "unassigned. Taking responsability "
+                                    "for it.",j);
+            clusterAddSlot(server.cluster->myself,j);
+        } else {
+            redisLog(REDIS_WARNING, "I've keys about slot %d that is "
+                                    "already assigned to a different node. "
+                                    "Setting it in importing state.",j);
+            server.cluster->importing_slots_from[j] = server.cluster->slots[j];
+        }
+    }
+    if (update_config) clusterSaveConfigOrDie();
     return REDIS_OK;
 }