dev/MultiforaDB
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

commands: accept a well-formed lsid, refuse what would be a lie

Browse Source 
A driver puts `lsid` on every acknowledged command already, because
`add_server_info` advertises `logicalSessionTimeoutMinutes`. So this is not new
plumbing, it is a decision about input that has been arriving all along and
being ignored. Accepting it and doing nothing is honest: a session here would
own nothing -- no transactions to scope, no retryable writes, cursors that
outlive their connection for their own reasons. There is deliberately no
session registry; it would be a mutex on the dispatch path guarding state
nothing reads, and M4's transaction state machine is what should decide its
shape.

`txnNumber` is a different matter, and ignoring it would be the lie this
commit exists to remove. A transactional write would run non-transactionally,
answer `ok`, and become durable; the client would find out at
`commitTransaction`, by which time the data is on disk. It is refused, with
`startTransaction` and `autocommit` alongside it for the same reason.

Every code and every message was measured against mongod 8.3.7 through a raw
OP_MSG probe -- the driver overwrites `lsid` with its own session, so a
malformed one cannot be sent through it and none of this was checkable the
usual way. Three things the measurement settled that guessing would have got
wrong: an unknown command with a malformed lsid answers CommandNotFound, so
the lookup comes first and this check belongs exactly where it sits; the codes
for a bad session id are IDL parser codes (40414, 40415) rather than anything
resembling the rest of our table; and a bad UUID length is InvalidUUID 207
while a bad subtype is TypeMismatch 14, which no amount of reasoning would
have produced.

Three divergences from mongod, all one cause: it keeps a per-command table of
which commands accept `txnNumber` at all, and answers Location50889 or
OperationNotSupportedInTransaction 263 for those that do not, before reaching
the standalone refusal. We have no such table and give the standalone answer
uniformly. For every CRUD command -- everything a driver would actually send
these fields on -- the replies are identical; they differ only on things like
`ping`, where mongod is more specific rather than differently right.

Checked before any lock is taken, and the test for that is the second half of
each refusal: keep using the engine afterwards, with a write that has to take
the catalog exclusive to create a collection. Mutation-checked by moving the
call below `lock_catalog` -- the test run hangs, which is precisely how the
nameless-command lock leak presented, since a leaked *shared* lock is
invisible to every reader.

174/174 unit tests.
This commit is contained in:
A.Shakhmatov
2026-08-09 15:06:35 +03:00
parent 0034da4293
commit 54092f44be
1 changed files with 340 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files

340
src/commands.zig
View File

