SERVER-35393 Flush asio output and limit SecureTransport out buffer

(cherry picked from commit 573ebb251784aa7665b978ca386b60799d74c878)

3 files changed, 47 insertions, 18 deletions

diff --git a/jstests/ssl/ssl_fragment.js b/jstests/ssl/ssl_fragment.js
index be4073c96ff..6d08da8e834 100644
--- a/jstests/ssl/ssl_fragment.js
+++ b/jstests/ssl/ssl_fragment.js
@@ -4,25 +4,44 @@
 (function() {
     'use strict';
 
-    var conn = MongoRunner.runMongod({
-        sslMode: "requireSSL",
-        sslPEMKeyFile: "jstests/libs/server.pem",
-    });
+    function runTest(conn) {
+        // SSL packets have a max size of ~16 kb so to test packet fragmentation support, create a
+        // string larger then 16kb.
+        const chunk = 'E$%G^56w4v5v54Vv$V@#t2#%t56u7B$ub%6 NU@ Y3qv4Yq%yq4C%yx$%zh';  // random data
+        let s = '';
+        while (s.length < (8 * 1024 * 1024)) {
+            s += chunk;
+        }
 
-    var large = "aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa";
-    var s = large;
+        const ssl_frag = conn.getCollection('test.ssl_frag');
+        assert.writeOK(ssl_frag.insert({_id: "large_str", foo: s}));
 
-    // SSL packets have a max size of ~16 kb so to test packet fragmentation support, create a
-    // string larger then 16kb.
-    for (let i = 0; i < 5 * 1700; i++) {
-        s += large;
+        const read = ssl_frag.find({_id: "large_str"}).toArray()[0].foo;
+        assert.eq(s, read, "Did not receive value written");
     }
 
-    let ssl_frag = conn.getCollection('test.ssl_frag');
-    assert.writeOK(ssl_frag.insert({_id: large}));
+    const options = {
+        sslMode: "requireSSL",
+        sslPEMKeyFile: "jstests/libs/server.pem",
+        networkMessageCompressors: 'disabled',
+    };
 
-    let docs = ssl_frag.find({});
-    assert.lt(2 * 16 * 1024, Object.bsonsize(docs), "test doc too small");
+    const mongod = MongoRunner.runMongod(options);
+    runTest(mongod);
+    MongoRunner.stopMongod(mongod);
 
-    MongoRunner.stopMongod(conn);
+    // TODO: Remove 'shardAsReplicaSet: false' when SERVER-32672 is fixed.
+    const st = new ShardingTest({
+        shards: 3,
+        mongos: 1,
+        config: 1,
+        other: {
+            configOptions: options,
+            mongosOptions: options,
+            shardOptions: options,
+            shardAsReplicaSet: false,
+        }
+    });
+    runTest(st.s0);
+    st.stop();
 })();
diff --git a/src/mongo/util/net/ssl/detail/impl/engine_apple.ipp b/src/mongo/util/net/ssl/detail/impl/engine_apple.ipp
index 523a4c714f1..f97e78893fb 100644
--- a/src/mongo/util/net/ssl/detail/impl/engine_apple.ipp
+++ b/src/mongo/util/net/ssl/detail/impl/engine_apple.ipp
@@ -42,6 +42,7 @@
 #include "mongo/util/mongoutils/str.h"
 #include "mongo/util/net/ssl/apple.hpp"
 #include "mongo/util/net/ssl/detail/engine.hpp"
+#include "mongo/util/net/ssl/detail/stream_core.hpp"
 #include "mongo/util/net/ssl/error.hpp"
 
 namespace asio {
@@ -49,6 +50,8 @@ namespace ssl {
 namespace detail {
 
 namespace {
+// Limit size of output buffer to avoid growing indefinitely.
+constexpr auto max_outbuf_size = stream_core::max_tls_record_size;
 
 const class osstatus_category : public error_category {
 public:
@@ -300,8 +303,11 @@ asio::mutable_buffer engine::get_output(const asio::mutable_buffer& data) {
 ::OSStatus engine::write_func(::SSLConnectionRef ctx, const void* data, size_t* data_len) {
     auto* this_ = const_cast<engine*>(static_cast<const engine*>(ctx));
     const auto* p = static_cast<const char*>(data);
+
+    const auto requested = *data_len;
+    *data_len = std::min<size_t>(requested, max_outbuf_size - this_->_outbuf.size());
     this_->_outbuf.insert(this_->_outbuf.end(), p, p + *data_len);
-    return ::errSecSuccess;
+    return (requested == *data_len) ? ::errSecSuccess : ::errSSLWouldBlock;
 }
 
 engine::want engine::read(const asio::mutable_buffer& data,
diff --git a/src/mongo/util/net/ssl/detail/io.hpp b/src/mongo/util/net/ssl/detail/io.hpp
index 6c0bc5f69fc..b943b1fb16f 100644
--- a/src/mongo/util/net/ssl/detail/io.hpp
+++ b/src/mongo/util/net/ssl/detail/io.hpp
@@ -51,7 +51,9 @@ std::size_t io(Stream& next_layer, stream_core& core, const Operation& op, asio:
                 core.output_ = core.engine_.get_output(core.output_buffer_);
                 // Get output data from the engine and write it to the underlying
                 // transport.
-                core.output_ += asio::write(next_layer, core.output_, ec);
+                while (!ec && core.output_.size()) {
+                    core.output_ += asio::write(next_layer, core.output_, ec);
+                }
 
                 // Try the operation again.
                 continue;
@@ -61,7 +63,9 @@ std::size_t io(Stream& next_layer, stream_core& core, const Operation& op, asio:
                 core.output_ = core.engine_.get_output(core.output_buffer_);
                 // Get output data from the engine and write it to the underlying
                 // transport.
-                core.output_ += asio::write(next_layer, core.output_, ec);
+                while (!ec && core.output_.size()) {
+                    core.output_ += asio::write(next_layer, core.output_, ec);
+                }
 
                 // Operation is complete. Return result to caller.
                 core.engine_.map_error_code(ec);