memory store for append/fetch/copy

8 files changed, 884 insertions, 84 deletions

diff --git a/mail/src/leap/mail/imap/account.py b/mail/src/leap/mail/imap/account.py
index ce83079..7641ea8 100644
--- a/mail/src/leap/mail/imap/account.py
+++ b/mail/src/leap/mail/imap/account.py
@@ -48,7 +48,7 @@ class SoledadBackedAccount(WithMsgFields, IndexedDB, MBoxParser):
     selected = None
     closed = False
 
-    def __init__(self, account_name, soledad=None):
+    def __init__(self, account_name, soledad=None, memstore=None):
         """
         Creates a SoledadAccountIndex that keeps track of the mailboxes
         and subscriptions handled by this account.
@@ -57,7 +57,9 @@ class SoledadBackedAccount(WithMsgFields, IndexedDB, MBoxParser):
         :type acct_name: str
 
         :param soledad: a Soledad instance.
-        :param soledad: Soledad
+        :type soledad: Soledad
+        :param memstore: a MemoryStore instance.
+        :type memstore: MemoryStore
         """
         leap_assert(soledad, "Need a soledad instance to initialize")
         leap_assert_type(soledad, Soledad)
@@ -67,6 +69,7 @@ class SoledadBackedAccount(WithMsgFields, IndexedDB, MBoxParser):
 
         self._account_name = self._parse_mailbox_name(account_name)
         self._soledad = soledad
+        self._memstore = memstore
 
         self.initialize_db()
 
@@ -131,7 +134,8 @@ class SoledadBackedAccount(WithMsgFields, IndexedDB, MBoxParser):
         if name not in self.mailboxes:
             raise imap4.MailboxException("No such mailbox: %r" % name)
 
-        return SoledadMailbox(name, soledad=self._soledad)
+        return SoledadMailbox(name, soledad=self._soledad,
+                              memstore=self._memstore)
 
     ##
     ## IAccount
@@ -221,8 +225,7 @@ class SoledadBackedAccount(WithMsgFields, IndexedDB, MBoxParser):
         self.selected = name
 
         return SoledadMailbox(
-            name, rw=readwrite,
-            soledad=self._soledad)
+            name, self._soledad, self._memstore, readwrite)
 
     def delete(self, name, force=False):
         """
diff --git a/mail/src/leap/mail/imap/interfaces.py b/mail/src/leap/mail/imap/interfaces.py
new file mode 100644
index 0000000..585165a
--- /dev/null
+++ b/mail/src/leap/mail/imap/interfaces.py
@@ -0,0 +1,93 @@
+# -*- coding: utf-8 -*-
+# interfaces.py
+# Copyright (C) 2014 LEAP
+#
+# This program is free software: you can redistribute it and/or modify
+# it under the terms of the GNU General Public License as published by
+# the Free Software Foundation, either version 3 of the License, or
+# (at your option) any later version.
+#
+# This program is distributed in the hope that it will be useful,
+# but WITHOUT ANY WARRANTY; without even the implied warranty of
+# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+# GNU General Public License for more details.
+#
+# You should have received a copy of the GNU General Public License
+# along with this program.  If not, see <http://www.gnu.org/licenses/>.
+"""
+Interfaces for the IMAP module.
+"""
+from zope.interface import Interface, Attribute
+
+
+class IMessageContainer(Interface):
+    """
+    I am a container around the different documents that a message
+    is split into.
+    """
+    fdoc = Attribute('The flags document for this message, if any.')
+    hdoc = Attribute('The headers document for this message, if any.')
+    cdocs = Attribute('The dict of content documents for this message, '
+                      'if any.')
+
+    def walk(self):
+        """
+        Return an iterator to the docs for all the parts.
+
+        :rtype: iterator
+        """
+
+
+class IMessageStore(Interface):
+    """
+    I represent a generic storage for LEAP Messages.
+    """
+
+    def create_message(self, mbox, uid, message):
+        """
+        Put the passed message into this IMessageStore.
+
+        :param mbox: the mbox this message belongs.
+        :param uid: the UID that identifies this message in this mailbox.
+        :param message: a IMessageContainer implementor.
+        """
+
+    def put_message(self, mbox, uid, message):
+        """
+        Put the passed message into this IMessageStore.
+
+        :param mbox: the mbox this message belongs.
+        :param uid: the UID that identifies this message in this mailbox.
+        :param message: a IMessageContainer implementor.
+        """
+
+    def remove_message(self, mbox, uid):
+        """
+        Remove the given message from this IMessageStore.
+
+        :param mbox: the mbox this message belongs.
+        :param uid: the UID that identifies this message in this mailbox.
+        """
+
+    def get_message(self, mbox, uid):
+        """
+        Get a IMessageContainer for the given mbox and uid combination.
+
+        :param mbox: the mbox this message belongs.
+        :param uid: the UID that identifies this message in this mailbox.
+        """
+
+
+class IMessageStoreWriter(Interface):
+    """
+    I represent a storage that is able to write its contents to another
+    different IMessageStore.
+    """
+
+    def write_messages(self, store):
+        """
+        Write the documents in this IMessageStore to a different
+        storage. Usually this will be done from a MemoryStorage to a DbStorage.
+
+        :param store: another IMessageStore implementor.
+        """
diff --git a/mail/src/leap/mail/imap/mailbox.py b/mail/src/leap/mail/imap/mailbox.py
index 0131ce0..9babe6b 100644
--- a/mail/src/leap/mail/imap/mailbox.py
+++ b/mail/src/leap/mail/imap/mailbox.py
@@ -37,6 +37,7 @@ from leap.common.events.events_pb2 import IMAP_UNREAD_MAIL
 from leap.common.check import leap_assert, leap_assert_type
 from leap.mail.decorators import deferred
 from leap.mail.imap.fields import WithMsgFields, fields
+from leap.mail.imap.memorystore import MessageDict
 from leap.mail.imap.messages import MessageCollection
 from leap.mail.imap.parser import MBoxParser
 
@@ -80,7 +81,7 @@ class SoledadMailbox(WithMsgFields, MBoxParser):
 
     next_uid_lock = threading.Lock()
 
-    def __init__(self, mbox, soledad=None, rw=1):
+    def __init__(self, mbox, soledad, memstore, rw=1):
         """
         SoledadMailbox constructor. Needs to get passed a name, plus a
         Soledad instance.