@@ -52,6 +52,16 @@ pub const ErrorCode = enum(i32) {
unauthorized = 13,
type_mismatch = 14,
operation_failed = 96,
// Session and transaction codes, measured against mongod 8.3.7 with a raw
// OP_MSG probe -- the driver rewrites `lsid` with its own session, so a
// malformed one cannot be sent through it and none of this could have been
// checked the usual way. The two five-digit ones are IDL parser codes: they
// are what mongod's generated command parsers answer, not hand-written
// checks, which is why they look unlike the rest of the table.
illegal_operation = 20,
invalid_uuid = 207,
idl_failed_to_parse = 40414,
idl_unknown_field = 40415,
};
/// Which lock (if any) a command needs on the engine. Contract: only
@@ -151,6 +161,12 @@ pub fn dispatch(ctx: *Context, msg: *wire.Message, reply: *wire.Reply) !void {
return reply.put_error(@intFromEnum(ErrorCode.command_not_found), "CommandNotFound", errmsg);
};
// Session and transaction fields are checked here: after the command is
// known -- mongod answers CommandNotFound to an unknown command carrying a
// malformed lsid, measured, not assumed -- and before any lock is taken,
// for the reason the comment below records at length.
if (try reject_bad_session_fields(msg, reply, name)) return;
// Lock the catalog (shared for most commands, exclusive for DDL), then
// the target collection, then run the handler. The collection lock is
// taken while the catalog lock is held, so a concurrent drop can never
@@ -381,6 +397,168 @@ fn cmd_server_status(ctx: *Context, _: *wire.Message, reply: *wire.Reply) !void
try reply.put_ok();
}
// ---------------------------------------------------------------------------
// Sessions
//
// This server keeps no session registry, and that is a decision rather than an
// omission: a session here would own nothing. There are no transactions to
// scope, no cursors that outlive their connection differently because of one,
// and no retryable writes -- a driver disables those for a standalone. A
// registry would be a mutex on the dispatch path guarding state nothing reads.
// M4 gets to design it, when the transaction state machine says what it needs.
//
// What is *not* optional is telling the truth about the fields a driver sends
// anyway. `lsid` rides on every acknowledged command already, because
// `add_server_info` advertises `logicalSessionTimeoutMinutes`. Accepting it and
// doing nothing is honest -- there is nothing to do. Accepting `txnNumber` and
// doing nothing is not: it would run a transactional write non-transactionally
// and answer ok, and the client would only find out at `commitTransaction`,
// long after the data was on disk.
// ---------------------------------------------------------------------------
/// mongod's name for a binary subtype, for the one message that quotes it.
fn subtype_name(subtype: u8) []const u8 {
return switch (subtype) {
0x00 => "general",
0x01 => "function",
0x02 => "binary",
0x03 => "uuid_old",
0x04 => "UUID",
0x05 => "MD5",
0x06 => "encrypt",
else => "unknown",
};
}
/// The first `lsid` field this server does not know, for the message that has
/// to name it. Walked again on the error path only, so that `Message.lsid`
/// stays a yes-or-no answer.
fn unknown_lsid_field(msg: *wire.Message) []const u8 {
const doc = switch (msg.body.get("lsid") orelse return "?") {
.doc => |d| d,
else => return "?",
};
for (doc) |pair| {
const known = std.mem.eql(u8, pair.key, "id") or std.mem.eql(u8, pair.key, "uid") or
std.mem.eql(u8, pair.key, "txnNumber") or std.mem.eql(u8, pair.key, "txnUUID");
if (!known) return pair.key;
}
return "?";
}
/// Writes the reply for a malformed `lsid` and answers whether it did.
fn reject_bad_lsid(msg: *wire.Message, reply: *wire.Reply, cmd: []const u8) !bool {
_ = msg.lsid() catch |err| {
const arena = reply.arena_alloc();
const text = switch (err) {
error.LsidNotDocument => try std.fmt.allocPrint(
arena,
"BSON field '{s}.lsid' is the wrong type '{s}', expected type 'object'",
.{ cmd, msg.body.get("lsid").?.type_name() },
),
error.LsidIdMissing => try std.fmt.allocPrint(
arena,
"BSON field '{s}.lsid.id' is missing but a required field",
.{cmd},
),
error.LsidIdNotBinary => try std.fmt.allocPrint(
arena,
"BSON field '{s}.lsid.id' is the wrong type '{s}', expected type 'binData'",
.{ cmd, bson.get_pair(msg.body.get("lsid").?.doc, "id").?.type_name() },
),
error.LsidIdNotUuid => try std.fmt.allocPrint(
arena,
"BSON field '{s}.lsid.id' is the wrong binData type '{s}', expected type 'UUID'",
.{ cmd, subtype_name(bson.get_pair(msg.body.get("lsid").?.doc, "id").?.binary.subtype) },
),
error.LsidUnknownField => try std.fmt.allocPrint(
arena,
"BSON field '{s}.lsid.{s}' is an unknown field.",
.{ cmd, unknown_lsid_field(msg) },
),
else => "",
};
switch (err) {
error.LsidNotDocument, error.LsidIdNotBinary, error.LsidIdNotUuid => try reply.put_error(
@intFromEnum(ErrorCode.type_mismatch),
"TypeMismatch",
text,
),
error.LsidIdMissing => try reply.put_error(
@intFromEnum(ErrorCode.idl_failed_to_parse),
"IDLFailedToParse",
text,
),
error.LsidUnknownField => try reply.put_error(
@intFromEnum(ErrorCode.idl_unknown_field),
"IDLUnknownField",
text,
),
error.LsidIdWrongLength => try reply.put_error(
@intFromEnum(ErrorCode.invalid_uuid),
"InvalidUUID",
"uuid must be a 16-byte binary field with UUID (4) subtype",
),
error.LsidTxnNumberWithoutTxnUuid => try invalid_arg(
reply,
"Cannot specify txnNumber in lsid without specifying txnUUID",
),
error.LsidInternalSession => try invalid_arg(
reply,
"Internal sessions are not supported outside of transactions",
),
}
return true;
};
return false;
}
/// Writes the reply for a transaction field this server cannot honour, and
/// answers whether it did.
///
/// Refusing rather than ignoring is the whole point. A driver that is told
/// `ok` for a write carrying a `txnNumber` has been told the write is part of
/// a transaction; it is not, it is already durable, and the first the client
/// hears of it is a failing `commitTransaction`. The order of the checks and
/// every message below are mongod 8.3.7's, measured.
fn reject_txn_fields(msg: *wire.Message, reply: *wire.Reply, cmd: []const u8) !bool {
const txn_number = msg.body.get("txnNumber");
if (msg.body.get("startTransaction") != null and msg.body.get("autocommit") == null) {
try invalid_arg(reply, "'startTransaction' field requires 'autocommit' field to also be specified");
return true;
}
if (msg.body.get("autocommit") != null and txn_number == null) {
try invalid_arg(reply, "'autocommit' field requires a transaction number to also be specified");
return true;
}
const n = txn_number orelse return false;
if (n != .int64 and n != .int32) {
const text = try std.fmt.allocPrint(
reply.arena_alloc(),
"BSON field '{s}.txnNumber' is the wrong type '{s}', expected type 'long'",
.{ cmd, n.type_name() },
);
try reply.put_error(@intFromEnum(ErrorCode.type_mismatch), "TypeMismatch", text);
return true;
}
if (msg.body.get("lsid") == null) {
try invalid_arg(reply, "Transaction number requires a session ID to also be specified");
return true;
}
try reply.put_error(
@intFromEnum(ErrorCode.illegal_operation),
"IllegalOperation",
"Transaction numbers are only allowed on a replica set member or mongos",
);
return true;
}
/// Answers whether an error reply was written, in which case dispatch is done.
fn reject_bad_session_fields(msg: *wire.Message, reply: *wire.Reply, cmd: []const u8) !bool {
if (try reject_bad_lsid(msg, reply, cmd)) return true;
return reject_txn_fields(msg, reply, cmd);
}
fn cmd_end_sessions(_: *Context, _: *wire.Message, reply: *wire.Reply) !void {
try reply.put_ok();
}
@@ -2869,6 +3047,168 @@ fn parse_fake_msg(name: []const u8, value: bson.Value, extra: []const bson.Pair)
return wire.Message.parse(testing.allocator, msg.items);
}
/// Runs one command and hands back its reply's `code`, or null on ok:1.
fn run_for_code(ctx: *Context, name: []const u8, value: bson.Value, extra: []const bson.Pair) !?i32 {
var reply = wire.Reply.init(testing.allocator);
defer reply.deinit();
var msg = try parse_fake_msg(name, value, extra);
defer msg.deinit();
try dispatch(ctx, &msg, &reply);
if (bson.get_pair(reply.pairs.items, "code")) |c| return c.int32;
try testing.expectEqual(@as(f64, 1.0), bson.get_pair(reply.pairs.items, "ok").?.double);
return null;
}
fn doc_count(ctx: *Context, coll: []const u8) !i64 {
var reply = wire.Reply.init(testing.allocator);
defer reply.deinit();
var msg = try parse_fake_msg("count", .{ .string = coll }, &.{});
defer msg.deinit();
try dispatch(ctx, &msg, &reply);
return bson.get_pair(reply.pairs.items, "n").?.int64;
}
test "a well-formed session id changes nothing and is not echoed" {
var threaded: std.Io.Threaded = .init_single_threaded;
defer threaded.deinit();
const io = threaded.io();
var tdb = try TestDb.init(io);
defer tdb.deinit();
var ctx = tdb.ctx(io);
const uuid = [_]u8{0x5A} ** 16;
const lsid = bson.Value{ .doc = &.{
.{ .key = "id", .value = .{ .binary = .{ .subtype = 4, .data = &uuid } } },
} };
var reply = wire.Reply.init(testing.allocator);
defer reply.deinit();
var msg = try parse_fake_msg("insert", .{ .string = "sess" }, &.{
.{ .key = "documents", .value = .{ .array = &.{.{ .doc = &.{.{ .key = "a", .value = .{ .int32 = 1 } }} }} } },
.{ .key = "lsid", .value = lsid },
});
defer msg.deinit();
try dispatch(&ctx, &msg, &reply);
try testing.expectEqual(@as(f64, 1.0), bson.get_pair(reply.pairs.items, "ok").?.double);
// mongod answers a well-formed lsid with exactly `{ok: 1}` and no echo,
// measured; a driver reads only `$clusterTime` and `operationTime` back.
try testing.expect(bson.get_pair(reply.pairs.items, "lsid") == null);
try testing.expectEqual(@as(i64, 1), try doc_count(&ctx, "sess"));
}
test "a malformed session id is refused without leaking a lock" {
// The second half is the point. `reject_bad_session_fields` runs before any
// lock is taken, and the way to show it is to keep using the engine after
// each refusal: a leaked *shared* catalog lock is invisible to readers and
// only stops the next write that needs it exclusively. That is exactly how
// the nameless-command bug hid for a milestone.
var threaded: std.Io.Threaded = .init_single_threaded;
defer threaded.deinit();
const io = threaded.io();
var tdb = try TestDb.init(io);
defer tdb.deinit();
var ctx = tdb.ctx(io);
const uuid = [_]u8{0x7C} ** 16;
const good = bson.Value{ .binary = .{ .subtype = 4, .data = &uuid } };
// Every code here was measured against mongod 8.3.7, not recalled.
const cases = [_]struct { code: i32, lsid: bson.Value }{
.{ .code = 14, .lsid = .{ .int32 = 5 } },
.{ .code = 40414, .lsid = .{ .doc = &.{} } },
.{ .code = 14, .lsid = .{ .doc = &.{.{ .key = "id", .value = .{ .string = "nope" } }} } },
.{ .code = 14, .lsid = .{ .doc = &.{
.{ .key = "id", .value = .{ .binary = .{ .subtype = 0, .data = &uuid } } },
} } },
.{ .code = 207, .lsid = .{ .doc = &.{
.{ .key = "id", .value = .{ .binary = .{ .subtype = 4, .data = uuid[0..15] } } },
} } },
.{ .code = 40415, .lsid = .{ .doc = &.{
.{ .key = "id", .value = good },
.{ .key = "bogus", .value = .{ .int32 = 1 } },
} } },
.{ .code = 72, .lsid = .{ .doc = &.{
.{ .key = "id", .value = good },
.{ .key = "txnUUID", .value = good },
} } },
};
for (cases, 0..) |c, i| {
const insert_doc = bson.Value{ .array = &.{.{ .doc = &.{.{ .key = "a", .value = .{ .int32 = 1 } }} }} };
const code = try run_for_code(&ctx, "insert", .{ .string = "leaky" }, &.{
.{ .key = "documents", .value = insert_doc },
.{ .key = "lsid", .value = c.lsid },
});
try testing.expectEqual(c.code, code.?);
// A write that needs the catalog exclusive to create a collection: the
// one operation a leaked shared lock would block.
var name_buf: [16]u8 = undefined;
const fresh = try std.fmt.bufPrint(&name_buf, "after{d}", .{i});
try testing.expectEqual(@as(?i32, null), try run_for_code(&ctx, "insert", .{ .string = fresh }, &.{
.{ .key = "documents", .value = insert_doc },
}));
}
}
test "a transactional write is refused rather than applied" {
// The count is the assertion. Ignoring `txnNumber` would answer ok, write
// the document, and leave the client to discover at commitTransaction that
// its transaction was never one -- long after the data was durable.
var threaded: std.Io.Threaded = .init_single_threaded;
defer threaded.deinit();
const io = threaded.io();
var tdb = try TestDb.init(io);
defer tdb.deinit();
var ctx = tdb.ctx(io);
const uuid = [_]u8{0x3E} ** 16;
const lsid = bson.Value{ .doc = &.{
.{ .key = "id", .value = .{ .binary = .{ .subtype = 4, .data = &uuid } } },
} };
const docs = bson.Value{ .array = &.{.{ .doc = &.{.{ .key = "a", .value = .{ .int32 = 1 } }} }} };
try testing.expectEqual(@as(i32, 20), (try run_for_code(&ctx, "insert", .{ .string = "txn" }, &.{
.{ .key = "documents", .value = docs },
.{ .key = "lsid", .value = lsid },
.{ .key = "txnNumber", .value = .{ .int64 = 1 } },
.{ .key = "startTransaction", .value = .{ .bool = true } },
.{ .key = "autocommit", .value = .{ .bool = false } },
})).?);
try testing.expectEqual(@as(i64, 0), try doc_count(&ctx, "txn"));
// The order the checks fire in is mongod's, measured: a missing session id
// is reported before the standalone refusal, and a bad type before both.
try testing.expectEqual(@as(i32, 72), (try run_for_code(&ctx, "insert", .{ .string = "txn" }, &.{
.{ .key = "documents", .value = docs },
.{ .key = "txnNumber", .value = .{ .int64 = 1 } },
})).?);
try testing.expectEqual(@as(i32, 14), (try run_for_code(&ctx, "insert", .{ .string = "txn" }, &.{
.{ .key = "documents", .value = docs },
.{ .key = "lsid", .value = lsid },
.{ .key = "txnNumber", .value = .{ .string = "nope" } },
})).?);
try testing.expectEqual(@as(i32, 72), (try run_for_code(&ctx, "insert", .{ .string = "txn" }, &.{
.{ .key = "documents", .value = docs },
.{ .key = "lsid", .value = lsid },
.{ .key = "startTransaction", .value = .{ .bool = true } },
})).?);
try testing.expectEqual(@as(i64, 0), try doc_count(&ctx, "txn"));
}
test "an unknown command is reported before its session id is judged" {
// Measured: mongod answers CommandNotFound to `{nosuchcmd: 1, lsid: 5}`,
// so the lookup comes first and this is where the check belongs.
var threaded: std.Io.Threaded = .init_single_threaded;
defer threaded.deinit();
const io = threaded.io();
var tdb = try TestDb.init(io);
defer tdb.deinit();
var ctx = tdb.ctx(io);
try testing.expectEqual(@as(i32, 59), (try run_for_code(&ctx, "nosuchcmd", .{ .int32 = 1 }, &.{
.{ .key = "lsid", .value = .{ .int32 = 5 } },
})).?);
}
test "concurrent insert/find commands on a threaded Io" {
// Exercises dispatch's lock classification end-to-end: writer fibers run
// `insert` under the exclusive lock, reader fibers run `count`/`find`