From e2c25a986b0a6445e9349678a599cf4cf1642e6d Mon Sep 17 00:00:00 2001 From: Aleksey Shakhmatov Date: Mon, 3 Aug 2026 18:55:50 +0300 Subject: [PATCH] wire/server: honour moreToCome on OP_MSG requests `Message.flags` was parsed and stored but never read. An OP_MSG request with moreToCome set is fire-and-forget: the client will not read a reply. Sending one anyway leaves it unread in the socket, so the next command on that connection reads the previous command's reply and waits forever for its own. This is not a corner case. Every unacknowledged write uses it, and the Node driver sends `endSessions` with `writeConcern: {w: 0}` whenever a client closes -- so an ordinary application that never asks for w:0 still hits it. Before: insertOne({w: 0}) -> ok, acknowledged=false countDocuments() (same conn) -> BSON element "cursor" is missing The command still runs; only the reply is suppressed. The e2e case pins maxPoolSize to 1, because with a larger pool the driver may hand the next operation a different connection and hide the bug. It asserts the connection still works afterwards, which is the part that matters -- not that the unacknowledged write itself returned. --- src/server.zig | 18 ++++++++++++++++++ src/wire.zig | 12 ++++++++++++ tests/e2e/e2e.js | 42 ++++++++++++++++++++++++++++++++++++++++++ 3 files changed, 72 insertions(+) diff --git a/src/server.zig b/src/server.zig index d94fcc1..5a86042 100644 --- a/src/server.zig +++ b/src/server.zig @@ -170,6 +170,24 @@ fn handle_connection_inner(io: std.Io, stream: std.Io.net.Stream, server: *Serve ) catch return; }; + // An OP_MSG request with moreToCome set is fire-and-forget: the driver + // will not read a reply, so sending one leaves it unread in the socket + // and every later command on this connection reads the wrong one. The + // command still runs -- only the reply is suppressed. + // + // This is not an obscure corner. The Node driver sends `endSessions` + // with `writeConcern: {w: 0}` when a client closes, and every + // unacknowledged write uses the same mechanism. Ignoring the flag + // desynced the connection on first use, which surfaced as an + // unrelated-looking timeout on the *next* command: + // + // insert crud-v1.coll ... Command failed, durationMS: 8004, + // failure: 'connection 2 to 127.0.0.1:27222 timed out' + // + // The engine was answering in microseconds the whole time; the driver + // was waiting for a reply it had already been sent, out of order. + if (msg.op_code == wire.op_code_msg and msg.more_to_come()) continue; + out_buf.clearRetainingCapacity(); const built = if (msg.op_code == wire.op_code_query) wire.write_reply_query(server.gpa, reply_request_id, msg.request_id, reply.pairs.items, &out_buf) diff --git a/src/wire.zig b/src/wire.zig index 7b788c8..0173340 100644 --- a/src/wire.zig +++ b/src/wire.zig @@ -28,6 +28,18 @@ pub const Message = struct { body: bson.Document, seqs: []const Seq, + /// OP_MSG flagBits: bit 0 checksumPresent, bit 1 moreToCome, bit 16 + /// exhaustAllowed. + pub const flag_more_to_come: u32 = 1 << 1; + + /// Whether the client declared it will not read a reply to this request + /// (unacknowledged writes, `endSessions`). Answering one anyway leaves an + /// unread reply in the socket and desyncs every later command on the + /// connection -- see the handler in server.zig. + pub fn more_to_come(self: *const Message) bool { + return self.flags & flag_more_to_come != 0; + } + pub fn parse(gpa: std.mem.Allocator, bytes: []const u8) !Message { if (bytes.len < 16) return error.InvalidMessage; const total: u32 = std.mem.readInt(u32, bytes[0..4], .little); diff --git a/tests/e2e/e2e.js b/tests/e2e/e2e.js index d74fdb9..09cab52 100644 --- a/tests/e2e/e2e.js +++ b/tests/e2e/e2e.js @@ -110,6 +110,48 @@ async function main() { const stillAlive = await users.aggregate([{ $match: {} }, { $count: 'n' }]).toArray(); check('server survives a bare $sort pipeline', stillAlive[0]?.n === 4, JSON.stringify(stillAlive)); + // --- a database-level command must not leak the catalog lock --- + // db.aggregate() sends {aggregate: 1}, which names no collection. dispatch + // used to resolve the namespace after taking the catalog lock and bail with a + // plain return, leaking it shared forever. Reads kept working, so the damage + // only showed on the next write that had to create a collection -- which is + // the second half of this check, and would hang rather than fail. + let dbLevelErr = null; + try { + await db.aggregate([{ $listLocalSessions: {} }]).toArray(); + } catch (e) { + dbLevelErr = e; + } + check( + 'database-level aggregate gives a real error, not an empty reply', + dbLevelErr !== null && typeof dbLevelErr.message === 'string' && dbLevelErr.message !== 'n/a', + String(dbLevelErr && dbLevelErr.message).slice(0, 60), + ); + const afterDbLevel = await db.collection('lock_probe').insertOne({ _id: 1 }); + check('a write creating a collection still completes afterwards', afterDbLevel.insertedId === 1); + + // --- unacknowledged writes must not desync the connection --- + // An OP_MSG request with moreToCome set gets no reply. Sending one anyway + // left it unread in the socket, so the *next* command on that connection + // read the wrong reply. maxPoolSize 1 pins both operations to one socket, + // which is what makes the bug visible; with a larger pool the driver may + // hand out a different connection and hide it. The driver also does this to + // itself on close, via endSessions with {w: 0}. + const w0client = new MongoClient(URL, { maxPoolSize: 1 }); + try { + await w0client.connect(); + const w0 = w0client.db('e2e').collection('unack'); + await w0.deleteMany({}); + await w0.insertOne({ _id: 1, v: 'acknowledged' }); + const unack = await w0.insertOne({ _id: 2, v: 'unacknowledged' }, { writeConcern: { w: 0 } }); + check('unacknowledged insert is not acknowledged', unack.acknowledged === false, JSON.stringify(unack)); + // The assertion that matters: the same socket still works afterwards. + const after = await w0.countDocuments({}); + check('connection survives an unacknowledged write', after === 2, `count=${after}`); + } finally { + await w0client.close(); + } + // --- duplicate key --- let dupErr = null; try {