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.
This commit is contained in:
@@ -170,6 +170,24 @@ fn handle_connection_inner(io: std.Io, stream: std.Io.net.Stream, server: *Serve
|
|||||||
) catch return;
|
) 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();
|
out_buf.clearRetainingCapacity();
|
||||||
const built = if (msg.op_code == wire.op_code_query)
|
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)
|
wire.write_reply_query(server.gpa, reply_request_id, msg.request_id, reply.pairs.items, &out_buf)
|
||||||
|
|||||||
12
src/wire.zig
12
src/wire.zig
@@ -28,6 +28,18 @@ pub const Message = struct {
|
|||||||
body: bson.Document,
|
body: bson.Document,
|
||||||
seqs: []const Seq,
|
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 {
|
pub fn parse(gpa: std.mem.Allocator, bytes: []const u8) !Message {
|
||||||
if (bytes.len < 16) return error.InvalidMessage;
|
if (bytes.len < 16) return error.InvalidMessage;
|
||||||
const total: u32 = std.mem.readInt(u32, bytes[0..4], .little);
|
const total: u32 = std.mem.readInt(u32, bytes[0..4], .little);
|
||||||
|
|||||||
@@ -110,6 +110,48 @@ async function main() {
|
|||||||
const stillAlive = await users.aggregate([{ $match: {} }, { $count: 'n' }]).toArray();
|
const stillAlive = await users.aggregate([{ $match: {} }, { $count: 'n' }]).toArray();
|
||||||
check('server survives a bare $sort pipeline', stillAlive[0]?.n === 4, JSON.stringify(stillAlive));
|
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 ---
|
// --- duplicate key ---
|
||||||
let dupErr = null;
|
let dupErr = null;
|
||||||
try {
|
try {
|
||||||
|
|||||||
Reference in New Issue
Block a user