@@ -91,9 +92,13 @@ class SoledadMailbox(WithMsgFields, MBoxParser):
         :param soledad: a Soledad instance.
         :type soledad: Soledad
 
-        :param rw: read-and-write flags
+        :param memstore: a MemoryStore instance
+        :type memstore: MemoryStore
+
+        :param rw: read-and-write flag for this mailbox
         :type rw: int
         """
+        print "got memstore: ", memstore
         leap_assert(mbox, "Need a mailbox name to initialize")
         leap_assert(soledad, "Need a soledad instance to initialize")
 
@@ -105,9 +110,10 @@ class SoledadMailbox(WithMsgFields, MBoxParser):
         self.rw = rw
 
         self._soledad = soledad
+        self._memstore = memstore
 
         self.messages = MessageCollection(
-            mbox=mbox, soledad=self._soledad)
+            mbox=mbox, soledad=self._soledad, memstore=self._memstore)
 
         if not self.getFlags():
             self.setFlags(self.INIT_FLAGS)
@@ -231,7 +237,10 @@ class SoledadMailbox(WithMsgFields, MBoxParser):
             # XXX It looks like it has been corrupted.
             # We need to be able to survive this.
             return None
-        return mbox.content.get(self.LAST_UID_KEY, 1)
+        last = mbox.content.get(self.LAST_UID_KEY, 1)
+        if self._memstore:
+            last = max(last, self._memstore.get_last_uid(mbox))
+        return last
 
     def _set_last_uid(self, uid):
         """
@@ -259,6 +268,7 @@ class SoledadMailbox(WithMsgFields, MBoxParser):
             value = count
 
         mbox.content[key] = value
