Fix bug in counter recalculation causing infinite loop

Author: Rustam Aliyev

modules/core/src/main/java/com/elasticinbox/core/cassandra/CassandraMessageDAO.java

@@ -41,6 +41,7 @@
 import java.util.UUID;
 
 import me.prettyprint.cassandra.serializers.StringSerializer;
+import me.prettyprint.cassandra.utils.TimeUUIDUtils;
 import me.prettyprint.hector.api.Keyspace;
 import me.prettyprint.hector.api.mutation.Mutator;
 
@@ -63,6 +64,7 @@
 import com.elasticinbox.core.model.Marker;
 import com.elasticinbox.core.model.Message;
 import com.elasticinbox.core.model.ReservedLabels;
+import com.google.common.collect.Iterables;
 
 public final class CassandraMessageDAO extends AbstractMessageDAO implements MessageDAO
 {
@@ -402,24 +404,29 @@ public void purge(final Mailbox mailbox, final Date age)
 	public Labels calculateCounters(final Mailbox mailbox)
 	{
 		Labels labels = new Labels();
-		UUID lastMessage = null;
 		Map<UUID, Message> messages;
 		Map<UUID, UUID> purgeIndex;
 		Set<UUID> deletedMessages = new HashSet<UUID>();
-		
+
+		logger.debug("Recalculating counters for {}", mailbox);
+
 		// get deleted message IDs from purge queue
 		// deleted messages should be excluded during calculation
+		UUID start = TimeUUIDUtils.getUniqueTimeUUIDinMillis();
 		do {
 			purgeIndex = PurgeIndexPersistence.get(
-					mailbox.getId(), new Date(), BatchConstants.BATCH_READS);
+					mailbox.getId(), start, BatchConstants.BATCH_READS);
+
 			deletedMessages.addAll(purgeIndex.values());
+			start = Iterables.getLast(purgeIndex.keySet());
 		}
 		while (purgeIndex.size() >= BatchConstants.BATCH_READS);
 
-		// read messages and calculate label counters
+		// reset start, read messages and calculate label counters
+		start = TimeUUIDUtils.getUniqueTimeUUIDinMillis();
 		do {
 			messages = MessagePersistence.getRange(
-					mailbox.getId(), lastMessage, BatchConstants.BATCH_READS);
+					mailbox.getId(), start, BatchConstants.BATCH_READS);
 
 			for (UUID messageId : messages.keySet())
 			{
@@ -432,7 +439,7 @@ public Labels calculateCounters(final Mailbox mailbox)
 					labels.incrementCounters(labelId, message.getLabelCounters());
 				}
 
-				lastMessage = messageId;
+				start = messageId;
 			}
 		}
 		while (messages.size() >= BatchConstants.BATCH_READS);

modules/core/src/main/java/com/elasticinbox/core/cassandra/persistence/MessagePersistence.java

@@ -165,7 +165,7 @@ public static Map<UUID, Message> getRange(final String mailbox,
 		// set keys, cf, range
 		q.setColumnFamily(CF_METADATA);
 		q.setKey(mailbox);
-		q.setRange(start, null, false, count);
+		q.setRange(start, null, true, count);
 
 		// execute
 		QueryResult<SuperSlice<UUID, String, byte[]>> r = q.execute();

modules/core/src/main/java/com/elasticinbox/core/cassandra/persistence/PurgeIndexPersistence.java

@@ -34,7 +34,7 @@
 
 import java.util.Collection;
 import java.util.Date;
-import java.util.HashMap;
+import java.util.LinkedHashMap;
 import java.util.List;
 import java.util.Map;
 import java.util.UUID;
@@ -101,10 +101,22 @@ public static void add(Mutator<String> mutator, final String mailbox,
 	public static Map<UUID, UUID> get(final String mailbox, final Date age,
 			final int count)
 	{
-		Map<UUID, UUID> messageIds = new HashMap<UUID, UUID>(count);
-
 		UUID start = TimeUUIDUtils.getTimeUUID(age.getTime());
-		
+		return get(mailbox, start, count);
+	}
+
+	/**
+	 * Get all message IDs deleted before given UUID
+	 * 
+	 * @param mailbox
+	 * @param age
+	 * @return
+	 */
+	public static Map<UUID, UUID> get(final String mailbox, final UUID start,
+			final int count)
+	{
+		Map<UUID, UUID> messageIds = new LinkedHashMap<UUID, UUID>(count);
+
 		String key = new StringBuilder(mailbox)
 				.append(LabelIndexPersistence.COMPOSITE_KEY_DELIMITER)
 				.append(PURGE_LABEL_ID).toString();
@@ -130,6 +142,8 @@ public static Map<UUID, UUID> get(final String mailbox, final Date age,
 
 		return messageIds;
 	}
+	
+	
 
 	/**
 	 * Remove all message IDs from purge index deleted before given date