diff --git a/src/Email.zig b/src/Email.zig index 3e77f31..557080f 100644 --- a/src/Email.zig +++ b/src/Email.zig @@ -368,7 +368,7 @@ pub const Message = struct { } fn collectAttachments(part: *gmime.GMimeObject, list: *std.ArrayList(AttachmentInfo), allocator: std.mem.Allocator) !void { - // Check if this is a multipart + // Recurse into multipart containers. if (gmime.g_type_check_instance_is_a(@as(*gmime.GTypeInstance, @ptrCast(part)), gmime.g_mime_multipart_get_type()) != 0) { const multipart: *gmime.GMimeMultipart = @ptrCast(part); const count_i = gmime.g_mime_multipart_get_count(multipart); @@ -384,31 +384,46 @@ pub const Message = struct { return; } - // Check if this part is an attachment + // Only parts carrying an attachment/inline content-disposition are + // treated as attachments. const disposition = gmime.g_mime_object_get_content_disposition(part); - if (disposition != null) { - const disp_str = gmime.g_mime_content_disposition_get_disposition(disposition); - if (disp_str != null and (std.mem.eql(u8, std.mem.span(disp_str), "attachment") or - std.mem.eql(u8, std.mem.span(disp_str), "inline"))) - { - const filename_c = gmime.g_mime_part_get_filename(@as(*gmime.GMimePart, @ptrCast(part))); - if (filename_c != null) { - const content_type_obj = gmime.g_mime_object_get_content_type(part); - const mime_type = if (content_type_obj != null) - gmime.g_mime_content_type_get_mime_type(content_type_obj) - else - null; + if (disposition == null) return; + const disp_str = gmime.g_mime_content_disposition_get_disposition(disposition); + if (disp_str == null) return; + if (!std.mem.eql(u8, std.mem.span(disp_str), "attachment") and + !std.mem.eql(u8, std.mem.span(disp_str), "inline")) return; - try list.append(allocator, .{ - .filename = try allocator.dupe(u8, std.mem.span(filename_c)), - .content_type = if (mime_type != null) - try allocator.dupe(u8, std.mem.span(mime_type)) - else - try allocator.dupe(u8, "application/octet-stream"), - }); - } - } + const content_type_obj = gmime.g_mime_object_get_content_type(part); + const mime_type = if (content_type_obj != null) + gmime.g_mime_content_type_get_mime_type(content_type_obj) + else + null; + defer if (mime_type != null) gmime.g_free(mime_type); + + // g_mime_part_get_filename() asserts GMIME_IS_PART(part). Calling it on + // a non-part -- notably a message/rfc822 GMimeMessagePart, which is + // common in forwarded mail (e.g. Yahoo) -- prints a gmime-CRITICAL + // assertion failure and returns NULL, silently dropping the + // attachment. Check the concrete GObject type before dispatching. + var filename: ?[]const u8 = null; + if (gmime.g_type_check_instance_is_a(@as(*gmime.GTypeInstance, @ptrCast(part)), gmime.g_mime_part_get_type()) != 0) { + const filename_c = gmime.g_mime_part_get_filename(@as(*gmime.GMimePart, @ptrCast(part))); + if (filename_c != null) filename = std.mem.span(filename_c); + } else if (gmime.g_type_check_instance_is_a(@as(*gmime.GTypeInstance, @ptrCast(part)), gmime.g_mime_message_part_get_type()) != 0) { + // Attached/forwarded message (message/rfc822): surface it as a + // single .eml attachment rather than recursing into it (which would + // scramble attachment indices used by /api/attachment/:id/:num). + filename = "attached-message.eml"; } + if (filename == null) return; + + try list.append(allocator, .{ + .filename = try allocator.dupe(u8, filename.?), + .content_type = if (mime_type != null) + try allocator.dupe(u8, std.mem.span(mime_type)) + else + try allocator.dupe(u8, "application/octet-stream"), + }); } }; diff --git a/src/main.zig b/src/main.zig index 1ee54d3..31c0476 100644 --- a/src/main.zig +++ b/src/main.zig @@ -5,12 +5,6 @@ const auth = @import("auth.zig"); const version = @import("build_options").git_revision; -fn exitAfterDelay() void { - std.Thread.sleep(500 * std.time.ns_per_ms); - std.log.err("Notmuch search is in unrecoverable state: exiting", .{}); - std.process.exit(1); -} - pub fn main() !u8 { var gpa = std.heap.GeneralPurposeAllocator(.{}){}; defer _ = gpa.deinit(); @@ -22,7 +16,7 @@ pub fn main() !u8 { const stdout = &stdout_f.interface; // SAFETY: buffer to be used by writer var stderr_buffer: [1024]u8 = undefined; - var stderr_f = std.fs.File.stdout().writer(&stderr_buffer); + var stderr_f = std.fs.File.stderr().writer(&stderr_buffer); const stderr = &stderr_f.interface; errdefer stdout.flush() catch @panic("could not flush stdout"); @@ -202,6 +196,12 @@ const SecurityHeaders = struct { }; fn queryHandler(db: *root.NotmuchDb, req: *httpz.Request, res: *httpz.Response) !void { + // notmuch/Xapian is not thread-safe and the DB is touched lazily during + // res.json below, so hold the lock across the whole handler. See + // root.NotmuchDb.mutex. + db.mutex.lock(); + defer db.mutex.unlock(); + const encoded_query = req.url.path[11..]; // Skip "/api/query/" if (encoded_query.len == 0) { res.status = 400; @@ -216,9 +216,7 @@ fn queryHandler(db: *root.NotmuchDb, req: *httpz.Request, res: *httpz.Response) var threads = db.search(query) catch |err| { if (err == error.CouldNotSearchThreads) { res.status = 503; - try res.json(.{ .@"error" = "CouldNotSearchThreads", .fatal = true }, .{}); - const exit_thread = std.Thread.spawn(.{}, exitAfterDelay, .{}) catch @panic("could not spawn thread to kill process"); - exit_thread.detach(); + try res.json(.{ .@"error" = "CouldNotSearchThreads" }, .{}); return; } res.status = 500; @@ -272,6 +270,9 @@ fn queryHandler(db: *root.NotmuchDb, req: *httpz.Request, res: *httpz.Response) } fn threadHandler(db: *root.NotmuchDb, req: *httpz.Request, res: *httpz.Response) !void { + db.mutex.lock(); + defer db.mutex.unlock(); + const thread_id = req.param("thread_id") orelse { res.status = 400; try res.json(.{ .@"error" = "Thread ID required" }, .{}); @@ -289,6 +290,9 @@ fn threadHandler(db: *root.NotmuchDb, req: *httpz.Request, res: *httpz.Response) } fn messageHandler(db: *root.NotmuchDb, req: *httpz.Request, res: *httpz.Response) !void { + db.mutex.lock(); + defer db.mutex.unlock(); + const message_id = req.param("message_id") orelse { res.status = 400; try res.json(.{ .@"error" = "Message ID required" }, .{}); @@ -306,6 +310,9 @@ fn messageHandler(db: *root.NotmuchDb, req: *httpz.Request, res: *httpz.Response } fn attachmentHandler(db: *root.NotmuchDb, req: *httpz.Request, res: *httpz.Response) !void { + db.mutex.lock(); + defer db.mutex.unlock(); + const message_id = req.param("message_id") orelse { res.status = 400; try res.json(.{ .@"error" = "Message ID required" }, .{}); @@ -418,3 +425,39 @@ test "threadHandler with valid thread" { try threadHandler(&db, t.req, t.res); try std.testing.expect(t.res.status != 404); } + +test "concurrent query handlers are thread-safe (regression: notmuch/Xapian SIGSEGV)" { + // Before NotmuchDb.mutex, httpz's handler thread pool called handlers on + // the single shared notmuch/Xapian handle concurrently. notmuch is not + // thread-safe, so concurrent requests corrupted Xapian's memory and the + // process segfaulted inside libxapian.so (exit 139). Hammer the *real* + // handler from many threads sharing one NotmuchDb: without the lock this + // reliably crashes; with it, it passes. Because it drives the actual + // handler, a future handler that forgets to take the lock also fails here. + const allocator = std.testing.allocator; + var db = try root.openNotmuchDb(allocator, "mail", null); + defer db.close(); + + const Worker = struct { + fn run(shared: *root.NotmuchDb) void { + var n: usize = 0; + while (n < 100) : (n += 1) { + var t = httpz.testing.init(.{}); + defer t.deinit(); + t.url("/api/query/*"); + // A returned error just means that request produced an error + // status; a locking bug shows up as a segfault, not an error. + queryHandler(shared, t.req, t.res) catch {}; + } + } + }; + + const thread_count = 16; + var threads: [thread_count]std.Thread = undefined; + var started: usize = 0; + while (started < thread_count) : (started += 1) { + threads[started] = std.Thread.spawn(.{}, Worker.run, .{&db}) catch break; + } + for (threads[0..started]) |th| th.join(); + try std.testing.expect(started > 0); +} diff --git a/src/notmuch.zig b/src/notmuch.zig index 90ac141..b554230 100644 --- a/src/notmuch.zig +++ b/src/notmuch.zig @@ -81,6 +81,19 @@ pub const Db = struct { db.handle = undefined; } + /// Refresh the read-only view of the database. + /// + /// A notmuch read handle sees a snapshot of the Xapian database as it was + /// at open time. When another process modifies the database (e.g. mail + /// delivery running `notmuch new`), operations on the stale handle can + /// begin returning Xapian exceptions. Calling reopen refreshes the + /// snapshot to the latest committed revision. The caller is responsible + /// for ensuring no other thread is using the handle concurrently. + pub fn reopen(db: *Db) !void { + const status = c.notmuch_database_reopen(db.handle, c.NOTMUCH_DATABASE_MODE_READ_ONLY); + if (status != c.NOTMUCH_STATUS_SUCCESS) return error.CouldNotReopenDatabase; + } + // // Execute a query for threads, returning a notmuch_threads_t object // which can be used to iterate over the results. The returned threads diff --git a/src/root.zig b/src/root.zig index 74b13a5..2e1090d 100644 --- a/src/root.zig +++ b/src/root.zig @@ -168,6 +168,20 @@ pub const NotmuchDb = struct { /// parameter. email_owned: bool, + /// Serializes all access to the notmuch/Xapian handle. + /// + /// notmuch and its underlying Xapian database are NOT thread-safe: a + /// single database handle -- and everything derived from it (queries, + /// threads, messages, tags) -- must not be used from multiple threads + /// concurrently, or Xapian corrupts its own memory and the process + /// segfaults. httpz dispatches requests across a thread pool (32 threads + /// by default), so every handler that touches the database must hold this + /// mutex for the FULL duration of the request. In particular the lock must + /// span res.json(): Threads/Thread jsonStringify iterate the database + /// lazily while serializing, so releasing the lock after search()/ + /// getThread() returns would still race. + mutex: std.Thread.Mutex = .{}, + pub fn close(self: *NotmuchDb) void { self.db.close(); self.db.deinit(); @@ -175,11 +189,30 @@ pub const NotmuchDb = struct { if (self.email_owned) self.email.deinit(); } + /// Run a threads query, recovering once from a stale database snapshot. + /// + /// A read-only notmuch handle sees a snapshot of the database as of open + /// time. Mail delivery runs `notmuch new`, and a query issued against a + /// snapshot that has since been rewritten can fail with a Xapian error. + /// When that happens we reopen to the latest revision and retry once, + /// rather than killing the whole process as the previous code did (it + /// called std.process.exit and leaned on the container restart policy). If + /// the reopen itself fails we surface the original query error. + /// + /// Callers must hold `self.mutex`. + fn openThreads(self: *NotmuchDb, query_z: [:0]const u8) !notmuch.Db.ThreadIterator { + return self.db.searchThreads(query_z) catch |first_err| { + self.db.reopen() catch return first_err; + return self.db.searchThreads(query_z); + }; + } + pub fn search(self: *NotmuchDb, query: []const u8) !Threads { var query_buf: [1024:0]u8 = undefined; const query_z = try std.fmt.bufPrintZ(&query_buf, "{s}", .{query}); const ti = try self.allocator.create(notmuch.Db.ThreadIterator); - ti.* = try self.db.searchThreads(query_z); + errdefer self.allocator.destroy(ti); + ti.* = try self.openThreads(query_z); return Threads.init(self.allocator, ti); } @@ -188,7 +221,7 @@ pub const NotmuchDb = struct { const query_z = try std.fmt.bufPrintZ(&query_buf, "thread:{s}", .{thread_id}); const iter_ptr = try self.allocator.create(notmuch.Db.ThreadIterator); errdefer self.allocator.destroy(iter_ptr); - iter_ptr.* = try self.db.searchThreads(query_z); + iter_ptr.* = try self.openThreads(query_z); errdefer iter_ptr.deinit(); const thread = iter_ptr.next(); @@ -235,7 +268,7 @@ pub const NotmuchDb = struct { pub fn getMessage(self: *NotmuchDb, message_id: []const u8) !MessageDetail { var query_buf: [1024:0]u8 = undefined; const query_z = try std.fmt.bufPrintZ(&query_buf, "mid:{s}", .{message_id}); - var thread_iter = try self.db.searchThreads(query_z); + var thread_iter = try self.openThreads(query_z); defer thread_iter.deinit(); const thread = thread_iter.next() orelse return error.MessageNotFound;