+        # XXX this should be set in the memorystore instead!!!
         self._soledad.put_doc(mbox)
 
     last_uid = property(
@@ -532,12 +542,17 @@ class SoledadMailbox(WithMsgFields, MBoxParser):
         # can treat them all the same.
         # Change this to the flag that twisted expects when we
         # switch to content-hash based index + local UID table.
+        print
+        print "FETCHING..."
 
         sequence = False
         #sequence = True if uid == 0 else False
 
         messages_asked = self._bound_seq(messages_asked)
+        print "asked: ", messages_asked
         seq_messg = self._filter_msg_seq(messages_asked)
+
+        print "seq: ", seq_messg
         getmsg = lambda uid: self.messages.get_msg_by_uid(uid)
 
         # for sequence numbers (uid = 0)
@@ -769,36 +784,41 @@ class SoledadMailbox(WithMsgFields, MBoxParser):
         uid_next = self.getUIDNext()
         msg = messageObject
 
-        # XXX DEBUG ----------------------------------------
-        #print "copying MESSAGE from %s (%s) to %s (%s)" % (
-            #msg._mbox, msg._uid, self.mbox, uid_next)
-
         # XXX should use a public api instead
         fdoc = msg._fdoc
+        hdoc = msg._hdoc
         if not fdoc:
             logger.debug("Tried to copy a MSG with no fdoc")
             return
 
+        #old_mbox = fdoc.content[self.MBOX_KEY]
+        #old_uid = fdoc.content[self.UID_KEY]
+        #old_key = old_mbox, old_uid
+        #print "copying from OLD MBOX ", old_mbox
+
+        # XXX bit doubt... to duplicate in memory
+        # or not to...?
+        # I think it should be ok to duplicate as long as we're
+        # careful at the hour of writes...
+        # We could use also proxies, but it will break when
+        # the original mailbox is flushed.
+
+        # XXX DEBUG ----------------------------------------
+        #print "copying MESSAGE from %s (%s) to %s (%s)" % (
+            #msg._mbox, msg._uid, self.mbox, uid_next)
+
         new_fdoc = copy.deepcopy(fdoc.content)
         new_fdoc[self.UID_KEY] = uid_next
         new_fdoc[self.MBOX_KEY] = self.mbox
-        self._do_add_doc(new_fdoc)
+        self._memstore.put(self.mbox, uid_next, MessageDict(
+            new_fdoc, hdoc.content))
 
-        # XXX should use a public api instead
-        hdoc = msg._hdoc
-        self.messages.add_hdocset_docid(hdoc.doc_id)
+        # XXX use memory store
+        if hasattr(hdoc, 'doc_id'):
+            self.messages.add_hdocset_docid(hdoc.doc_id)
 
         deferLater(reactor, 1, self.notify_new)
 
-    def _do_add_doc(self, doc):
-        """
-        Defer the adding of a new doc.
-
-        :param doc: document to be created in soledad.
-        :type doc: dict
-        """
-        self._soledad.create_doc(doc)
-
     # convenience fun
 
     def deleteAllDocs(self):
diff --git a/mail/src/leap/mail/imap/memorystore.py b/mail/src/leap/mail/imap/memorystore.py
new file mode 100644
index 0000000..b8829e0
--- /dev/null
+++ b/mail/src/leap/mail/imap/memorystore.py
@@ -0,0 +1,478 @@
+# -*- coding: utf-8 -*-
+# memorystore.py
+# Copyright (C) 2014 LEAP
+#
+# This program is free software: you can redistribute it and/or modify
+# it under the terms of the GNU General Public License as published by
+# the Free Software Foundation, either version 3 of the License, or
+# (at your option) any later version.
+#
+# This program is distributed in the hope that it will be useful,
+# but WITHOUT ANY WARRANTY; without even the implied warranty of
+# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+# GNU General Public License for more details.
+#
+# You should have received a copy of the GNU General Public License
+# along with this program.  If not, see <http://www.gnu.org/licenses/>.
+"""
+In-memory transient store for a LEAPIMAPServer.
+"""
+import contextlib
+import logging
+import weakref
+
+from collections import namedtuple
+
+from twisted.internet.task import LoopingCall
+from zope.interface import implements
+
+from leap.mail import size
+from leap.mail.messageflow import MessageProducer
+from leap.mail.messageparts import MessagePartType
+from leap.mail.imap import interfaces
+from leap.mail.imap.fields import fields
+
+logger = logging.getLogger(__name__)
+
+
+"""
+A MessagePartDoc is a light wrapper around the dictionary-like
+data that we pass along for message parts. It can be used almost everywhere
+that you would expect a SoledadDocument, since it has a dict under the
+`content` attribute.
+
+We also keep some metadata on it, relative in part to the message as a whole,
+and sometimes to a part in particular only.
+
+* `new` indicates that the document has just been created. SoledadStore
+  should just create a new doc for all the related message parts.
+* `store` indicates the type of store a given MessagePartDoc lives in.
+  We currently use this to indicate that  the document comes from memeory,
+  but we should probably get rid of it as soon as we extend the use of the
+  SoledadStore interface along LeapMessage, MessageCollection and Mailbox.
+* `part` is one of the MessagePartType enums.
+
+* `dirty` indicates that, while we already have the document in Soledad,
+  we have modified its state in memory, so we need to put_doc instead while
+  dumping the MemoryStore contents.
+  `dirty` attribute would only apply to flags-docs and linkage-docs.
+
+
+  XXX this is still not implemented!
+
+"""
+
+MessagePartDoc = namedtuple(
+    'MessagePartDoc',
+    ['new', 'dirty', 'part', 'store', 'content'])
+
+
+class ReferenciableDict(dict):
+    """
+    A dict that can be weak-referenced.
+
+    Some builtin objects are not weak-referenciable unless
+    subclassed. So we do.
+
+    Used to return pointers to the items in the MemoryStore.
+    """
+
+
+class MessageWrapper(object):
+    """
+    A simple nested dictionary container around the different message subparts.
+    """
+    implements(interfaces.IMessageContainer)
+
+    FDOC = "fdoc"
+    HDOC = "hdoc"
+    CDOCS = "cdocs"
+
+    # XXX can use this to limit the memory footprint,
+    # or is it too premature to optimize?
+    # Does it work well together with the interfaces.implements?
+
+    #__slots__ = ["_dict", "_new", "_dirty", "memstore"]
+
+    def __init__(self, fdoc=None, hdoc=None, cdocs=None,
+                 from_dict=None, memstore=None,
+                 new=True, dirty=False):
+        self._dict = {}
+
+        self._new = new
+        self._dirty = dirty
+        self.memstore = memstore
+
+        if from_dict is not None:
+            self.from_dict(from_dict)
+        else:
+            if fdoc is not None:
+                self._dict[self.FDOC] = ReferenciableDict(fdoc)
+            if hdoc is not None:
+                self._dict[self.HDOC] = ReferenciableDict(hdoc)
+            if cdocs is not None:
+                self._dict[self.CDOCS] = ReferenciableDict(cdocs)
+
+    # properties
+
+    @property
+    def new(self):
+        return self._new
+
+    def set_new(self, value=True):
+        self._new = value
+
+    @property
+    def dirty(self):
+        return self._dirty
+
+    def set_dirty(self, value=True):
+        self._dirty = value
+
+    # IMessageContainer
+
+    @property
+    def fdoc(self):
+        _fdoc = self._dict.get(self.FDOC, None)
+        if _fdoc:
+            content_ref = weakref.proxy(_fdoc)
+        else:
+            logger.warning("NO FDOC!!!")
+            content_ref = {}
+        return MessagePartDoc(new=self.new, dirty=self.dirty,
+                              store=self._storetype,
+                              part=MessagePartType.fdoc,
+                              content=content_ref)
+
+    @property
+    def hdoc(self):
+        _hdoc = self._dict.get(self.HDOC, None)
+        if _hdoc:
+            content_ref = weakref.proxy(_hdoc)
+        else:
+            logger.warning("NO HDOC!!!!")
+            content_ref = {}
+        return MessagePartDoc(new=self.new, dirty=self.dirty,
+                              store=self._storetype,
+                              part=MessagePartType.hdoc,
+                              content=content_ref)
+
+    @property
+    def cdocs(self):
+        _cdocs = self._dict.get(self.CDOCS, None)
+        if _cdocs:
+            return weakref.proxy(_cdocs)
+        else:
+            return {}
+
+    def walk(self):
+        """
+        Generator that iterates through all the parts, returning
+        MessagePartDoc.
+        """
+        yield self.fdoc
+        yield self.hdoc
+        for cdoc in self.cdocs.values():
+            # XXX this will break ----
+            content_ref = weakref.proxy(cdoc)
+            yield MessagePartDoc(new=self.new, dirty=self.dirty,
+                                 store=self._storetype,
+                                 part=MessagePartType.cdoc,
+                                 content=content_ref)
+
+    # i/o
+
+    def as_dict(self):
+        """
+        Return a dict representation of the parts contained.
+        """
+        return self._dict
+
+    def from_dict(self, msg_dict):
+        """
+        Populate MessageWrapper parts from a dictionary.
+        It expects the same format that we use in a
+        MessageWrapper.
+        """
+        fdoc, hdoc, cdocs = map(
+            lambda part: msg_dict.get(part, None),
+            [self.FDOC, self.HDOC, self.CDOCS])
+        self._dict[self.FDOC] = fdoc
+        self._dict[self.HDOC] = hdoc
+        self._dict[self.CDOCS] = cdocs
+
+
+@contextlib.contextmanager
+def set_bool_flag(obj, att):
+    """
+    Set a boolean flag to True while we're doing our thing.
+    Just to let the world know.
+    """
+    setattr(obj, att, True)
+    try:
+        yield True
+    except RuntimeError as exc:
+        logger.exception(exc)
+    finally:
+        setattr(obj, att, False)
+
+
+class MemoryStore(object):
+    """
+    An in-memory store to where we can write the different parts that
+    we split the messages into and buffer them until we write them to the
+    permanent storage.
+
+    It uses MessageWrapper instances to represent the message-parts, which are
+    indexed by mailbox name and UID.
+
+    It also can be passed a permanent storage as a paremeter (any implementor
+    of IMessageStore, in this case a SoledadStore). In this case, a periodic
+    dump of the messages stored in memory will be done. The period of the
+    writes to the permanent storage is controled by the write_period parameter
+    in the constructor.
+    """
+    implements(interfaces.IMessageStore)
+    implements(interfaces.IMessageStoreWriter)
+
+    producer = None
+
+    # TODO We will want to index by chash when we transition to local-only
+    # UIDs.
+    # TODO should store RECENT-FLAGS too
+    # TODO should store HDOCSET too (use weakrefs!) -- will need to subclass
+    # TODO do use dirty flag (maybe use namedtuples for that) so we can use it
+    # also as a read-cache.
+
+    WRITING_FLAG = "_writing"
+
+    def __init__(self, permanent_store=None, write_period=60):
+        """
+        Initialize a MemoryStore.
+
+        :param permanent_store: a IMessageStore implementor to dump
+                                messages to.
+        :type permanent_store: IMessageStore
+        :param write_period: the interval to dump messages to disk, in seconds.
+        :type write_period: int
+        """
+        self._permanent_store = permanent_store
+        self._write_period = write_period
+
+        # Internal Storage
+        self._msg_store = {}
+        self._phash_store = {}
+
+        # TODO ----------------- implement mailbox-level flags store too! ----
+        self._rflags_store = {}
+        self._hdocset_store = {}
+        # TODO ----------------- implement mailbox-level flags store too! ----
+
+        # New and dirty flags, to set MessageWrapper State.
+        self._new = set([])
+        self._dirty = set([])
+
+        # Flag for signaling we're busy writing to the disk storage.
+        setattr(self, self.WRITING_FLAG, False)
+
+        if self._permanent_store is not None:
+            # this producer spits its messages to the permanent store
+            # consumer using a queue. We will use that to put
+            # our messages to be written.
+            self.producer = MessageProducer(permanent_store,
+                                            period=0.1)
+            # looping call for dumping to SoledadStore
+            self._write_loop = LoopingCall(self.write_messages,
+                                           permanent_store)
+
+            # We can start the write loop right now, why wait?
+            self._start_write_loop()
+
+    def _start_write_loop(self):
+        """
+        Start loop for writing to disk database.
+        """
+        if not self._write_loop.running:
+            self._write_loop.start(self._write_period, now=True)
+
+    def _stop_write_loop(self):
+        """
+        Stop loop for writing to disk database.
+        """
+        if self._write_loop.running:
+            self._write_loop.stop()
+
+    # IMessageStore
+
+    # XXX this would work well for whole message operations.
+    # We would have to add a put_flags operation to modify only
+    # the flags doc (and set the dirty flag accordingly)
+
+    def create_message(self, mbox, uid, message):
+        """
+        Create the passed message into this MemoryStore.
+
+        By default we consider that any message is a new message.
+        """
+        print "adding new doc to memstore %s (%s)" % (mbox, uid)
+        key = mbox, uid
+        self._new.add(key)
+
+        msg_dict = message.as_dict()
+        self._msg_store[key] = msg_dict
+
+        cdocs = message.cdocs
+
+        dirty = key in self._dirty
+        new = key in self._new
+
+        # XXX should capture this in log...
+
+        for cdoc_key in cdocs.keys():
+            print "saving cdoc"
+            cdoc = self._msg_store[key]['cdocs'][cdoc_key]
+
+            # XXX this should be done in the MessageWrapper constructor
+            # instead...
+            # first we make it weak-referenciable
+            referenciable_cdoc = ReferenciableDict(cdoc)
+            self._msg_store[key]['cdocs'][cdoc_key] = MessagePartDoc(
+                new=new, dirty=dirty, store="mem",
+                part=MessagePartType.cdoc,
+                content=referenciable_cdoc)
+            phash = cdoc.get(fields.PAYLOAD_HASH_KEY, None)
+            if not phash:
+                continue
+            self._phash_store[phash] = weakref.proxy(referenciable_cdoc)
+
+    def put_message(self, mbox, uid, msg):
+        """
+        Put an existing message.
+        """
+        return NotImplementedError()
+
+    def get_message(self, mbox, uid):
+        """
+        Get a MessageWrapper for the given mbox and uid combination.
+
+        :return: MessageWrapper or None
+        """
+        key = mbox, uid
+        msg_dict = self._msg_store.get(key, None)
+        if msg_dict:
+            new, dirty = self._get_new_dirty_state(key)
+            return MessageWrapper(from_dict=msg_dict,
+                                  memstore=weakref.proxy(self))
+        else:
+            return None
+
+    def remove_message(self, mbox, uid):
+        """
+        Remove a Message from this MemoryStore.
+        """
+        raise NotImplementedError()
+
+    # IMessageStoreWriter
+
+    def write_messages(self, store):
+        """
+        Write the message documents in this MemoryStore to a different store.
+        """
+        # XXX pass if it's writing (ie, the queue is not empty...)
+        # See how to make the writing_flag aware of the queue state...
+        print "writing messages to producer..."
+
+        with set_bool_flag(self, self.WRITING_FLAG):
+            for msg_wrapper in self.all_msg_iter():
+                self.producer.push(msg_wrapper)
+
+    # MemoryStore specific methods.
+
+    def get_uids(self, mbox):
+        """
+        Get all uids for a given mbox.
+        """
+        all_keys = self._msg_store.keys()
+        return [uid for m, uid in all_keys if m == mbox]
+
+    def get_last_uid(self, mbox):
+        """
+        Get the highest UID for a given mbox.
+        """
+        # XXX should get from msg_store keys instead!
+        if not self._new:
+            return 0
+        return max(self.get_uids(mbox))
+
+    def count_new_mbox(self, mbox):
+        """
+        Count the new messages by inbox.
+        """
+        return len([(m, uid) for m, uid in self._new if mbox == mbox])
+
+    def count_new(self):
+        """
+        Count all the new messages in the MemoryStore.
+        """
+        return len(self._new)
+
+    def get_by_phash(self, phash):
+        """
+        Return a content-document by its payload-hash.
+        """
+        doc = self._phash_store.get(phash, None)
+
+        # XXX have to keep a mapping between phash and its linkage
+        # info, to know if this payload is been already saved or not.
+        # We will be able to get this from the linkage-docs,
+        # not yet implemented.
+        new = True
+        dirty = False
+        return MessagePartDoc(
+            new=new, dirty=dirty, store="mem",
+            part=MessagePartType.cdoc,
+            content=doc)
+
+    def all_msg_iter(self):
+        """
+        Return generator that iterates through all messages in the store.
+        """
+        return (self.get_message(*key)
+                for key in sorted(self._msg_store.keys()))
+
+    def _get_new_dirty_state(self, key):
+        """
+        Return `new` and `dirty` flags for a given message.
+        """
+        return map(lambda _set: key in _set, (self._new, self._dirty))
+
+    @property
+    def is_writing(self):
+        """
+        Property that returns whether the store is currently writing its
+        internal state to a permanent storage.
+
+        Used to evaluate whether the CHECK command can inform that the field
+        is clear to proceed, or waiting for the write operations to complete
+        is needed instead.
+
+        :rtype: bool
+        """
+        # XXX this should probably return a deferred !!!
+        return getattr(self, self.WRITING_FLAG)
+
+    def put_part(self, part_type, value):
+        """
+        Put the passed part into this IMessageStore.
+        `part` should be one of: fdoc, hdoc, cdoc
+        """
+        # XXX turn that into a enum
+
+    # Memory management.
+
+    def get_size(self):
+        """
+        Return the size of the internal storage.
+        Use for calculating the limit beyond which we should flush the store.
+        """
+        return size.get_size(self._msg_store)
diff --git a/mail/src/leap/mail/imap/messages.py b/mail/src/leap/mail/imap/messages.py
index 34304ea..ef0b0a1 100644
--- a/mail/src/leap/mail/imap/messages.py
+++ b/mail/src/leap/mail/imap/messages.py
@@ -42,6 +42,7 @@ from leap.mail.utils import first, find_charset
 from leap.mail.decorators import deferred
 from leap.mail.imap.index import IndexedDB
 from leap.mail.imap.fields import fields, WithMsgFields
+from leap.mail.imap.memorystore import MessageDict
 from leap.mail.imap.parser import MailParser, MBoxParser
 from leap.mail.messageflow import IMessageConsumer
 
@@ -49,11 +50,20 @@ logger = logging.getLogger(__name__)
 
 # TODO ------------------------------------------------------------
 
+# [ ] Add ref to incoming message during add_msg
 # [ ] Add linked-from info.
 # [ ] Delete incoming mail only after successful write!
 # [ ] Remove UID from syncable db. Store only those indexes locally.
 
 
+# XXX no longer needed, since i'm using proxies instead of direct weakrefs
+def maybe_call(thing):
+    """
+    Return the same thing, or the result of its invocation if it is a callable.
+    """
+    return thing() if callable(thing) else thing
+
+
 def lowerdict(_dict):
     """
     Return a dict with the keys in lowercase.
@@ -333,7 +343,7 @@ class LeapMessage(fields, MailParser, MBoxParser):
 
     implements(imap4.IMessage)
 
-    def __init__(self, soledad, uid, mbox, collection=None):
+    def __init__(self, soledad, uid, mbox, collection=None, container=None):
         """
         Initializes a LeapMessage.
 
@@ -345,12 +355,15 @@ class LeapMessage(fields, MailParser, MBoxParser):
         :type mbox: basestring
         :param collection: a reference to the parent collection object
         :type collection: MessageCollection
+        :param container: a IMessageContainer implementor instance
+        :type container: IMessageContainer
         """
         MailParser.__init__(self)
         self._soledad = soledad
         self._uid = int(uid)
         self._mbox = self._parse_mailbox_name(mbox)
         self._collection = collection
+        self._container = container
 
         self.__chash = None
         self.__bdoc = None
@@ -361,13 +374,29 @@ class LeapMessage(fields, MailParser, MBoxParser):
         An accessor to the flags document.
         """
         if all(map(bool, (self._uid, self._mbox))):
-            fdoc = self._get_flags_doc()
+            fdoc = None
+            if self._container is not None:
+                fdoc = self._container.fdoc
+            if not fdoc:
+                fdoc = self._get_flags_doc()
             if fdoc:
-                self.__chash = fdoc.content.get(
+                fdoc_content = maybe_call(fdoc.content)
+                self.__chash = fdoc_content.get(
                     fields.CONTENT_HASH_KEY, None)
             return fdoc
 
     @property
+    def _hdoc(self):
+        """
+        An accessor to the headers document.
+        """
+        if self._container is not None:
+            hdoc = self._container.hdoc
+            if hdoc:
+                return hdoc
+        return self._get_headers_doc()
+
+    @property
     def _chash(self):
         """
         An accessor to the content hash for this message.
@@ -375,18 +404,11 @@ class LeapMessage(fields, MailParser, MBoxParser):
         if not self._fdoc:
             return None
         if not self.__chash and self._fdoc:
-            self.__chash = self._fdoc.content.get(
+            self.__chash = maybe_call(self._fdoc.content).get(
                 fields.CONTENT_HASH_KEY, None)
         return self.__chash
 
     @property
-    def _hdoc(self):
-        """
-        An accessor to the headers document.
-        """
-        return self._get_headers_doc()
-
-    @property
     def _bdoc(self):
         """
         An accessor to the body document.
@@ -422,7 +444,7 @@ class LeapMessage(fields, MailParser, MBoxParser):
         flags = []
         fdoc = self._fdoc
         if fdoc:
-            flags = fdoc.content.get(self.FLAGS_KEY, None)
+            flags = maybe_call(fdoc.content).get(self.FLAGS_KEY, None)
 
         msgcol = self._collection
 
@@ -449,6 +471,8 @@ class LeapMessage(fields, MailParser, MBoxParser):
         :return: a SoledadDocument instance
         :rtype: SoledadDocument
         """
+        # XXX use memory store ...!
+
         leap_assert(isinstance(flags, tuple), "flags need to be a tuple")
         log.msg('setting flags: %s (%s)' % (self._uid, flags))
 
@@ -461,7 +485,9 @@ class LeapMessage(fields, MailParser, MBoxParser):
         doc.content[self.FLAGS_KEY] = flags
         doc.content[self.SEEN_KEY] = self.SEEN_FLAG in flags
         doc.content[self.DEL_KEY] = self.DELETED_FLAG in flags
-        self._soledad.put_doc(doc)
+
+        if getattr(doc, 'store', None) != "mem":
+            self._soledad.put_doc(doc)
 
     def addFlags(self, flags):
         """
@@ -521,18 +547,26 @@ class LeapMessage(fields, MailParser, MBoxParser):
         """
         # TODO refactor with getBodyFile in MessagePart
         fd = StringIO.StringIO()
-        bdoc = self._bdoc
-        if bdoc:
-            body = self._bdoc.content.get(self.RAW_KEY, "")
-            content_type = bdoc.content.get('content-type', "")
+        if self._bdoc is not None:
+            bdoc_content = self._bdoc.content
+            body = bdoc_content.get(self.RAW_KEY, "")
+            content_type = bdoc_content.get('content-type', "")
             charset = find_charset(content_type)
+            logger.debug('got charset from content-type: %s' % charset)
             if charset is None:
                 charset = self._get_charset(body)
             try:
                 body = body.encode(charset)
             except UnicodeError as e:
-                logger.error("Unicode error, using 'replace'. {0!r}".format(e))
-                body = body.encode(charset, 'replace')
+                logger.error("Unicode error {0}".format(e))
+                logger.debug("Attempted to encode with: %s" % charset)
+                try:
+                    body = body.encode(charset, 'replace')
+                except UnicodeError as e:
+                    try:
+                        body = body.encode('utf-8', 'replace')
+                    except:
+                        pass
 
         # We are still returning funky characters from here.
         else:
@@ -567,7 +601,8 @@ class LeapMessage(fields, MailParser, MBoxParser):
         """
         size = None
         if self._fdoc:
-            size = self._fdoc.content.get(self.SIZE_KEY, False)
+            fdoc_content = maybe_call(self._fdoc.content)
+            size = fdoc_content.get(self.SIZE_KEY, False)
         else:
             logger.warning("No FLAGS doc for %s:%s" % (self._mbox,
                                                        self._uid))
@@ -632,7 +667,8 @@ class LeapMessage(fields, MailParser, MBoxParser):
         Return the headers dict for this message.
         """
         if self._hdoc is not None:
-            headers = self._hdoc.content.get(self.HEADERS_KEY, {})
+            hdoc_content = maybe_call(self._hdoc.content)
+            headers = hdoc_content.get(self.HEADERS_KEY, {})
             return headers
 
         else:
@@ -646,7 +682,8 @@ class LeapMessage(fields, MailParser, MBoxParser):
         Return True if this message is multipart.
         """
         if self._fdoc:
-            is_multipart = self._fdoc.content.get(self.MULTIPART_KEY, False)
+            fdoc_content = maybe_call(self._fdoc.content)
+            is_multipart = fdoc_content.get(self.MULTIPART_KEY, False)
             return is_multipart
         else:
             logger.warning(
@@ -688,7 +725,8 @@ class LeapMessage(fields, MailParser, MBoxParser):
             logger.warning("Tried to get part but no HDOC found!")
             return None
 
-        pmap = self._hdoc.content.get(fields.PARTS_MAP_KEY, {})
+        hdoc_content = maybe_call(self._hdoc.content)
+        pmap = hdoc_content.get(fields.PARTS_MAP_KEY, {})
         return pmap[str(part)]
 
     def _get_flags_doc(self):
@@ -724,16 +762,33 @@ class LeapMessage(fields, MailParser, MBoxParser):
         Return the document that keeps the body for this
         message.
         """
-        body_phash = self._hdoc.content.get(
+        hdoc_content = maybe_call(self._hdoc.content)
+        body_phash = hdoc_content.get(
             fields.BODY_KEY, None)
         if not body_phash:
             logger.warning("No body phash for this document!")
             return None
-        body_docs = self._soledad.get_from_index(
-            fields.TYPE_P_HASH_IDX,
-            fields.TYPE_CONTENT_VAL, str(body_phash))
 
-        return first(body_docs)
+        # XXX get from memstore too...
+        # if memstore: memstore.get_phrash
+        # memstore should keep a dict with weakrefs to the
+        # phash doc...
+
+        if self._container is not None:
+            bdoc = self._container.memstore.get_by_phash(body_phash)
+            if bdoc:
+                return bdoc
+            else:
+                print "no doc for that phash found!"
+
+        # no memstore or no doc found there
+        if self._soledad:
+            body_docs = self._soledad.get_from_index(
+                fields.TYPE_P_HASH_IDX,
+                fields.TYPE_CONTENT_VAL, str(body_phash))
+            return first(body_docs)
+        else:
+            logger.error("No phash in container, and no soledad found!")
 
     def __getitem__(self, key):
         """
@@ -746,7 +801,7 @@ class LeapMessage(fields, MailParser, MBoxParser):
         :return: The content value indexed by C{key} or None
         :rtype: str
         """
-        return self._fdoc.content.get(key, None)
+        return maybe_call(self._fdoc.content).get(key, None)
 
     # setters
 
@@ -790,6 +845,8 @@ class LeapMessage(fields, MailParser, MBoxParser):
         # until we think about a good way of deorphaning.
         # Maybe a crawler of unreferenced docs.
 
+        # XXX remove from memory store!!!
+
         # XXX implement elijah's idea of using a PUT document as a
         # token to ensure consistency in the removal.
 
@@ -957,7 +1014,7 @@ class MessageCollection(WithMsgFields, IndexedDB, MailParser, MBoxParser,
     """
     A collection of messages, surprisingly.
 
-    It is tied to a selected mailbox name that is passed to constructor.
+    It is tied to a selected mailbox name that is passed to its constructor.
     Implements a filter query over the messages contained in a soledad
     database.
     """
@@ -1058,7 +1115,7 @@ class MessageCollection(WithMsgFields, IndexedDB, MailParser, MBoxParser,
     _hdocset_lock = threading.Lock()
     _hdocset_property_lock = threading.Lock()
 
-    def __init__(self, mbox=None, soledad=None):
+    def __init__(self, mbox=None, soledad=None, memstore=None):
         """
         Constructor for MessageCollection.
 
@@ -1068,13 +1125,18 @@ class MessageCollection(WithMsgFields, IndexedDB, MailParser, MBoxParser,
         MessageCollection for each mailbox, instead of treating them
         as a property of each message.
 
+        We are passed an instance of MemoryStore, the same for the
+        SoledadBackedAccount, that we use as a read cache and a buffer
+        for writes.
+
         :param mbox: the name of the mailbox. It is the name
                      with which we filter the query over the
-                     messages database
+                     messages database.
         :type mbox: str
-
         :param soledad: Soledad database
         :type soledad: Soledad instance
+        :param memstore: a MemoryStore instance
+        :type memstore: MemoryStore
         """
         MailParser.__init__(self)
         leap_assert(mbox, "Need a mailbox name to initialize")
@@ -1086,6 +1148,8 @@ class MessageCollection(WithMsgFields, IndexedDB, MailParser, MBoxParser,
         # okay, all in order, keep going...
         self.mbox = self._parse_mailbox_name(mbox)
         self._soledad = soledad
+        self._memstore = memstore
+
         self.__rflags = None
         self.__hdocset = None
         self.initialize_db()
@@ -1241,6 +1305,7 @@ class MessageCollection(WithMsgFields, IndexedDB, MailParser, MBoxParser,
         # when all the processing is done.
 
         # TODO add the linked-from info !
+        # TODO add reference to the original message
 
         logger.debug('adding message')
         if flags is None:
@@ -1273,24 +1338,29 @@ class MessageCollection(WithMsgFields, IndexedDB, MailParser, MBoxParser,
             hd[key] = parts_map[key]
         del parts_map
 
-        # Saving ----------------------------------------
-        self.set_recent_flag(uid)
+        # The MessageContainer expects a dict, zero-indexed
+        # XXX review-me
+        cdocs = dict((index, doc) for index, doc in
+                     enumerate(walk.get_raw_docs(msg, parts)))
+        print "cdocs is", cdocs
 
-        # first, regular docs: flags and headers
-        self._soledad.create_doc(fd)
+        # Saving ----------------------------------------
         # XXX should check for content duplication on headers too
         # but with chash. !!!
-        hdoc = self._soledad.create_doc(hd)
+
+        # XXX adapt hdocset to use memstore
+        #hdoc = self._soledad.create_doc(hd)
         # We add the newly created hdoc to the fast-access set of
         # headers documents associated with the mailbox.
-        self.add_hdocset_docid(hdoc.doc_id)
+        #self.add_hdocset_docid(hdoc.doc_id)
 
-        # and last, but not least, try to create
-        # content docs if not already there.
-        cdocs = walk.get_raw_docs(msg, parts)
-        for cdoc in cdocs:
-            if not self._content_does_exist(cdoc):
-                self._soledad.create_doc(cdoc)
+        # XXX move to memory store too
+        # self.set_recent_flag(uid)
+
+        # TODO ---- add reference to original doc, to be deleted
+        # after writes are done.
+        msg_container = MessageDict(fd, hd, cdocs)
+        self._memstore.put(self.mbox, uid, msg_container)
 
     def _remove_cb(self, result):
         return result
@@ -1321,6 +1391,8 @@ class MessageCollection(WithMsgFields, IndexedDB, MailParser, MBoxParser,
     #
 
     # recent flags
+    # XXX FIXME -------------------------------------
+    # This should be rewritten to use memory store.
 
     def _get_recent_flags(self):
         """
@@ -1390,6 +1462,9 @@ class MessageCollection(WithMsgFields, IndexedDB, MailParser, MBoxParser,
 
     # headers-docs-set
 
+    # XXX FIXME -------------------------------------
+    # This should be rewritten to use memory store.
+
     def _get_hdocset(self):
         """
         An accessor for the hdocs-set for this mailbox.
@@ -1532,7 +1607,16 @@ class MessageCollection(WithMsgFields, IndexedDB, MailParser, MBoxParser,
                  or None if not found.
         :rtype: LeapMessage
         """
-        msg = LeapMessage(self._soledad, uid, self.mbox, collection=self)
+        print "getting msg by id!"
+        msg_container = self._memstore.get(self.mbox, uid)
+        print "msg container", msg_container
+        if msg_container is not None:
+            print "getting LeapMessage (from memstore)"
+            msg = LeapMessage(None, uid, self.mbox, collection=self,
+                              container=msg_container)
+            print "got msg:", msg
+        else:
+            msg = LeapMessage(self._soledad, uid, self.mbox, collection=self)
         if not msg.does_exist():
             return None
         return msg
@@ -1570,11 +1654,19 @@ class MessageCollection(WithMsgFields, IndexedDB, MailParser, MBoxParser,
         ascending order.
         """
         # XXX we should get this from the uid table, local-only
-        all_uids = (doc.content[self.UID_KEY] for doc in
-                    self._soledad.get_from_index(
-                        fields.TYPE_MBOX_IDX,
-                        fields.TYPE_FLAGS_VAL, self.mbox))
-        return (u for u in sorted(all_uids))
+        # XXX FIXME -------------
+        # This should be cached in the memstoretoo
+        db_uids = set([doc.content[self.UID_KEY] for doc in
+                       self._soledad.get_from_index(
+                           fields.TYPE_MBOX_IDX,
+                           fields.TYPE_FLAGS_VAL, self.mbox)])
+        if self._memstore is not None:
+            mem_uids = self._memstore.get_uids(self.mbox)
+            uids = db_uids.union(set(mem_uids))
+        else:
+            uids = db_uids
+
+        return (u for u in sorted(uids))
 
     def reset_last_uid(self, param):
         """
@@ -1592,12 +1684,21 @@ class MessageCollection(WithMsgFields, IndexedDB, MailParser, MBoxParser,
         """
         Return a dict with all flags documents for this mailbox.
         """
+        # XXX get all from memstore and cahce it there
         all_flags = dict(((
             doc.content[self.UID_KEY],
             doc.content[self.FLAGS_KEY]) for doc in
             self._soledad.get_from_index(
                 fields.TYPE_MBOX_IDX,
                 fields.TYPE_FLAGS_VAL, self.mbox)))
+        if self._memstore is not None:
+            # XXX
+            uids = self._memstore.get_uids(self.mbox)
+            fdocs = [(uid, self._memstore.get(self.mbox, uid).fdoc)
+                     for uid in uids]
+            for uid, doc in fdocs:
+                all_flags[uid] = doc.content[self.FLAGS_KEY]
+
         return all_flags
 
     def all_flags_chash(self):
@@ -1630,9 +1731,12 @@ class MessageCollection(WithMsgFields, IndexedDB, MailParser, MBoxParser,
 
         :rtype: int
         """
+        # XXX We could cache this in memstore too until next write...
         count = self._soledad.get_count_from_index(
             fields.TYPE_MBOX_IDX,
             fields.TYPE_FLAGS_VAL, self.mbox)
+        if self._memstore is not None:
+            count += self._memstore.count_new()
         return count
 
     # unseen messages
diff --git a/mail/src/leap/mail/imap/service/imap.py b/mail/src/leap/mail/imap/service/imap.py
index ad22da6..71b9950 100644
--- a/mail/src/leap/mail/imap/service/imap.py
+++ b/mail/src/leap/mail/imap/service/imap.py
@@ -36,6 +36,7 @@ from leap.common.check import leap_assert, leap_assert_type, leap_check
 from leap.keymanager import KeyManager
 from leap.mail.imap.account import SoledadBackedAccount
 from leap.mail.imap.fetch import LeapIncomingMail
+from leap.mail.imap.memorystore import MemoryStore
 from leap.soledad.client import Soledad
 
 # The default port in which imap service will run
@@ -69,6 +70,8 @@ except Exception:
 ######################################################
 
 
+# TODO move this to imap.server
+
 class LeapIMAPServer(imap4.IMAP4Server):
     """
     An IMAP4 Server with mailboxes backed by soledad
@@ -256,11 +259,15 @@ class LeapIMAPFactory(ServerFactory):
         self._uuid = uuid
         self._userid = userid
         self._soledad = soledad
+        self._memstore = MemoryStore()
 
         theAccount = SoledadBackedAccount(
-            uuid, soledad=soledad)
+            uuid, soledad=soledad,
+            memstore=self._memstore)
         self.theAccount = theAccount
 
+        # XXX how to pass the store along?
+
     def buildProtocol(self, addr):
         "Return a protocol suitable for the job."
         imapProtocol = LeapIMAPServer(
@@ -323,3 +330,14 @@ def run_service(*args, **kwargs):
 
     # not ok, signal error.
     leap_events.signal(IMAP_SERVICE_FAILED_TO_START, str(port))
+
+    def checkpoint(self):
+        """
+        Called when the client issues a CHECK command.
+
+        This should perform any checkpoint operations required by the server.
+        It may be a long running operation, but may not block.  If it returns
+        a deferred, the client will only be informed of success (or failure)
+        when the deferred's callback (or errback) is invoked.
+        """
+        return None
diff --git a/mail/src/leap/mail/messageflow.py b/mail/src/leap/mail/messageflow.py
index ac26e45..ed6abcd 100644
--- a/mail/src/leap/mail/messageflow.py
+++ b/mail/src/leap/mail/messageflow.py
@@ -25,12 +25,15 @@ from zope.interface import Interface, implements
 
 
 class IMessageConsumer(Interface):
+    """
+    I consume messages from a queue.
+    """
 
     def consume(self, queue):
         """
         Consumes the passed item.
 
-        :param item: q queue where we put the object to be consumed.
+        :param item: a queue where we put the object to be consumed.
         :type item: object
         """
         # TODO we could add an optional type to be passed
@@ -40,6 +43,28 @@ class IMessageConsumer(Interface):
         # the queue, maybe wrapped in an object with a retries attribute.
 
 
+class IMessageProducer(Interface):
+    """
+    I produce messages and put them in a store to be consumed by other
+    entities.
+    """
+
+    def push(self, item):
+        """
+        Push a new item in the queue.
+        """
+
+    def start(self):
+        """
+        Start producing items.
+        """
+
+    def stop(self):
+        """
+        Stop producing items.
+        """
+
+
 class DummyMsgConsumer(object):
 
     implements(IMessageConsumer)
@@ -62,6 +87,8 @@ class MessageProducer(object):
     deferred chain and leave further processing detached from the calling loop,
     as in the case of smtp.
     """
+    implements(IMessageProducer)
+
     # TODO this can be seen as a first step towards properly implementing
     # components that implement IPushProducer / IConsumer  interfaces.
     # However, I need to think more about how to pause the streaming.
@@ -92,7 +119,7 @@ class MessageProducer(object):
 
     def _check_for_new(self):
         """
-        Checks for new items in the internal queue, and calls the consume
+        Check for new items in the internal queue, and calls the consume
         method in the consumer.
 
         If the queue is found empty, the loop is stopped. It will be started
@@ -102,11 +129,11 @@ class MessageProducer(object):
         if self._queue.empty():
             self.stop()
 
-    # public methods
+    # public methods: IMessageProducer
 
-    def put(self, item):
+    def push(self, item):
         """
-        Puts a new item in the queue.
+        Push a new item in the queue.
 
         If the queue was empty, we will start the loop again.
         """
@@ -117,7 +144,7 @@ class MessageProducer(object):
 
     def start(self):
         """
-        Starts polling for new items.
+        Start polling for new items.
         """
         if not self._loop.running:
             self._loop.start(self._period, now=True)
diff --git a/mail/src/leap/mail/size.py b/mail/src/leap/mail/size.py
new file mode 100644
index 0000000..4880d71
--- /dev/null
+++ b/mail/src/leap/mail/size.py
@@ -0,0 +1,57 @@
+# -*- coding: utf-8 -*-
+# size.py
+# Copyright (C) 2014 LEAP
+#
+# This program is free software: you can redistribute it and/or modify
+# it under the terms of the GNU General Public License as published by
+# the Free Software Foundation, either version 3 of the License, or
+# (at your option) any later version.
+#
+# This program is distributed in the hope that it will be useful,
+# but WITHOUT ANY WARRANTY; without even the implied warranty of
+# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+# GNU General Public License for more details.
+#
+# You should have received a copy of the GNU General Public License
+# along with this program.  If not, see <http://www.gnu.org/licenses/>.
+"""
+Recursively get size of objects.
+"""
+from gc import collect
+from itertools import chain
+from sys import getsizeof
+
+
+def _get_size(item, seen):
+    known_types = {dict: lambda d: chain.from_iterable(d.items())}
+    default_size = getsizeof(0)
+
+    def size_walk(item):
+        if id(item) in seen:
+            return 0
+        seen.add(id(item))
+        s = getsizeof(item, default_size)
+        for _type, fun in known_types.iteritems():
+            if isinstance(item, _type):
+                s += sum(map(size_walk, fun(item)))
+                break
+        return s
+
+    return size_walk(item)
+
+
+def get_size(item):
+    """
+    Return the cumulative size of a given object.
+
+    Currently it supports only dictionaries, and seemingly leaks
+    some memory, so use with care.
+
+    :param item: the item which size wants to be computed
+    """
+    seen = set()
+    size = _get_size(item, seen)
+    #print "len(seen) ", len(seen)
+    del seen
+    collect()
+    return size