]> git.notmuchmail.org Git - notmuch/blobdiff - lib/database.cc
lib: check talloc success in compact
[notmuch] / lib / database.cc
index 9c2f4ecdc58ff2bcb6de93eba16b8ddc3dfd0b00..7a8702e263c46985bb21818d3f8dd039e39799d6 100644 (file)
  */
 
 #include "database-private.h"
+#include "parse-time-vrp.h"
 
 #include <iostream>
 
 #include <sys/time.h>
+#include <sys/stat.h>
 #include <signal.h>
+#include <ftw.h>
 
 #include <glib.h> /* g_free, GPtrArray, GHashTable */
+#include <glib-object.h> /* g_type_init */
+
+#include <gmime/gmime.h> /* g_mime_init */
 
 using namespace std;
 
@@ -80,13 +86,17 @@ typedef struct {
  *                     STRING is the name of a file within that
  *                     directory for this mail message.
  *
- *    A mail document also has two values:
+ *    A mail document also has four values:
  *
  *     TIMESTAMP:      The time_t value corresponding to the message's
  *                     Date header.
  *
  *     MESSAGE_ID:     The unique ID of the mail mess (see "id" above)
  *
+ *     FROM:           The value of the "From" header
+ *
+ *     SUBJECT:        The value of the "Subject" header
+ *
  * In addition, terms from the content of the message are added with
  * "from", "to", "attachment", and "subject" prefixes for use by the
  * user in searching. Similarly, terms from the path of the mail
@@ -209,21 +219,6 @@ static prefix_t PROBABILISTIC_PREFIX[]= {
     { "folder",                        "XFOLDER"}
 };
 
-int
-_internal_error (const char *format, ...)
-{
-    va_list va_args;
-
-    va_start (va_args, format);
-
-    fprintf (stderr, "Internal error: ");
-    vfprintf (stderr, format, va_args);
-
-    exit (1);
-
-    return 1;
-}
-
 const char *
 _find_prefix (const char *name)
 {
@@ -273,6 +268,10 @@ notmuch_status_to_string (notmuch_status_t status)
        return "Tag value is too long (exceeds NOTMUCH_TAG_MAX)";
     case NOTMUCH_STATUS_UNBALANCED_FREEZE_THAW:
        return "Unbalanced number of calls to notmuch_message_freeze/thaw";
+    case NOTMUCH_STATUS_UNBALANCED_ATOMIC:
+       return "Unbalanced number of calls to notmuch_database_begin_atomic/end_atomic";
+    case NOTMUCH_STATUS_UNSUPPORTED_OPERATION:
+       return "Unsupported operation";
     default:
     case NOTMUCH_STATUS_LAST_STATUS:
        return "Unknown error status value";
@@ -358,13 +357,17 @@ _message_id_compressed (void *ctx, const char *message_id)
     return compressed;
 }
 
-notmuch_message_t *
+notmuch_status_t
 notmuch_database_find_message (notmuch_database_t *notmuch,
-                              const char *message_id)
+                              const char *message_id,
+                              notmuch_message_t **message_ret)
 {
     notmuch_private_status_t status;
     unsigned int doc_id;
 
+    if (message_ret == NULL)
+       return NOTMUCH_STATUS_NULL_POINTER;
+
     if (strlen (message_id) > NOTMUCH_MESSAGE_ID_MAX)
        message_id = _message_id_compressed (notmuch, message_id);
 
@@ -373,14 +376,21 @@ notmuch_database_find_message (notmuch_database_t *notmuch,
                                                       message_id, &doc_id);
 
        if (status == NOTMUCH_PRIVATE_STATUS_NO_DOCUMENT_FOUND)
-           return NULL;
+           *message_ret = NULL;
+       else {
+           *message_ret = _notmuch_message_create (notmuch, notmuch, doc_id,
+                                                   NULL);
+           if (*message_ret == NULL)
+               return NOTMUCH_STATUS_OUT_OF_MEMORY;
+       }
 
-       return _notmuch_message_create (notmuch, notmuch, doc_id, NULL);
+       return NOTMUCH_STATUS_SUCCESS;
     } catch (const Xapian::Error &error) {
        fprintf (stderr, "A Xapian exception occurred finding message: %s.\n",
                 error.get_msg().c_str());
        notmuch->exception_reported = TRUE;
-       return NULL;
+       *message_ret = NULL;
+       return NOTMUCH_STATUS_XAPIAN_EXCEPTION;
     }
 }
 
