Updates to support MongoDB 5.1+

In MongoDB 3.6, the `OP_MSG` and `OP_COMPRESSED` opcodes were added to
the wired message protocol. `OP_MSG` was added to standardize the
MongoDB message format, and `OP_COMPRESSED` takes things a step
further by compressing the message to increase network efficiency. As
of MongoDB 5.1, the old opcodes were removed, and all messages are
sent with either `OP_MSG` or `OP_COMPRESSED`.

Because this driver did not have the ability to send messages with the
`OP_MSG` opcode, it couldn't be used with any versions of MongoDB
greater than 5.1.

To bring this driver into the modern era, this PR ports over the
changes made in the [emqx fork](https://github.com/emqx/mongodb-erlang/)
of this driver to support the `OP_MSG` opcode.

Additionally, there are some slightly unrelated, but relevant changes
in this PR as well such as fixing some dialyzer errors and updating the
GitHub actions workflow definitions to get the tests up and running again.

Author: samwar

.github/workflows/dialyzer.yml

@@ -7,14 +7,14 @@ on:
 
 jobs:
   dialyzer:
-    runs-on: ubuntu-20.04
+    runs-on: ubuntu-24.04
     container:
       image: erlang:24-slim
     steps:
-      - uses: actions/checkout@v2.0.0
+      - uses: actions/checkout@v4
       - name: Cache PLTs
         id: cache-plts
-        uses: actions/cache@v2
+        uses: actions/cache@v4
         with:
           path: ~/.cache/rebar3/
           key: ${{ runner.os }}-erlang-${{ hashFiles(format('{0}{1}', github.workspace, '/rebar.lock')) }}

.github/workflows/test.yml

@@ -7,37 +7,37 @@ on:
 
 jobs:
   test:
-    runs-on: ubuntu-20.04
+    runs-on: ubuntu-24.04
     strategy:
       matrix:
-        erlang: [22, 23]
+        erlang: [24, 25]
         mongodb: ["4.4.8", "5.0.2"]
     container:
       image: erlang:${{ matrix.erlang }}
     steps:
-      - uses: actions/checkout@v2.0.0
+      - uses: actions/checkout@v4
       - run: ./scripts/install_mongo_debian.sh ${{ matrix.mongodb }}
       - run: ./scripts/start_mongo_single_node.sh
       - run: ./scripts/start_mongo_cluster.sh
       - run: ./rebar3 eunit
       - run: ./rebar3 ct
       - name: Archive Replica Set Logs
-        uses: actions/upload-artifact@v2
+        uses: actions/upload-artifact@v4
         if: failure()
         with:
-          name: mongodb_replica_set_logs
+          name: erlang-${{ matrix.erlang }}-mongodb-${{ matrix.mongodb }}-mongodb_replica_set_logs
           path: rs0-logs
           retention-days: 1
       - name: Archive Single Node Log
-        uses: actions/upload-artifact@v2
+        uses: actions/upload-artifact@v4
         if: failure()
         with:
-          name: single_node.log
+          name: erlang-${{ matrix.erlang }}-mongodb-${{ matrix.mongodb }}-single_node.log
           path: single_node.log
           retention-days: 1
       - name: CT Logs
-        uses: actions/upload-artifact@v2
+        uses: actions/upload-artifact@v4
         with:
-          name: ct_logs
+          name: erlang-${{ matrix.erlang }}-mongodb-${{ matrix.mongodb }}-ct_logs
           path: _build/test/logs/
           retention-days: 5

.github/workflows/test_coverage.yml

@@ -7,39 +7,39 @@ on:
 
 jobs:
   test_coverage:
-    runs-on: ubuntu-20.04
+    runs-on: ubuntu-24.04
     container:
-      image: erlang:23
+      image: erlang:24
     steps:
-      - uses: actions/checkout@v2.0.0
+      - uses: actions/checkout@v4
       - run: ./scripts/install_mongo_debian.sh 5.0.2
       - run: ./scripts/start_mongo_single_node.sh
       - run: ./scripts/start_mongo_cluster.sh
       - run: ./rebar3 eunit --cover --cover_export_name eunit.coverdata
       - run: ./rebar3 ct --cover --cover_export_name ct.coverdata
       - run: rebar3 cover --verbose
       - name: Archive Replica Set Logs
-        uses: actions/upload-artifact@v2
+        uses: actions/upload-artifact@v4
         if: failure()
         with:
           name: mongodb_replica_set_logs
           path: rs0-logs
           retention-days: 1
       - name: Archive Single Node Log
-        uses: actions/upload-artifact@v2
+        uses: actions/upload-artifact@v4
         if: failure()
         with:
           name: single_node.log
           path: single_node.log
           retention-days: 1
       - name: Coverage Report
-        uses: actions/upload-artifact@v2
+        uses: actions/upload-artifact@v4
         with:
           name: Coverage Report
           path: _build/test/cover/
           retention-days: 5
       - name: CT Logs
-        uses: actions/upload-artifact@v2
+        uses: actions/upload-artifact@v4
         with:
           name: ct_logs
           path: _build/test/logs/

.gitignore

@@ -6,4 +6,7 @@ variables-ct*
 *.iml
 data
 _build
-.idea
+.idea
+*.log
+rebar.lock
+*.crashdump

README.md

@@ -42,9 +42,30 @@ If you are choosing between using
 [mongos](https://docs.mongodb.com/manual/reference/program/mongos/) and
 using mongo shard with `mongo_api` - prefer mongos and use `mc_worker_api`.
 
+In MongoDB 3.6, the `OP_MSG` and `OP_COMPRESSED` opcodes were added to the wired
+message protocol. `OP_MSG` was added to standardize the MongoDB message format,
+and `OP_COMPRESSED` takes things a step further by compressing the message to increase
+network efficiency. As of MongoDB 5.1, the old opcodes were deprecated, and all
+messages are sent with either `OP_MSG` or `OP_COMPRESSED`. 
+
+By default, this driver tries to automatically detect which MongoDB messaging protocol
+to use - the legacy one that existed before the `OP_MSG` opcode was introduced, or the
+modern messaging protocol based on `OP_MSG`. This is accomplished by setting the default
+value for the `use_legacy_protocol` env variable to `auto`. One can force the driver to
+use the legacy opcodes or the `OP_MSG`opcode by setting the application env
+`use_legacy_protocol` to `true` or `false` (for example by calling
+`application:set_env(mongodb, use_legacy_protocol, false)`).
+
+It's also possible to define the usage of legacy protocol on a per-connection basis.
+Simple pass `{use_legacy_protocol, false | true}` to mc_worker start options.
+
+As the `OP_MSG` opcode has existed in MongoDB since 3.6, and the driver defaults to auto-
+detecting which set of opcodes to use, ALL of the messages will be sent using the `OP_MSG`
+opcode if you are using a version for MongoDB greater 3.6 unless you explicitly set the
+`use_legacy_protocol` variable to `true`.
+
 mc_worker_api -- direct connection client
 ---------------------------------
-
 ### Connecting
 To connect to a database `test` on mongodb server listening on
 `localhost:27017` (or any address & port of your choosing)

include/mongo_protocol.hrl

@@ -8,6 +8,7 @@
 -type colldb() :: collection() | {database(), collection()}.
 -type collection() :: binary() | atom(). % without db prefix
 -type database() :: binary() | atom().
+-type command() :: insert | update | delete.
 
 
 %% write
@@ -45,6 +46,24 @@
   projector = #{} :: mc_worker_api:projector()
 }).
 
