Fix index in AppendEntries Response and logTruncate logic.

Fixed index to use the current index in the logEntry list for the
AppendEntries Response.
In logTruncate, truncate upto the index requested (excluding the index).

Author: srned

src/cluster.c

@@ -396,6 +396,7 @@ void clusterProcessCommand(prezClient *c) {
 
     dictAdd(server.cluster->synced_nodes,
             sdsnewlen(myself->name,PREZ_CLUSTER_NAMELEN),&node_synced);
+    myself->prev_log_index = entry.index;
 
     if(dictSize(server.cluster->nodes) == 1) {
         commit_index = logCurrentIndex();
@@ -789,6 +790,8 @@ void clusterProcessResponseAppendEntries(clusterLink *link,
                         dictSize(server.cluster->nodes));
                 while((de = dictNext(di)) != NULL) {
                     clusterNode *cnode = dictGetVal(de);
+                    prezLog(PREZ_DEBUG,"node, port:%d, prev_index:%lld",
+                            cnode->port,cnode->prev_log_index);
                     log_indices[i] = cnode->prev_log_index;
                 }
                 dictReleaseIterator(di);
@@ -799,7 +802,7 @@ void clusterProcessResponseAppendEntries(clusterLink *link,
                 if (commit_index > server.cluster->commit_index) {
                     logSync();
                     logCommitIndex(commit_index);
-                    prezLog(PREZ_DEBUG, "commit index: %lld", commit_index);
+                    prezLog(PREZ_DEBUG, "AE Response commit index: %lld", commit_index);
                 }
                 zfree(log_indices);
             }
@@ -880,7 +883,7 @@ void clusterSendResponseAppendEntries(clusterLink *link, int ok) {
     prezLog(PREZ_DEBUG, "Sending AppendEntries Response buf:%s, sizeof(clustermsg): %lu, totlen: %d\n", 
             buf, sizeof(clusterMsg), ntohl(hdr->totlen));
     hdr->data.responseappendentries.entries.term = server.cluster->current_term;
-    hdr->data.responseappendentries.entries.index = server.cluster->current_index;
+    hdr->data.responseappendentries.entries.index = logCurrentIndex();
     hdr->data.responseappendentries.entries.commit_index = server.cluster->commit_index;
     hdr->data.responseappendentries.entries.ok = ok;
     clusterSendMessage(link,buf,ntohl(hdr->totlen));

src/log.c

@@ -163,17 +163,20 @@ int logTruncate(long long index, long long term) {
             return PREZ_ERR;
         }
 
-        entry = listNodeValue(listIndex(server.cluster->log_entries, 
-                    index - server.cluster->start_index-1));
-        ftruncate(server.cluster->log_fd, entry->position);
-        server.cluster->log_current_size = entry->position;
-        listRewind(server.cluster->log_entries, &li);
-        li.next = listIndex(server.cluster->log_entries,
-                index - server.cluster->start_index-1);
-        while ((ln = listNext(&li)) != NULL) {
-            logEntryNode *le = listNodeValue(ln);
-            listDelNode(server.cluster->log_entries,ln);
-            zfree(le);
+        if (index < server.cluster->start_index +
+                listLength(server.cluster->log_entries)) {
+            entry = listNodeValue(listIndex(server.cluster->log_entries,
+                        index - server.cluster->start_index-1));
+            ftruncate(server.cluster->log_fd, entry->position);
+            server.cluster->log_current_size = entry->position;
+            listRewind(server.cluster->log_entries, &li);
+            li.next = listIndex(server.cluster->log_entries,
+                    index - server.cluster->start_index-1);
+            while ((ln = listNext(&li)) != NULL) {
+                logEntryNode *le = listNodeValue(ln);
+                listDelNode(server.cluster->log_entries,ln);
+                zfree(le);
+            }
         }
     }
     return PREZ_OK;
@@ -247,11 +250,13 @@ int logWriteEntry(logEntry e) {
     //memcpy(&en->log_entry,&e,sizeof(logEntry));
     en->position = server.cluster->log_current_size;
     server.cluster->log_current_size += sdslen(buf);
-    prezLog(PREZ_DEBUG,"en term:%lld/%lld index:%lld, %lu", 
+    prezLog(PREZ_DEBUG,"Add to List: en term:%lld/%lld index:%lld, %lu",
             en->log_entry.term,
             e.term,
             en->log_entry.index, sizeof(en));
     listAddNodeTail(server.cluster->log_entries,en);
+    prezLog(PREZ_DEBUG,"listLength:%lu",
+            listLength(server.cluster->log_entries));
 
     return PREZ_OK;
 }
@@ -310,7 +315,10 @@ long long logCurrentIndex(void) {
     ln = listIndex(server.cluster->log_entries, -1);
     if (ln) {
         entry = ln->value;
+        prezLog(PREZ_DEBUG,"currentIndex:%lld",
+                entry->log_entry.index);
         return(entry->log_entry.index);
     }
+    prezLog(PREZ_DEBUG,"currentIndex:0");
     return 0;
 }