lobo/zetviel
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages 1 Wiki Activity Actions

attempt to fix concurrency issue crashing server

Browse source
This commit is contained in:
Emil Lerch 2026-07-22 06:21:45 -07:00
parent 90542d8c5b
commit 61efcae568
Signed by: lobo
GPG key ID: A7B62D657EF764F8
4 changed files with 140 additions and 36 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

61
src/Email.zig
View file

@ -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"),
});
}
};

63
src/main.zig
View file

@ -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);
}

13
src/notmuch.zig
View file

@ -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

39
src/root.zig
View file

@ -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;