Large Mailbox Append Fix

When syncing a large mailbox it used to be that sync_client would
stream the entire thing, and sync_server would run our of memory.

Now sync_client only sends config_sync_batch_size number of
messages before committing.  This is all well and good until you
find that it's still setting last_uid to the full value on the
source mailbox.

So next time, sync_server chooses sync_combine_commit instead of
sync_append_commit.  This rewrites the entire index and cache file
and causes our servers to become very sad under the giant IO hit
that flushes every cache.

The attached patch creates yet another bespoke linked-list
implementation and generates a list of config_sync_batch_size
messages to sync, then streams them all at once with the the
last_uid set to the UID of the final message in this batch.
At the end, it sets last_uid to the mailbox's last_uid instead.
Index: cyrus-imapd-2.3.9/imap/sync_client.c
===================================================================
--- cyrus-imapd-2.3.9.orig/imap/sync_client.c	2007-09-18 02:56:26.000000000 -0400
+++ cyrus-imapd-2.3.9/imap/sync_client.c	2007-09-19 01:49:53.000000000 -0400
@@ -1229,6 +1229,58 @@
     return(r);
 }
 
+static int index_list_work(struct mailbox *mailbox, struct sync_index_list *index_list)
+{
+    struct sync_index *index;
+    int r = 0;
+    int c = ' ';
+    static struct buf token;   /* BSS */
+
+    if (index_list->count == 0) {
+        sync_index_list_free(&index_list);
+        return(r);
+    }
+
+    prot_printf(toserver, "UPLOAD %lu %lu",
+                index_list->last_uid, mailbox->last_appenddate); 
+
+    for (index = index_list->head; index; index = index->next) {
+        r = upload_message_work(mailbox, index->msgno, &(index->record));
+	      if (r) break;
+    }
+
+    sync_index_list_free(&index_list);
+
+    prot_printf(toserver, "\r\n");
+    prot_flush(toserver);
+
+    if (r) {
+        sync_parse_code("UPLOAD", fromserver, SYNC_PARSE_EAT_OKLINE, NULL);
+        return(r);
+    }
+    r = sync_parse_code("UPLOAD", fromserver, SYNC_PARSE_NOEAT_OKLINE, NULL);
+    if (r) return(r);
+
+    if ((c = getword(fromserver, &token)) != ' ') {
+        eatline(fromserver, c);
+        syslog(LOG_ERR, "Garbage on Upload response");
+        return(IMAP_PROTOCOL_ERROR);
+    }
+    eatline(fromserver, c);
+
+    /* Clear out msgid_on_server list if server restarted */
+    if (!strcmp(token.s, "[RESTART]")) {
+        int hash_size = msgid_onserver->hash_size;
+
+        sync_msgid_list_free(&msgid_onserver);
+        msgid_onserver = sync_msgid_list_create(hash_size);
+
+        syslog(LOG_INFO, "UPLOAD: received RESTART");
+    }
+
+    return (0);
+}
+
 static int upload_messages_list(struct mailbox *mailbox,
 				struct sync_msg_list *list)
 {
@@ -1236,9 +1288,7 @@
     int r = 0;
     struct index_record record;
     struct sync_msg *msg;
-    int count;
-    int c = ' ';
-    static struct buf token;   /* BSS */
+    struct sync_index_list *index_list;
     int max_count = config_getint(IMAPOPT_SYNC_BATCH_SIZE);
 
     if (max_count <= 0) max_count = INT_MAX;
@@ -1249,10 +1299,9 @@
         return(IMAP_IOERROR);
     }
 
-repeatupload:
-
+    index_list = sync_index_list_create();
     msg = list->head;
