From df422297856ae4555225a17a3061828501c8d45e Mon Sep 17 00:00:00 2001 From: Nqvinh7603 Date: Sat, 7 Nov 2026 16:41:50 +0700 Subject: [PATCH] Support OP_MSG moreToCome for unacknowledged writes Drivers send unacknowledged (w=0) writes as OP_MSG with the moreToCome flag bit set and do not read a reply. The decoder rejected any flag other than checksumPresent, so these writes were dropped and the connection was closed, with no error visible to the client. Accept the flag, execute the command as usual, and send no reply, since a reply the client does not read would desynchronize the connection. --- .../mongo/wire/MongoDatabaseHandler.java | 5 +++ .../mongo/wire/MongoWireProtocolHandler.java | 7 +++- .../mongo/wire/message/MongoMessage.java | 13 +++++++ .../mongo/wire/MongoDatabaseHandlerTest.java | 34 +++++++++++++++++++ .../de/bwaldvogel/mongo/MongoServerTest.java | 22 ++++++++++++ 5 files changed, 80 insertions(+), 1 deletion(-) diff --git a/core/src/main/java/de/bwaldvogel/mongo/wire/MongoDatabaseHandler.java b/core/src/main/java/de/bwaldvogel/mongo/wire/MongoDatabaseHandler.java index 41e59d922..eeedde1c7 100644 --- a/core/src/main/java/de/bwaldvogel/mongo/wire/MongoDatabaseHandler.java +++ b/core/src/main/java/de/bwaldvogel/mongo/wire/MongoDatabaseHandler.java @@ -60,6 +60,11 @@ protected void channelRead0(ChannelHandlerContext ctx, ClientRequest object) { ctx.channel().writeAndFlush(mongoReply); } else if (object instanceof MongoMessage mongoMessage) { MongoMessage response = handleMessage(mongoMessage); + if (mongoMessage.isMoreToCome()) { + // the client does not read a reply, so sending one would desynchronize the connection + log.debug("not replying to {} with moreToCome set", mongoMessage); + return; + } ctx.channel().writeAndFlush(response); } else { throw new MongoServerException("unknown message: " + object); diff --git a/core/src/main/java/de/bwaldvogel/mongo/wire/MongoWireProtocolHandler.java b/core/src/main/java/de/bwaldvogel/mongo/wire/MongoWireProtocolHandler.java index a8c3b69fd..a6519cbe1 100644 --- a/core/src/main/java/de/bwaldvogel/mongo/wire/MongoWireProtocolHandler.java +++ b/core/src/main/java/de/bwaldvogel/mongo/wire/MongoWireProtocolHandler.java @@ -142,6 +142,11 @@ private ClientRequest handleMessage(Channel channel, MessageHeader header, ByteB flags.add(MessageFlag.CHECKSUM_PRESENT); } + if (MessageFlag.MORE_TO_COME.isSet(flagBits)) { + flagBits = MessageFlag.MORE_TO_COME.removeFrom(flagBits); + flags.add(MessageFlag.MORE_TO_COME); + } + if (flagBits != 0) { throw new UnsupportedOperationException("flags=" + flagBits + " not yet supported"); } @@ -179,7 +184,7 @@ private ClientRequest handleMessage(Channel channel, MessageHeader header, ByteB Assert.isNull(old); } - return new MongoMessage(channel, header, body); + return new MongoMessage(channel, header, body, flags.contains(MessageFlag.MORE_TO_COME)); } private void decodeKindDocumentSequence(ByteBuf buffer, Document documentSequence) { diff --git a/core/src/main/java/de/bwaldvogel/mongo/wire/message/MongoMessage.java b/core/src/main/java/de/bwaldvogel/mongo/wire/message/MongoMessage.java index cf7051a5a..f8213ea91 100644 --- a/core/src/main/java/de/bwaldvogel/mongo/wire/message/MongoMessage.java +++ b/core/src/main/java/de/bwaldvogel/mongo/wire/message/MongoMessage.java @@ -10,10 +10,16 @@ public class MongoMessage extends ClientRequest { private final Document document; private final int flags = 0; + private final boolean moreToCome; public MongoMessage(Channel channel, MessageHeader header, Document document) { + this(channel, header, document, false); + } + + public MongoMessage(Channel channel, MessageHeader header, Document document, boolean moreToCome) { super(channel, header, null); this.document = document; + this.moreToCome = moreToCome; } public Document getDocument() { @@ -24,6 +30,13 @@ public int getFlags() { return flags; } + /** + * Whether the client set the moreToCome flag: it does not wait for a reply, as for an unacknowledged (w=0) write. + */ + public boolean isMoreToCome() { + return moreToCome; + } + @Override public String getDatabaseName() { return (String) document.get("$db"); diff --git a/core/src/test/java/de/bwaldvogel/mongo/wire/MongoDatabaseHandlerTest.java b/core/src/test/java/de/bwaldvogel/mongo/wire/MongoDatabaseHandlerTest.java index d4a06ab54..f7c65b131 100644 --- a/core/src/test/java/de/bwaldvogel/mongo/wire/MongoDatabaseHandlerTest.java +++ b/core/src/test/java/de/bwaldvogel/mongo/wire/MongoDatabaseHandlerTest.java @@ -4,6 +4,7 @@ import static org.assertj.core.api.Assertions.assertThat; import static org.mockito.ArgumentMatchers.any; import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.never; import static org.mockito.Mockito.verify; import static org.mockito.Mockito.when; @@ -15,6 +16,7 @@ import de.bwaldvogel.mongo.wire.message.MongoMessage; import de.bwaldvogel.mongo.wire.message.MongoQuery; import io.netty.channel.Channel; +import io.netty.channel.ChannelHandlerContext; class MongoDatabaseHandlerTest { @@ -71,4 +73,36 @@ void testHandleMessageUnknownError() throws Exception { assertThat(responseMessageDoc.get("errmsg")).isEqualTo("Unknown error: unexpected"); assertThat(responseMessageDoc.get("ok")).isEqualTo(0); } + + @Test + void testMessageWithMoreToComeIsHandledWithoutReply() throws Exception { + MongoBackend backend = mock(MongoBackend.class); + Channel channel = mock(Channel.class); + ChannelHandlerContext ctx = mock(ChannelHandlerContext.class); + when(ctx.channel()).thenReturn(channel); + + MongoDatabaseHandler handler = new MongoDatabaseHandler(backend, null); + MongoMessage requestMessage = new MongoMessage(channel, new MessageHeader(0, 0), json("insert: 'coll'"), true); + + handler.channelRead0(ctx, requestMessage); + + verify(backend).handleMessage(requestMessage); + verify(channel, never()).writeAndFlush(any()); + } + + @Test + void testMessageWithoutMoreToComeIsReplied() throws Exception { + MongoBackend backend = mock(MongoBackend.class); + Channel channel = mock(Channel.class); + ChannelHandlerContext ctx = mock(ChannelHandlerContext.class); + when(ctx.channel()).thenReturn(channel); + + MongoDatabaseHandler handler = new MongoDatabaseHandler(backend, null); + MongoMessage requestMessage = new MongoMessage(channel, new MessageHeader(0, 0), json("insert: 'coll'")); + + handler.channelRead0(ctx, requestMessage); + + verify(channel).writeAndFlush(any(MongoMessage.class)); + } + } diff --git a/test-common/src/main/java/de/bwaldvogel/mongo/MongoServerTest.java b/test-common/src/main/java/de/bwaldvogel/mongo/MongoServerTest.java index 531feaf48..3f1242153 100644 --- a/test-common/src/main/java/de/bwaldvogel/mongo/MongoServerTest.java +++ b/test-common/src/main/java/de/bwaldvogel/mongo/MongoServerTest.java @@ -24,8 +24,10 @@ import com.mongodb.ConnectionString; import com.mongodb.MongoClientSettings; import com.mongodb.MongoException; +import com.mongodb.WriteConcern; import com.mongodb.client.MongoClient; import com.mongodb.client.MongoClients; +import com.mongodb.client.MongoCollection; import de.bwaldvogel.mongo.backend.CollectionUtils; @@ -74,6 +76,26 @@ void testBindAndGetConnectionStringThenConnect() throws Exception { } } + @Test + @Timeout(TEST_TIMEOUT_SECONDS) + void testUnacknowledgedWrites() throws Exception { + // a single connection, so the count is read after the unacknowledged writes were processed + MongoClientSettings clientSettings = MongoClientSettings.builder() + .applyConnectionString(new ConnectionString(server.bindAndGetConnectionString())) + .applyToConnectionPoolSettings(pool -> pool.maxSize(1)) + .build(); + try (MongoClient client = MongoClients.create(clientSettings)) { + MongoCollection collection = client.getDatabase("testdb").getCollection("testcoll"); + MongoCollection unacknowledged = collection.withWriteConcern(WriteConcern.UNACKNOWLEDGED); + + for (int i = 0; i < 3; i++) { + assertThat(unacknowledged.insertOne(new Document("_id", i)).wasAcknowledged()).isFalse(); + } + + assertThat(collection.countDocuments()).isEqualTo(3); + } + } + @Test @Timeout(TEST_TIMEOUT_SECONDS) void testStopListening() throws Exception {