+-record(op_msg_write_op, {
+  command :: command(),
+  collection :: colldb(),
+  database :: undefined | mc_worker_api:database(),
+  extra_fields = [] :: bson:document() | nonempty_list({binary(),any()}),
+  documents_name = <<"documents">> :: bson:utf8(),
+  documents = [] :: any()
+}).
+
+-record(op_msg_response, {
+  response_doc :: map()
+}).
+
+-record(op_msg_command, {
+  database :: undefined | mc_worker_api:database(),
+  command_doc :: bson:document() | nonempty_list({binary(),any()})
+}).
+
 -record(getmore, {
   collection :: colldb(),
   batchsize = 0 :: mc_worker_api:batchsize(),
@@ -73,9 +92,9 @@
 -record(reply, {
   cursornotfound :: boolean(),
   queryerror :: boolean(),
-  awaitcapable = false :: boolean(),
+  awaitcapable = false :: boolean() | undefined,
   cursorid :: mc_worker_api:cursorid(),
-  startingfrom = 0 :: integer(),
+  startingfrom = 0 :: integer() | undefined,
   documents :: [map()]
 }).
 -endif.

include/mongo_types.hrl

@@ -31,11 +31,24 @@
 | {ssl, boolean()}
 | {ssl_opts, proplists:proplist()}
 | {register, atom() | fun()}.
+-type socket() :: gen_tcp:socket() | ssl:sslsocket().
 -type write_mode() :: unsafe | safe | {safe, bson:document()}.
 -type read_mode() :: master | slave_ok.
 -type service() :: {Host :: inet:hostname() | inet:ip_address(), Post :: 0..65535}.
 -type options() :: [option()].
 -type option() :: {timeout, timeout()} | {ssl, boolean()} | ssl | {database, database()} | {read_mode, read_mode()} | {write_mode, write_mode()}.
 -type cursor() :: pid().
--type query() :: #query{}.
--endif.
+-type query() :: #'query'{}.
+-type op_msg_command() :: #op_msg_command{}.
+-type op_msg_write_op() :: #op_msg_write_op{}.
+-type op_msg_response() :: #op_msg_response{}.
+-type request() :: query()
+| op_msg_command()
+| op_msg_write_op()
+| #killcursor{}
+| #insert{}
+| #update{}
+| #delete{}
+| #getmore{}
+| #ensure_index{}.
+-endif.

include/mongoc.hrl

@@ -76,8 +76,8 @@
 | {password, binary()}
 | {w_mode, mc_worker_api:write_mode()}.
 -type readprefs() :: [readpref()].
--type readpref() :: #{rp_mode => readmode()}
+-type readpref() :: #{rp_mode => readmode()} | #{mode => binary()} | #{mode => binary(), tags => tuple()}
 |{rp_tags, [tuple()]}.
 -type reason() :: atom().
 
--endif.
+-endif.