-    for (count = 0; count < max_count && msgno <= mailbox->exists ; msgno++) {
+    for (msgno = 1; msgno <= mailbox->exists ; msgno++) {
         r = mailbox_read_index_record(mailbox, msgno, &record);
 
         if (r) {
@@ -1285,55 +1334,23 @@
             continue;
         }
 
-        if (count++ == 0)
-            prot_printf(toserver, "UPLOAD %lu %lu",
-                     mailbox->last_uid, mailbox->last_appenddate); 
-
         /* Message with this UUID exists on client but not server */
-        if ((r=upload_message_work(mailbox, msgno, &record)))
-	    break;
-
-        if (msg && (msg->uid == record.uid))  /* Overwritten on server */
-            msg = msg->next;
-    }
-
-    if (count == 0)
-        return(r);
-
-    prot_printf(toserver, "\r\n"); 
-    prot_flush(toserver);
-
-    if (r) {
-        sync_parse_code("UPLOAD", fromserver, SYNC_PARSE_EAT_OKLINE, NULL);
-        return(r);
-    }
-    r = sync_parse_code("UPLOAD", fromserver, SYNC_PARSE_NOEAT_OKLINE, NULL);
-    if (r) return(r);
+        sync_index_list_add(index_list, msgno, &record);
 
-    if ((c = getword(fromserver, &token)) != ' ') {
-        eatline(fromserver, c);
-        syslog(LOG_ERR, "Garbage on Upload response");
-        return(IMAP_PROTOCOL_ERROR);
-    }
-    eatline(fromserver, c);
-
-    /* Clear out msgid_on_server list if server restarted */
-    if (!strcmp(token.s, "[RESTART]")) {
-        int hash_size = msgid_onserver->hash_size;
-
-        sync_msgid_list_free(&msgid_onserver);
-        msgid_onserver = sync_msgid_list_create(hash_size);
-
-	syslog(LOG_INFO, "UPLOAD: received RESTART");
+        /* don't push too many messages at once */
+        if (index_list->count >= max_count) {
+            syslog(LOG_NOTICE, "Hit upload limit %d at UID %lu for %s, sending",
+                   max_count, index_list->last_uid, mailbox->name);
+            r = index_list_work(mailbox, index_list);
+            if (r) return (r);
+            index_list = sync_index_list_create();
+        }
     }
 
-    /* don't overload the server with too many uploads at once! */
-    if (count >= max_count) {
-	syslog(LOG_INFO, "UPLOAD: hit %d uploads at msgno %d", count, msgno);
-	goto repeatupload;
-    }
+    index_list->last_uid = mailbox->last_uid;
+    r = index_list_work(mailbox, index_list);
 
-    return(0);
+    return(r);
 }
 
 static int upload_messages_from(struct mailbox *mailbox,
Index: cyrus-imapd-2.3.9/imap/sync_support.c
===================================================================
--- cyrus-imapd-2.3.9.orig/imap/sync_support.c	2007-09-18 02:50:20.000000000 -0400
+++ cyrus-imapd-2.3.9/imap/sync_support.c	2007-09-18 03:16:26.000000000 -0400
@@ -413,6 +413,56 @@
 
 /* ====================================================================== */
 
+/* sync_index stuff */
+
+struct sync_index_list *sync_index_list_create()
+{
+    struct sync_index_list *l = xzmalloc(sizeof (struct sync_index_list));
+
+    l->head     = NULL;
+    l->tail     = NULL;
+    l->count    = 0;
+    l->last_uid = 0;
+
+    return l;
+}
+
+void sync_index_list_add(struct sync_index_list *l,
+        unsigned long msgno, struct index_record *record)
+{
+    struct sync_index *result = xzmalloc(sizeof(struct sync_index));
+
+    result->msgno = msgno;
+    memcpy(&(result->record), record, sizeof(struct index_record));
+
+    if (l->tail)
+        l->tail = l->tail->next = result;
+    else
+        l->head = l->tail = result;
+
+    l->count++;
+    l->last_uid = record->uid;
+}
+
+void sync_index_list_free(struct sync_index_list **lp)
+{
+    struct sync_index_list *l = *lp;
+    struct sync_index *current, *next;
+
+    current = l->head;
+    while (current) {
+        next = current->next;
+        free(current);
+        current = next;
+    }
+    free(l);
+
+    *lp = NULL;
+}
+
+
+/* ====================================================================== */
+
 /* sync_msg stuff */
 
 struct sync_msg_list *sync_msg_list_create(char **flagname,
Index: cyrus-imapd-2.3.9/imap/sync_support.h
===================================================================
--- cyrus-imapd-2.3.9.orig/imap/sync_support.h	2007-09-18 02:50:20.000000000 -0400
+++ cyrus-imapd-2.3.9/imap/sync_support.h	2007-09-18 03:10:15.000000000 -0400
@@ -48,6 +48,7 @@
 #define INCLUDED_SYNC_SUPPORT_H
 
 #include "prot.h"
+#include "mailbox.h"
 
 #define SYNC_MSGID_LIST_HASH_SIZE        (65536)
 #define SYNC_MESSAGE_LIST_HASH_SIZE      (65536)
@@ -97,6 +98,29 @@
 
 /* ====================================================================== */
 
+/* sync_index_list records index records for upload */
+
+struct sync_index {
+    struct sync_index *next;
+    struct index_record record;
+    unsigned long msgno;
+};
+
+struct sync_index_list {
+    struct sync_index *head, *tail;
+    unsigned long count;
+    unsigned long last_uid;
+};
+
+struct sync_index_list *sync_index_list_create(void);
+
+void sync_index_list_add(struct sync_index_list *l,
+             unsigned long msgno, struct index_record *record);
+
+void sync_index_list_free(struct sync_index_list **lp);
+
+/* ====================================================================== */
+
 /* sync_msg_list records message lists in client */
 
 struct sync_msg {