@@ -495,8 +505,10 @@ _parse_message_id (void *ctx, const char *message_id, const char **next)
  * 'message_id' in the result (to avoid mass confusion when a single
  * message references itself cyclically---and yes, mail messages are
  * not infrequent in the wild that do this---don't ask me why).
-*/
-static void
+ *
+ * Return the last reference parsed, if it is not equal to message_id.
+ */
+static char *
 parse_references (void *ctx,
                  const char *message_id,
                  GHashTable *hash,
@@ -505,7 +517,7 @@ parse_references (void *ctx,
     char *ref;
 
     if (refs == NULL || *refs == '\0')
-       return;
+       return NULL;
 
     while (*refs) {
        ref = _parse_message_id (ctx, refs, &refs);
@@ -513,11 +525,23 @@ parse_references (void *ctx,
        if (ref && strcmp (ref, message_id))
            g_hash_table_insert (hash, ref, NULL);
     }
+
+    /* The return value of this function is used to add a parent
+     * reference to the database.  We should avoid making a message
+     * its own parent, thus the following check.
+     */
+
+    if (ref && strcmp(ref, message_id)) {
+       return ref;
+    } else {
+       return NULL;
+    }
 }
 
-notmuch_database_t *
-notmuch_database_create (const char *path)
+notmuch_status_t
+notmuch_database_create (const char *path, notmuch_database_t **database)
 {
+    notmuch_status_t status = NOTMUCH_STATUS_SUCCESS;
     notmuch_database_t *notmuch = NULL;
     char *notmuch_path = NULL;
     struct stat st;
@@ -525,6 +549,7 @@ notmuch_database_create (const char *path)
 
     if (path == NULL) {
        fprintf (stderr, "Error: Cannot create a database for a NULL path.\n");
+       status = NOTMUCH_STATUS_NULL_POINTER;
        goto DONE;
     }
 
@@ -532,12 +557,14 @@ notmuch_database_create (const char *path)
     if (err) {
        fprintf (stderr, "Error: Cannot create database at %s: %s.\n",
                 path, strerror (errno));
+       status = NOTMUCH_STATUS_FILE_ERROR;
        goto DONE;
     }
 
     if (! S_ISDIR (st.st_mode)) {
        fprintf (stderr, "Error: Cannot create database at %s: Not a directory.\n",
                 path);
+       status = NOTMUCH_STATUS_FILE_ERROR;
        goto DONE;
     }
 
@@ -548,18 +575,30 @@ notmuch_database_create (const char *path)
     if (err) {
        fprintf (stderr, "Error: Cannot create directory %s: %s.\n",
                 notmuch_path, strerror (errno));
+       status = NOTMUCH_STATUS_FILE_ERROR;
        goto DONE;
     }
 
-    notmuch = notmuch_database_open (path,
-                                    NOTMUCH_DATABASE_MODE_READ_WRITE);
-    notmuch_database_upgrade (notmuch, NULL, NULL);
+    status = notmuch_database_open (path,
+                                   NOTMUCH_DATABASE_MODE_READ_WRITE,
+                                   &notmuch);
+    if (status)
+       goto DONE;
+    status = notmuch_database_upgrade (notmuch, NULL, NULL);
+    if (status) {
+       notmuch_database_close(notmuch);
+       notmuch = NULL;
+    }
 
   DONE:
     if (notmuch_path)
        talloc_free (notmuch_path);
 
-    return notmuch;
+    if (database)
+       *database = notmuch;
+    else
+       talloc_free (notmuch);
+    return status;
 }
 
 notmuch_status_t
@@ -573,19 +612,29 @@ _notmuch_database_ensure_writable (notmuch_database_t *notmuch)
     return NOTMUCH_STATUS_SUCCESS;
 }
 
-notmuch_database_t *
+notmuch_status_t
 notmuch_database_open (const char *path,
-                      notmuch_database_mode_t mode)
+                      notmuch_database_mode_t mode,
+                      notmuch_database_t **database)
 {
+    notmuch_status_t status = NOTMUCH_STATUS_SUCCESS;
+    void *local = talloc_new (NULL);
     notmuch_database_t *notmuch = NULL;
-    char *notmuch_path = NULL, *xapian_path = NULL;
+    char *notmuch_path, *xapian_path;
     struct stat st;
     int err;
     unsigned int i, version;
+    static int initialized = 0;
 
-    if (asprintf (&notmuch_path, "%s/%s", path, ".notmuch") == -1) {
-       notmuch_path = NULL;
+    if (path == NULL) {
+       fprintf (stderr, "Error: Cannot open a database for a NULL path.\n");
+       status = NOTMUCH_STATUS_NULL_POINTER;
+       goto DONE;
+    }
+
+    if (! (notmuch_path = talloc_asprintf (local, "%s/%s", path, ".notmuch"))) {
        fprintf (stderr, "Out of memory\n");
+       status = NOTMUCH_STATUS_OUT_OF_MEMORY;
        goto DONE;
     }
 
@@ -593,16 +642,28 @@ notmuch_database_open (const char *path,
     if (err) {
        fprintf (stderr, "Error opening database at %s: %s\n",
                 notmuch_path, strerror (errno));
+       status = NOTMUCH_STATUS_FILE_ERROR;
        goto DONE;
     }
 
-    if (asprintf (&xapian_path, "%s/%s", notmuch_path, "xapian") == -1) {
-       xapian_path = NULL;
+    if (! (xapian_path = talloc_asprintf (local, "%s/%s", notmuch_path, "xapian"))) {
        fprintf (stderr, "Out of memory\n");
+       status = NOTMUCH_STATUS_OUT_OF_MEMORY;
        goto DONE;
     }
 
-    notmuch = talloc (NULL, notmuch_database_t);
+    /* Initialize the GLib type system and threads */
+#if !GLIB_CHECK_VERSION(2, 35, 1)
+    g_type_init ();
+#endif
+
+    /* Initialize gmime */
+    if (! initialized) {
+       g_mime_init (GMIME_ENABLE_RFC2047_WORKAROUNDS);
+       initialized = 1;
+    }
+
+    notmuch = talloc_zero (NULL, notmuch_database_t);
     notmuch->exception_reported = FALSE;
     notmuch->path = talloc_strdup (notmuch, path);
 
@@ -611,6 +672,7 @@ notmuch_database_open (const char *path,
 
     notmuch->needs_upgrade = FALSE;
     notmuch->mode = mode;
+    notmuch->atomic_nesting = 0;
     try {
        string last_thread_id;
 
@@ -627,8 +689,9 @@ notmuch_database_open (const char *path,
                         "       read-write mode.\n",
                         notmuch_path, version, NOTMUCH_DATABASE_VERSION);
                notmuch->mode = NOTMUCH_DATABASE_MODE_READ_ONLY;
-               notmuch_database_close (notmuch);
+               notmuch_database_destroy (notmuch);
                notmuch = NULL;
+               status = NOTMUCH_STATUS_FILE_ERROR;
                goto DONE;
            }
 
@@ -667,12 +730,14 @@ notmuch_database_open (const char *path,
        notmuch->term_gen = new Xapian::TermGenerator;
        notmuch->term_gen->set_stemmer (Xapian::Stem ("english"));
        notmuch->value_range_processor = new Xapian::NumberValueRangeProcessor (NOTMUCH_VALUE_TIMESTAMP);
+       notmuch->date_range_processor = new ParseTimeValueRangeProcessor (NOTMUCH_VALUE_TIMESTAMP);
 
        notmuch->query_parser->set_default_op (Xapian::Query::OP_AND);
        notmuch->query_parser->set_database (*notmuch->xapian_db);
        notmuch->query_parser->set_stemmer (Xapian::Stem ("english"));
        notmuch->query_parser->set_stemming_strategy (Xapian::QueryParser::STEM_SOME);
        notmuch->query_parser->add_valuerangeprocessor (notmuch->value_range_processor);
+       notmuch->query_parser->add_valuerangeprocessor (notmuch->date_range_processor);
 
        for (i = 0; i < ARRAY_SIZE (BOOLEAN_PREFIX_EXTERNAL); i++) {
            prefix_t *prefix = &BOOLEAN_PREFIX_EXTERNAL[i];
@@ -687,23 +752,27 @@ notmuch_database_open (const char *path,
     } catch (const Xapian::Error &error) {
        fprintf (stderr, "A Xapian exception occurred opening database: %s\n",
                 error.get_msg().c_str());
+       notmuch_database_destroy (notmuch);
        notmuch = NULL;
+       status = NOTMUCH_STATUS_XAPIAN_EXCEPTION;
     }
 
   DONE:
-    if (notmuch_path)
-       free (notmuch_path);
-    if (xapian_path)
-       free (xapian_path);
+    talloc_free (local);
 
-    return notmuch;
+    if (database)
+       *database = notmuch;
+    else
+       talloc_free (notmuch);
+    return status;
 }
 
 void
 notmuch_database_close (notmuch_database_t *notmuch)
 {
     try {
-       if (notmuch->mode == NOTMUCH_DATABASE_MODE_READ_WRITE)
+       if (notmuch->xapian_db != NULL &&
+           notmuch->mode == NOTMUCH_DATABASE_MODE_READ_WRITE)
            (static_cast <Xapian::WritableDatabase *> (notmuch->xapian_db))->flush ();
     } catch (const Xapian::Error &error) {
        if (! notmuch->exception_reported) {
@@ -712,10 +781,187 @@ notmuch_database_close (notmuch_database_t *notmuch)
        }
     }
 
+    /* Many Xapian objects (and thus notmuch objects) hold references to
+     * the database, so merely deleting the database may not suffice to
+     * close it.  Thus, we explicitly close it here. */
+    if (notmuch->xapian_db != NULL) {
+       try {
+           notmuch->xapian_db->close();
+       } catch (const Xapian::Error &error) {
+           /* do nothing */
+       }
+    }
+
     delete notmuch->term_gen;
+    notmuch->term_gen = NULL;
     delete notmuch->query_parser;
+    notmuch->query_parser = NULL;
     delete notmuch->xapian_db;
+    notmuch->xapian_db = NULL;
     delete notmuch->value_range_processor;
+    notmuch->value_range_processor = NULL;
+    delete notmuch->date_range_processor;
+    notmuch->date_range_processor = NULL;
+}
+
+#if HAVE_XAPIAN_COMPACT
+static int unlink_cb (const char *path,
+                     unused (const struct stat *sb),
+                     unused (int type),
+                     unused (struct FTW *ftw))
+{
+    return remove(path);
+}
+
+static int rmtree (const char *path)
+{
+    return nftw(path, unlink_cb, 64, FTW_DEPTH | FTW_PHYS);
+}
+
+class NotmuchCompactor : public Xapian::Compactor
+{
+    notmuch_compact_status_cb_t status_cb;
+
+public:
+    NotmuchCompactor(notmuch_compact_status_cb_t cb) : status_cb(cb) { }
+
+    virtual void
+    set_status (const std::string &table, const std::string &status)
+    {
+       char* msg;
+
+       if (status_cb == NULL)
+           return;
+
+       if (status.length() == 0)
+           msg = talloc_asprintf (NULL, "compacting table %s", table.c_str());
+       else
+           msg = talloc_asprintf (NULL, "     %s", status.c_str());
+
+       if (msg == NULL) {
+           return;
+       }
+
+       status_cb(msg);
+       talloc_free(msg);
+    }
+};
+
+/* Compacts the given database, optionally saving the original database
+ * in backup_path. Additionally, a callback function can be provided to
+ * give the user feedback on the progress of the (likely long-lived)
+ * compaction process.
+ *
+ * The backup path must point to a directory on the same volume as the
+ * original database. Passing a NULL backup_path will result in the
+ * uncompacted database being deleted after compaction has finished.
+ * Note that the database write lock will be held during the
+ * compaction process to protect data integrity.
+ */
+notmuch_status_t
+notmuch_database_compact (const char* path,
+                         const char* backup_path,
+                         notmuch_compact_status_cb_t status_cb)
+{
+    void *local;
+    char *notmuch_path, *xapian_path, *compact_xapian_path;
+    char *old_xapian_path = NULL;
+    notmuch_status_t ret = NOTMUCH_STATUS_SUCCESS;
+    notmuch_database_t *notmuch = NULL;
+    struct stat statbuf;
+
+    local = talloc_new (NULL);
+    if (! local)
+       return NOTMUCH_STATUS_OUT_OF_MEMORY;
+
+    ret = notmuch_database_open(path, NOTMUCH_DATABASE_MODE_READ_WRITE, &notmuch);
+    if (ret) {
+       goto DONE;
+    }
+
+    if (! (notmuch_path = talloc_asprintf (local, "%s/%s", path, ".notmuch"))) {
+       ret = NOTMUCH_STATUS_OUT_OF_MEMORY;
+       goto DONE;
+    }
+
+    if (! (xapian_path = talloc_asprintf (local, "%s/%s", notmuch_path, "xapian"))) {
+       ret = NOTMUCH_STATUS_OUT_OF_MEMORY;
+       goto DONE;
+    }
+
+    if (! (compact_xapian_path = talloc_asprintf (local, "%s.compact", xapian_path))) {
+       ret = NOTMUCH_STATUS_OUT_OF_MEMORY;
+       goto DONE;
+    }
+
+    if (backup_path != NULL) {
+       if (! (old_xapian_path = talloc_asprintf (local, "%s/xapian.old", backup_path))) {
+           ret = NOTMUCH_STATUS_OUT_OF_MEMORY;
+           goto DONE;
+       }
+
+       if (stat(old_xapian_path, &statbuf) != -1) {
+           fprintf (stderr, "Backup path already exists: %s\n", old_xapian_path);
+           ret = NOTMUCH_STATUS_FILE_ERROR;
+           goto DONE;
+       }
+       if (errno != ENOENT) {
+           fprintf (stderr, "Unknown error while stat()ing backup path: %s\n",
+                    strerror(errno));
+           goto DONE;
+       }
+    }
+
+    try {
+       NotmuchCompactor compactor(status_cb);
+
+       compactor.set_renumber(false);
+       compactor.add_source(xapian_path);
+       compactor.set_destdir(compact_xapian_path);
+       compactor.compact();
+    } catch (Xapian::InvalidArgumentError e) {
+       fprintf (stderr, "Error while compacting: %s\n", e.get_msg().c_str());
+       ret = NOTMUCH_STATUS_XAPIAN_EXCEPTION;
+       goto DONE;
+    }
+
+    if (old_xapian_path != NULL) {
+       if (rename(xapian_path, old_xapian_path)) {
+           fprintf (stderr, "Error moving old database out of the way\n");
+           ret = NOTMUCH_STATUS_FILE_ERROR;
+           goto DONE;
+       }
+    } else {
+       rmtree(xapian_path);
+    }
+
+    if (rename(compact_xapian_path, xapian_path)) {
+       fprintf (stderr, "Error moving compacted database\n");
+       ret = NOTMUCH_STATUS_FILE_ERROR;
+       goto DONE;
+    }
+
+    notmuch_database_close(notmuch);
+
+DONE:
+    talloc_free(local);
+    return ret;
+}
+#else
+notmuch_status_t
+notmuch_database_compact (unused (const char* path),
+                         unused (const char* backup_path),
+                         unused (notmuch_compact_status_cb_t status_cb))
+{
+    fprintf (stderr, "notmuch was compiled against a xapian version lacking compaction support.\n");
+    return NOTMUCH_STATUS_UNSUPPORTED_OPERATION;
+}
+#endif
+
+void
+notmuch_database_destroy (notmuch_database_t *notmuch)
+{
+    notmuch_database_close (notmuch);
     talloc_free (notmuch);
 }
 
@@ -887,8 +1133,8 @@ notmuch_database_upgrade (notmuch_database_t *notmuch,
                mtime = Xapian::sortable_unserialise (
                    document.get_value (NOTMUCH_VALUE_TIMESTAMP));
 
-               directory = notmuch_database_get_directory (notmuch,
-                                                           term.c_str() + 10);
+               directory = _notmuch_directory_create (notmuch, term.c_str() + 10,
+                                                      NOTMUCH_FIND_CREATE, &status);
                notmuch_directory_set_mtime (directory, mtime);
                notmuch_directory_destroy (directory);
            }
@@ -974,6 +1220,61 @@ notmuch_database_upgrade (notmuch_database_t *notmuch,
     return NOTMUCH_STATUS_SUCCESS;
 }
 
+notmuch_status_t
+notmuch_database_begin_atomic (notmuch_database_t *notmuch)
+{
+    if (notmuch->mode == NOTMUCH_DATABASE_MODE_READ_ONLY ||
+       notmuch->atomic_nesting > 0)
+       goto DONE;
+
+    try {
+       (static_cast <Xapian::WritableDatabase *> (notmuch->xapian_db))->begin_transaction (false);
+    } catch (const Xapian::Error &error) {
+       fprintf (stderr, "A Xapian exception occurred beginning transaction: %s.\n",
+                error.get_msg().c_str());
+       notmuch->exception_reported = TRUE;
+       return NOTMUCH_STATUS_XAPIAN_EXCEPTION;
+    }
+
+DONE:
+    notmuch->atomic_nesting++;
+    return NOTMUCH_STATUS_SUCCESS;
+}
+
+notmuch_status_t
+notmuch_database_end_atomic (notmuch_database_t *notmuch)
+{
+    Xapian::WritableDatabase *db;
+
+    if (notmuch->atomic_nesting == 0)
+       return NOTMUCH_STATUS_UNBALANCED_ATOMIC;
+
+    if (notmuch->mode == NOTMUCH_DATABASE_MODE_READ_ONLY ||
+       notmuch->atomic_nesting > 1)
+       goto DONE;
+
+    db = static_cast <Xapian::WritableDatabase *> (notmuch->xapian_db);
+    try {
+       db->commit_transaction ();
+
+       /* This is a hack for testing.  Xapian never flushes on a
+        * non-flushed commit, even if the flush threshold is 1.
+        * However, we rely on flushing to test atomicity. */
+       const char *thresh = getenv ("XAPIAN_FLUSH_THRESHOLD");
+       if (thresh && atoi (thresh) == 1)
+           db->flush ();
+    } catch (const Xapian::Error &error) {
+       fprintf (stderr, "A Xapian exception occurred committing transaction: %s.\n",
+                error.get_msg().c_str());
+       notmuch->exception_reported = TRUE;
+       return NOTMUCH_STATUS_XAPIAN_EXCEPTION;
+    }
+
+DONE:
+    notmuch->atomic_nesting--;
+    return NOTMUCH_STATUS_SUCCESS;
+}
+
 /* We allow the user to use arbitrarily long paths for directories. But
  * we have a term-length limit. So if we exceed that, we'll use the
  * SHA-1 of the path for the database term.
@@ -1074,9 +1375,17 @@ _notmuch_database_split_path (void *ctx,
     return NOTMUCH_STATUS_SUCCESS;
 }
 
+/* Find the document ID of the specified directory.
+ *
+ * If (flags & NOTMUCH_FIND_CREATE), a new directory document will be
+ * created if one does not exist for 'path'.  Otherwise, if the
+ * directory document does not exist, this sets *directory_id to
+ * ((unsigned int)-1) and returns NOTMUCH_STATUS_SUCCESS.
+ */
 notmuch_status_t
 _notmuch_database_find_directory_id (notmuch_database_t *notmuch,
                                     const char *path,
+                                    notmuch_find_flags_t flags,
                                     unsigned int *directory_id)
 {
     notmuch_directory_t *directory;
@@ -1087,8 +1396,8 @@ _notmuch_database_find_directory_id (notmuch_database_t *notmuch,
        return NOTMUCH_STATUS_SUCCESS;
     }
 
-    directory = _notmuch_directory_create (notmuch, path, &status);
-    if (status) {
+    directory = _notmuch_directory_create (notmuch, path, flags, &status);
+    if (status || !directory) {
        *directory_id = -1;
        return status;
     }
@@ -1117,13 +1426,16 @@ _notmuch_database_get_directory_path (void *ctx,
  * database path), return a new string (with 'ctx' as the talloc
  * owner) suitable for use as a direntry term value.
  *
- * The necessary directory documents will be created in the database
- * as needed.
+ * If (flags & NOTMUCH_FIND_CREATE), the necessary directory documents
+ * will be created in the database as needed.  Otherwise, if the
+ * necessary directory documents do not exist, this sets
+ * *direntry to NULL and returns NOTMUCH_STATUS_SUCCESS.
  */
 notmuch_status_t
 _notmuch_database_filename_to_direntry (void *ctx,
                                        notmuch_database_t *notmuch,
                                        const char *filename,
+                                       notmuch_find_flags_t flags,
                                        char **direntry)
 {
     const char *relative, *directory, *basename;
@@ -1137,10 +1449,12 @@ _notmuch_database_filename_to_direntry (void *ctx,
     if (status)
        return status;
 
-    status = _notmuch_database_find_directory_id (notmuch, directory,
+    status = _notmuch_database_find_directory_id (notmuch, directory, flags,
                                                  &directory_id);
-    if (status)
+    if (status || directory_id == (unsigned int)-1) {
+       *direntry = NULL;
        return status;
+    }
 
     *direntry = talloc_asprintf (ctx, "%u:%s", directory_id, basename);
 
@@ -1181,20 +1495,27 @@ _notmuch_database_relative_path (notmuch_database_t *notmuch,
     return relative;
 }
 
-notmuch_directory_t *
+notmuch_status_t
 notmuch_database_get_directory (notmuch_database_t *notmuch,
-                               const char *path)
+                               const char *path,
+                               notmuch_directory_t **directory)
 {
     notmuch_status_t status;
 
+    if (directory == NULL)
+       return NOTMUCH_STATUS_NULL_POINTER;
+    *directory = NULL;
+
     try {
-       return _notmuch_directory_create (notmuch, path, &status);
+       *directory = _notmuch_directory_create (notmuch, path,
+                                               NOTMUCH_FIND_LOOKUP, &status);
     } catch (const Xapian::Error &error) {
        fprintf (stderr, "A Xapian exception occurred getting directory: %s.\n",
                 error.get_msg().c_str());
        notmuch->exception_reported = TRUE;
-       return NULL;
+       status = NOTMUCH_STATUS_XAPIAN_EXCEPTION;
     }
+    return status;
 }
 
 /* Allocate a document ID that satisfies the following criteria:
@@ -1253,7 +1574,9 @@ _get_metadata_thread_id_key (void *ctx, const char *message_id)
 
 /* Find the thread ID to which the message with 'message_id' belongs.
  *
- * Always returns a newly talloced string belonging to 'ctx'.
+ * Note: 'thread_id_ret' must not be NULL!
+ * On success '*thread_id_ret' is set to a newly talloced string belonging to
+ * 'ctx'.
  *
  * Note: If there is no message in the database with the given
  * 'message_id' then a new thread_id will be allocated for this
@@ -1261,25 +1584,30 @@ _get_metadata_thread_id_key (void *ctx, const char *message_id)
  * thread ID can be looked up if the message is added to the database
  * later).
  */
-static const char *
+static notmuch_status_t
 _resolve_message_id_to_thread_id (notmuch_database_t *notmuch,
                                  void *ctx,
-                                 const char *message_id)
+                                 const char *message_id,
+                                 const char **thread_id_ret)
 {
+    notmuch_status_t status;
     notmuch_message_t *message;
     string thread_id_string;
-    const char *thread_id;
     char *metadata_key;
     Xapian::WritableDatabase *db;
 
-    message = notmuch_database_find_message (notmuch, message_id);
+    status = notmuch_database_find_message (notmuch, message_id, &message);
+
+    if (status)
+       return status;
 
     if (message) {
-       thread_id = talloc_steal (ctx, notmuch_message_get_thread_id (message));
+       *thread_id_ret = talloc_steal (ctx,
+                                      notmuch_message_get_thread_id (message));
 
        notmuch_message_destroy (message);
 
-       return thread_id;
+       return NOTMUCH_STATUS_SUCCESS;
     }
 
     /* Message has not been seen yet.
@@ -1293,15 +1621,16 @@ _resolve_message_id_to_thread_id (notmuch_database_t *notmuch,
     thread_id_string = notmuch->xapian_db->get_metadata (metadata_key);
 
     if (thread_id_string.empty()) {
-       thread_id = _notmuch_database_generate_thread_id (notmuch);
-       db->set_metadata (metadata_key, thread_id);
+       *thread_id_ret = talloc_strdup (ctx,
+                                       _notmuch_database_generate_thread_id (notmuch));
+       db->set_metadata (metadata_key, *thread_id_ret);
     } else {
-       thread_id = thread_id_string.c_str();
+       *thread_id_ret = talloc_strdup (ctx, thread_id_string.c_str());
     }
 
     talloc_free (metadata_key);
 
-    return talloc_strdup (ctx, thread_id);
+    return NOTMUCH_STATUS_SUCCESS;
 }
 
 static notmuch_status_t
@@ -1354,43 +1683,51 @@ _notmuch_database_link_message_to_parents (notmuch_database_t *notmuch,
 {
     GHashTable *parents = NULL;
     const char *refs, *in_reply_to, *in_reply_to_message_id;
+    const char *last_ref_message_id, *this_message_id;
     GList *l, *keys = NULL;
     notmuch_status_t ret = NOTMUCH_STATUS_SUCCESS;
 
     parents = g_hash_table_new_full (g_str_hash, g_str_equal,
                                     _my_talloc_free_for_g_hash, NULL);
+    this_message_id = notmuch_message_get_message_id (message);
 
     refs = notmuch_message_file_get_header (message_file, "references");
-    parse_references (message, notmuch_message_get_message_id (message),
-                     parents, refs);
+    last_ref_message_id = parse_references (message,
+                                           this_message_id,
+                                           parents, refs);
 
     in_reply_to = notmuch_message_file_get_header (message_file, "in-reply-to");
-    parse_references (message, notmuch_message_get_message_id (message),
-                     parents, in_reply_to);
-
-    /* Carefully avoid adding any self-referential in-reply-to term. */
-    in_reply_to_message_id = _parse_message_id (message, in_reply_to, NULL);
-    if (in_reply_to_message_id &&
-       strcmp (in_reply_to_message_id,
-               notmuch_message_get_message_id (message)))
-    {
+    in_reply_to_message_id = parse_references (message,
+                                              this_message_id,
+                                              parents, in_reply_to);
+
+    /* For the parent of this message, use the last message ID of the
+     * References header, if available.  If not, fall back to the
+     * first message ID in the In-Reply-To header. */
+    if (last_ref_message_id) {
+        _notmuch_message_add_term (message, "replyto",
+                                   last_ref_message_id);
+    } else if (in_reply_to_message_id) {
        _notmuch_message_add_term (message, "replyto",
-                            _parse_message_id (message, in_reply_to, NULL));
+                            in_reply_to_message_id);
     }
 
     keys = g_hash_table_get_keys (parents);
     for (l = keys; l; l = l->next) {
        char *parent_message_id;
-       const char *parent_thread_id;
+       const char *parent_thread_id = NULL;
 
        parent_message_id = (char *) l->data;
 
        _notmuch_message_add_term (message, "reference",
                                   parent_message_id);
 
-       parent_thread_id = _resolve_message_id_to_thread_id (notmuch,
-                                                            message,
-                                                            parent_message_id);
+       ret = _resolve_message_id_to_thread_id (notmuch,
+                                               message,
+                                               parent_message_id,
+                                               &parent_thread_id);
+       if (ret)
+           goto DONE;
 
        if (*thread_id == NULL) {
            *thread_id = talloc_strdup (message, parent_thread_id);
@@ -1543,7 +1880,7 @@ notmuch_database_add_message (notmuch_database_t *notmuch,
 {
     notmuch_message_file_t *message_file;
     notmuch_message_t *message = NULL;
-    notmuch_status_t ret = NOTMUCH_STATUS_SUCCESS;
+    notmuch_status_t ret = NOTMUCH_STATUS_SUCCESS, ret2;
     notmuch_private_status_t private_status;
 
     const char *date, *header;
@@ -1561,6 +1898,12 @@ notmuch_database_add_message (notmuch_database_t *notmuch,
     if (message_file == NULL)
        return NOTMUCH_STATUS_FILE_ERROR;
 
+    /* Adding a message may change many documents.  Do this all
+     * atomically. */
+    ret = notmuch_database_begin_atomic (notmuch);
+    if (ret)
+       goto DONE;
+
     notmuch_message_file_restrict_headers (message_file,
                                           "date",
                                           "from",
@@ -1654,9 +1997,11 @@ notmuch_database_add_message (notmuch_database_t *notmuch,
                goto DONE;
 
            date = notmuch_message_file_get_header (message_file, "date");
-           _notmuch_message_set_date (message, date);
+           _notmuch_message_set_header_values (message, date, from, subject);
 
-           _notmuch_message_index_file (message, filename);
+           ret = _notmuch_message_index_file (message, filename);
+           if (ret)
+               goto DONE;
        } else {
            ret = NOTMUCH_STATUS_DUPLICATE_MESSAGE_ID;
        }
@@ -1682,6 +2027,12 @@ notmuch_database_add_message (notmuch_database_t *notmuch,
     if (message_file)
        notmuch_message_file_close (message_file);
 
+    ret2 = notmuch_database_end_atomic (notmuch);
+    if ((ret == NOTMUCH_STATUS_SUCCESS ||
+        ret == NOTMUCH_STATUS_DUPLICATE_MESSAGE_ID) &&
+       ret2 != NOTMUCH_STATUS_SUCCESS)
+       ret = ret2;
+
     return ret;
 }
 
@@ -1689,71 +2040,76 @@ notmuch_status_t
 notmuch_database_remove_message (notmuch_database_t *notmuch,
                                 const char *filename)
 {
-    Xapian::WritableDatabase *db;
+    notmuch_status_t status;
+    notmuch_message_t *message;
+
+    status = notmuch_database_find_message_by_filename (notmuch, filename,
+                                                       &message);
+
+    if (status == NOTMUCH_STATUS_SUCCESS && message) {
+           status = _notmuch_message_remove_filename (message, filename);
+           if (status == NOTMUCH_STATUS_SUCCESS)
+               _notmuch_message_delete (message);
+           else if (status == NOTMUCH_STATUS_DUPLICATE_MESSAGE_ID)
+               _notmuch_message_sync (message);
+
+           notmuch_message_destroy (message);
+    }
+
+    return status;
+}
+
+notmuch_status_t
+notmuch_database_find_message_by_filename (notmuch_database_t *notmuch,
+                                          const char *filename,
+                                          notmuch_message_t **message_ret)
+{
     void *local;
     const char *prefix = _find_prefix ("file-direntry");
     char *direntry, *term;
     Xapian::PostingIterator i, end;
-    Xapian::Document document;
     notmuch_status_t status;
 
-    status = _notmuch_database_ensure_writable (notmuch);
-    if (status)
-       return status;
+    if (message_ret == NULL)
+       return NOTMUCH_STATUS_NULL_POINTER;
 
-    local = talloc_new (notmuch);
+    /* return NULL on any failure */
+    *message_ret = NULL;
 
-    db = static_cast <Xapian::WritableDatabase *> (notmuch->xapian_db);
+    local = talloc_new (notmuch);
 
     try {
-
-       status = _notmuch_database_filename_to_direntry (local, notmuch,
-                                                        filename, &direntry);
-       if (status)
-           return status;
+       status = _notmuch_database_filename_to_direntry (
+           local, notmuch, filename, NOTMUCH_FIND_LOOKUP, &direntry);
+       if (status || !direntry)
+           goto DONE;
 
        term = talloc_asprintf (local, "%s%s", prefix, direntry);
 
        find_doc_ids_for_term (notmuch, term, &i, &end);
 
-       for ( ; i != end; i++) {
-           Xapian::TermIterator j;
-           notmuch_message_t *message;
+       if (i != end) {
            notmuch_private_status_t private_status;
 
-           message = _notmuch_message_create (local, notmuch,
-                                              *i, &private_status);
-           if (message == NULL)
-               return COERCE_STATUS (private_status,
-                                     "Inconsistent document ID in datbase.");
-
-           _notmuch_message_remove_filename (message, filename);
-           _notmuch_message_sync (message);
-
-           /* Take care to find document after sync'ing filename removal. */
-           document = find_document_for_doc_id (notmuch, *i);
-           j = document.termlist_begin ();
-           j.skip_to (prefix);
-
-           /* Was this the last file-direntry in the message? */
-           if (j == document.termlist_end () ||
-               strncmp ((*j).c_str (), prefix, strlen (prefix)))
-           {
-               db->delete_document (document.get_docid ());
-               status = NOTMUCH_STATUS_SUCCESS;
-           } else {
-               status = NOTMUCH_STATUS_DUPLICATE_MESSAGE_ID;
-           }
+           *message_ret = _notmuch_message_create (notmuch, notmuch, *i,
+                                                   &private_status);
+           if (*message_ret == NULL)
+               status = NOTMUCH_STATUS_OUT_OF_MEMORY;
        }
     } catch (const Xapian::Error &error) {
-       fprintf (stderr, "Error: A Xapian exception occurred removing message: %s\n",
+       fprintf (stderr, "Error: A Xapian exception occurred finding message by filename: %s\n",
                 error.get_msg().c_str());
        notmuch->exception_reported = TRUE;
        status = NOTMUCH_STATUS_XAPIAN_EXCEPTION;
     }
 
+  DONE:
     talloc_free (local);
 
+    if (status && *message_ret) {
+       notmuch_message_destroy (*message_ret);
+       *message_ret = NULL;
+    }
     return status;
 }