http: mitigate delayed close timeout race with connection write buffer flush

api/envoy/config/filter/network/http_connection_manager/v2/http_connection_manager.proto

@@ -200,8 +200,10 @@ message HttpConnectionManager {
 
   // The delayed close timeout is for downstream connections managed by the HTTP connection manager.
   // It is defined as a grace period after connection close processing has been locally initiated
-  // during which Envoy will flush the write buffers for the connection and await the peer to close
-  // (i.e., a TCP FIN/RST is received by Envoy from the downstream connection).
+  // during which Envoy will wait for the peer to close (i.e., a TCP FIN/RST is received by Envoy
+  // from the downstream connection) prior to Envoy closing the socket associated with that
+  // connection. The timer associated with this timeout will be started only *after* all pending
+  // data in the connection's write buffer has been flushed.
   //
   // Delaying Envoy's connection close and giving the peer the opportunity to initiate the close
   // sequence mitigates a race condition that exists when downstream clients do not drain/process

docs/root/intro/version_history.rst

@@ -47,6 +47,7 @@ Version history
 * http: added :ref:`max request headers size <envoy_api_field_config.filter.network.http_connection_manager.v2.HttpConnectionManager.max_request_headers_kb>`. The default behaviour is unchanged.
 * http: added modifyDecodingBuffer/modifyEncodingBuffer to allow modifying the buffered request/response data.
 * http: added encodeComplete/decodeComplete. These are invoked at the end of the stream, after all data has been encoded/decoded respectively. Default implementation is a no-op.
+* http: fixed a bug with the :ref:`delayed_close_timeout<envoy_api_field_config.filter.network.http_connection_manager.v2.HttpConnectionManager.delayed_close_timeout>` where it could trigger prior to flushing the write buffer for the downstream connection.
 * outlier_detection: added support for :ref:`outlier detection event protobuf-based logging <arch_overview_outlier_detection_logging>`.
 * mysql: added a MySQL proxy filter that is capable of parsing SQL queries over MySQL wire protocol. Refer to :ref:`MySQL proxy<config_network_filters_mysql_proxy>` for more details.
 * performance: new buffer implementation (disabled by default; to test it, add "--use-libevent-buffers 0" to the command-line arguments when starting Envoy).

source/common/network/connection_impl.cc

@@ -99,6 +99,7 @@ void ConnectionImpl::close(ConnectionCloseType type) {
 
   uint64_t data_to_write = write_buffer_->length();
   ENVOY_CONN_LOG(debug, "closing data_to_write={} type={}", *this, data_to_write, enumToInt(type));
+  const bool delayed_close_timeout_set = delayedCloseTimeout().count() > 0;
   if (data_to_write == 0 || type == ConnectionCloseType::NoFlush ||
       !transport_socket_->canFlushClose()) {
     if (data_to_write > 0) {
@@ -107,7 +108,13 @@ void ConnectionImpl::close(ConnectionCloseType type) {
       transport_socket_->doWrite(*write_buffer_, true);
     }
 
-    closeSocket(ConnectionEvent::LocalClose);
+    if (type == ConnectionCloseType::FlushWriteAndDelay && delayed_close_timeout_set) {
+      // The socket is being closed and there is no more data to write. Since a delayed close has
+      // been requested, start the delayed close timer.
+      initializeDelayedCloseTimer();
+    } else {
+      closeSocket(ConnectionEvent::LocalClose);
+    }
   } else {
     ASSERT(type == ConnectionCloseType::FlushWrite ||
            type == ConnectionCloseType::FlushWriteAndDelay);
@@ -123,35 +130,21 @@ void ConnectionImpl::close(ConnectionCloseType type) {
     //          ConnectionManagerImpl::checkForDeferredClose()
     //     2) A second close is issued by a subsequent call to
     //        ConnectionManagerImpl::checkForDeferredClose() prior to returning from onData()
-    if (delayed_close_) {
+    if (inDelayedClose()) {
       return;
     }
 
-    delayed_close_ = true;
-    const bool delayed_close_timeout_set = delayedCloseTimeout().count() > 0;
-
-    // NOTE: the delayed close timeout (if set) affects both FlushWrite and FlushWriteAndDelay
-    // closes:
-    //   1. For FlushWrite, the timeout sets an upper bound on how long to wait for the flush to
-    //   complete before the connection is locally closed.
-    //   2. For FlushWriteAndDelay, the timeout specifies an upper bound on how long to wait for the
-    //   flush to complete and the peer to close the connection before it is locally closed.
-
     // All close types that follow do not actually close() the socket immediately so that buffered
     // data can be written. However, we do want to stop reading to apply TCP backpressure.
     read_enabled_ = false;
 
-    // Force a closeSocket() after the write buffer is flushed if the close_type calls for it or if
-    // no delayed close timeout is set.
-    close_after_flush_ = !delayed_close_timeout_set || type == ConnectionCloseType::FlushWrite;
-
-    // Create and activate a timer which will immediately close the connection if triggered.
-    // A config value of 0 disables the timeout.
-    if (delayed_close_timeout_set) {
-      delayed_close_timer_ = dispatcher_.createTimer([this]() -> void { onDelayedCloseTimeout(); });
-      ENVOY_CONN_LOG(debug, "setting delayed close timer with timeout {} ms", *this,
-                     delayedCloseTimeout().count());
-      delayed_close_timer_->enableTimer(delayedCloseTimeout());
+    if (!delayed_close_timeout_set || type == ConnectionCloseType::FlushWrite) {
+      // Force a closeSocket() after the write buffer is flushed if the close_type calls for it or
+      // if no delayed close timeout is set.
+      delayed_close_state_ = DelayedCloseState::CloseAfterFlush;
+    } else {
+      // A delayed close timer will be scheduled when the flush is completed.
+      delayed_close_state_ = DelayedCloseState::CloseAfterFlushAndTimeout;
     }
 
     file_event_->setEnabled(Event::FileReadyType::Write |
@@ -162,7 +155,7 @@ void ConnectionImpl::close(ConnectionCloseType type) {
 Connection::State ConnectionImpl::state() const {
   if (!ioHandle().isOpen()) {
     return State::Closed;
-  } else if (delayed_close_) {
+  } else if (inDelayedClose()) {
     return State::Closing;
   } else {
     return State::Open;
@@ -539,9 +532,14 @@ void ConnectionImpl::onWriteReady() {
     // write callback. This can happen if we manage to complete the SSL handshake in the write
     // callback, raise a connected event, and close the connection.
     closeSocket(ConnectionEvent::RemoteClose);
-  } else if ((close_after_flush_ && new_buffer_size == 0) || bothSidesHalfClosed()) {
+  } else if ((inDelayedClose() && new_buffer_size == 0) || bothSidesHalfClosed()) {
     ENVOY_CONN_LOG(debug, "write flush complete", *this);
-    closeSocket(ConnectionEvent::LocalClose);
+    if (delayed_close_state_ == DelayedCloseState::CloseAfterFlushAndTimeout) {
+      initializeDelayedCloseTimer();
+    } else {
+      ASSERT(bothSidesHalfClosed() || delayed_close_state_ == DelayedCloseState::CloseAfterFlush);
+      closeSocket(ConnectionEvent::LocalClose);
+    }
   } else if (result.action_ == PostIoAction::KeepOpen && result.bytes_processed_ > 0) {
     for (BytesSentCb& cb : bytes_sent_callbacks_) {
       cb(result.bytes_processed_);
@@ -587,13 +585,22 @@ bool ConnectionImpl::bothSidesHalfClosed() {
 }
 
 void ConnectionImpl::onDelayedCloseTimeout() {
+  delayed_close_timer_.reset(nullptr);
   ENVOY_CONN_LOG(debug, "triggered delayed close", *this);
   if (connection_stats_ != nullptr && connection_stats_->delayed_close_timeouts_ != nullptr) {
     connection_stats_->delayed_close_timeouts_->inc();
   }
   closeSocket(ConnectionEvent::LocalClose);
 }
 
+void ConnectionImpl::initializeDelayedCloseTimer() {
+  const auto timeout = delayedCloseTimeout().count();
+  ASSERT(timeout > 0);
+  delayed_close_timer_ = dispatcher_.createTimer([this]() -> void { onDelayedCloseTimeout(); });
+  ENVOY_CONN_LOG(debug, "setting delayed close timer with timeout {} ms", *this, timeout);
+  delayed_close_timer_->enableTimer(delayedCloseTimeout());
+}
+
 absl::string_view ConnectionImpl::transportFailureReason() const {
   return transport_socket_->failureReason();
 }

source/common/network/connection_impl.h

@@ -167,16 +167,22 @@ class ConnectionImpl : public virtual Connection,
   // Callback issued when a delayed close timeout triggers.
   void onDelayedCloseTimeout();
 
+  void initializeDelayedCloseTimer();
+  bool inDelayedClose() const { return delayed_close_state_ != DelayedCloseState::None; }
+
   static std::atomic<uint64_t> next_global_id_;
 
+  // States associated with delayed closing of the connection (i.e., when the underlying socket is
+  // not immediately close()d as a result of a ConnectionImpl::close()).
+  enum class DelayedCloseState { None, CloseAfterFlush, CloseAfterFlushAndTimeout };
+  DelayedCloseState delayed_close_state_{DelayedCloseState::None};
+
   Event::Dispatcher& dispatcher_;
   const uint64_t id_;
   Event::TimerPtr delayed_close_timer_;
   std::list<ConnectionCallbacks*> callbacks_;
   std::list<BytesSentCb> bytes_sent_callbacks_;
   bool read_enabled_{true};
-  bool close_after_flush_{false};
-  bool delayed_close_{false};
   bool above_high_watermark_{false};
   bool detect_early_close_{true};
   bool enable_half_close_{false};

test/common/network/connection_impl_test.cc

@@ -176,9 +176,11 @@ class ConnectionImplTest : public testing::TestWithParam<Address::IpVersion> {
 
 protected:
   struct ConnectionMocks {
-    std::unique_ptr<NiceMock<Event::MockDispatcher>> dispatcher;
-    Event::MockTimer* timer;
-    std::unique_ptr<NiceMock<MockTransportSocket>> transport_socket;
+    std::unique_ptr<NiceMock<Event::MockDispatcher>> dispatcher_;
+    Event::MockTimer* timer_;
+    std::unique_ptr<NiceMock<MockTransportSocket>> transport_socket_;
+    NiceMock<Event::MockFileEvent>* file_event_;
+    Event::FileReadyCb* file_ready_cb_;
   };
 
   ConnectionMocks createConnectionMocks() {
@@ -193,17 +195,20 @@ class ConnectionImplTest : public testing::TestWithParam<Address::IpVersion> {
 
     // This timer will be returned (transferring ownership) to the ConnectionImpl when createTimer()
     // is called to allocate the delayed close timer.
-    auto timer = new Event::MockTimer(dispatcher.get());
+    Event::MockTimer* timer = new Event::MockTimer(dispatcher.get());
 
-    auto file_event = std::make_unique<NiceMock<Event::MockFileEvent>>();
-    EXPECT_CALL(*dispatcher, createFileEvent_(0, _, _, _)).WillOnce(Return(file_event.release()));
+    NiceMock<Event::MockFileEvent>* file_event = new NiceMock<Event::MockFileEvent>;
+    EXPECT_CALL(*dispatcher, createFileEvent_(0, _, _, _))
+        .WillOnce(DoAll(SaveArg<1>(&file_ready_cb_), Return(file_event)));
 
     auto transport_socket = std::make_unique<NiceMock<MockTransportSocket>>();
-    EXPECT_CALL(*transport_socket, canFlushClose()).WillOnce(Return(true));
+    EXPECT_CALL(*transport_socket, canFlushClose()).WillRepeatedly(Return(true));
 
-    return ConnectionMocks{std::move(dispatcher), timer, std::move(transport_socket)};
+    return ConnectionMocks{std::move(dispatcher), timer, std::move(transport_socket), file_event,
+                           &file_ready_cb_};
   }
 
+  Event::FileReadyCb file_ready_cb_;
   Event::SimulatedTimeSystem time_system_;
   Api::ApiPtr api_;
   Event::DispatcherPtr dispatcher_;
@@ -987,41 +992,6 @@ TEST_P(ConnectionImplTest, FlushWriteCloseTest) {
   dispatcher_->run(Event::Dispatcher::RunType::Block);
 }
 
-// Test that a FlushWrite close will create and enable a timer which closes the connection when
-// triggered.
-TEST_P(ConnectionImplTest, FlushWriteCloseTimeoutTest) {
-  ConnectionMocks mocks = createConnectionMocks();
-  IoHandlePtr io_handle = std::make_unique<IoSocketHandleImpl>(0);
-  auto server_connection = std::make_unique<Network::ConnectionImpl>(
-      *mocks.dispatcher,
-      std::make_unique<ConnectionSocketImpl>(std::move(io_handle), nullptr, nullptr),
-      std::move(mocks.transport_socket), true);
-
-  InSequence s1;
-
-  // Enable delayed connection close processing by setting a non-zero timeout value. The actual
-  // value (> 0) doesn't matter since the callback is triggered below.
-  server_connection->setDelayedCloseTimeout(std::chrono::milliseconds(100));
-
-  NiceMockConnectionStats stats;
-  server_connection->setConnectionStats(stats.toBufferStats());
-
-  Buffer::OwnedImpl data("data");
-  server_connection->write(data, false);
-
-  // Data is pending in the write buffer, which will trigger the FlushWrite close to go into delayed
-  // close processing.
-  EXPECT_CALL(*mocks.timer, enableTimer(_)).Times(1);
-  server_connection->close(ConnectionCloseType::FlushWrite);
-
-  EXPECT_CALL(stats.delayed_close_timeouts_, inc()).Times(1);
-  // Since the callback is being invoked manually, disableTimer() will be called when the connection
-  // is closed by the callback.
-  EXPECT_CALL(*mocks.timer, disableTimer()).Times(1);
-  // Issue the delayed close callback to ensure connection is closed.
-  mocks.timer->callback_();
-}
-
 // Test that a FlushWriteAndDelay close causes Envoy to flush the write and wait for the client/peer
 // to close (until a configured timeout which is not expected to trigger in this test).
 TEST_P(ConnectionImplTest, FlushWriteAndDelayCloseTest) {
@@ -1144,27 +1114,85 @@ TEST_P(ConnectionImplTest, FlushWriteAndDelayConfigDisabledTest) {
   server_connection->close(ConnectionCloseType::NoFlush);
 }
 
+// Test that the delayed close timer is created when a connection is closed while having a write
+// buffer pending to flush.
+TEST_P(ConnectionImplTest, DelayedCloseTimeoutWithPendingWriteBuffer) {
+  ConnectionMocks mocks = createConnectionMocks();
+  MockTransportSocket* transport_socket = mocks.transport_socket_.get();
+  IoHandlePtr io_handle = std::make_unique<IoSocketHandleImpl>(0);
+  auto server_connection = std::make_unique<Network::ConnectionImpl>(
+      *mocks.dispatcher_,
+      std::make_unique<ConnectionSocketImpl>(std::move(io_handle), nullptr, nullptr),
+      std::move(mocks.transport_socket_), true);
+
+  InSequence s1;
+
+  // The actual timeout is insignificant, we just need to enable delayed close processing by
+  // setting it to > 0.
+  server_connection->setDelayedCloseTimeout(std::chrono::milliseconds(100));
+
+  EXPECT_CALL(*mocks.file_event_, activate(Event::FileReadyType::Write))
+      .WillOnce(Invoke(*mocks.file_ready_cb_));
+  EXPECT_CALL(*transport_socket, doWrite(BufferStringEqual("data"), _))
+      .WillOnce(Invoke([&](Buffer::Instance&, bool) -> IoResult {
+        // The first time doWrite() is called will be after the server_connection->write(). Avoid
+        // flushing the write buffer.
+        return IoResult{PostIoAction::KeepOpen, 0, false};
+      }))
+      .WillOnce(Invoke([&](Buffer::Instance& buffer, bool) -> IoResult {
+        // The second time doWrite() is called is after server_connection->close() has been called
+        // and the file_ready_cb is triggered (simulating a write ready event triggering on the
+        // socket).
+        buffer.drain(buffer.length());
+        return IoResult{PostIoAction::KeepOpen, buffer.length(), false};
+      }));
+
+  Buffer::OwnedImpl data("data");
+  server_connection->write(data, false);
+
+  // The delayed close timer won't be activated after this call since there is pending data in the
+  // write buffer.
+  server_connection->close(ConnectionCloseType::FlushWriteAndDelay);
+
+  // Flush the buffer and verify that the delayed close timer is enabled.
+  EXPECT_CALL(*mocks.timer_, enableTimer(_)).Times(1);
+  (*mocks.file_ready_cb_)(Event::FileReadyType::Write);
+
+  // Force the delayed close timeout to trigger so the connection is cleaned up.
+  mocks.timer_->callback_();
+}
+
 // Test that tearing down the connection will disable the delayed close timer.
 TEST_P(ConnectionImplTest, DelayedCloseTimeoutDisableOnSocketClose) {
   ConnectionMocks mocks = createConnectionMocks();
+  MockTransportSocket* transport_socket = mocks.transport_socket_.get();
   IoHandlePtr io_handle = std::make_unique<IoSocketHandleImpl>(0);
   auto server_connection = std::make_unique<Network::ConnectionImpl>(
-      *mocks.dispatcher,
+      *mocks.dispatcher_,
       std::make_unique<ConnectionSocketImpl>(std::move(io_handle), nullptr, nullptr),
-      std::move(mocks.transport_socket), true);
+      std::move(mocks.transport_socket_), true);
 
   InSequence s1;
 
-  // The actual timeout is insignificant, we just need to enable delayed close processing by setting
-  // it to > 0.
+  // The actual timeout is insignificant, we just need to enable delayed close processing by
+  // setting it to > 0.
   server_connection->setDelayedCloseTimeout(std::chrono::milliseconds(100));
 
   Buffer::OwnedImpl data("data");
+  EXPECT_CALL(*mocks.file_event_, activate(Event::FileReadyType::Write))
+      .WillOnce(Invoke(*mocks.file_ready_cb_));
+  // The buffer must be drained when write() is called on the connection to allow the close() to
+  // enable the timer.
+  EXPECT_CALL(*transport_socket, doWrite(BufferStringEqual("data"), _))
+      .WillOnce(Invoke([&](Buffer::Instance& buffer, bool) -> IoResult {
+        buffer.drain(buffer.length());
+        return IoResult{PostIoAction::KeepOpen, buffer.length(), false};
+      }));
   server_connection->write(data, false);
-  EXPECT_CALL(*mocks.timer, enableTimer(_)).Times(1);
+  EXPECT_CALL(*mocks.timer_, enableTimer(_)).Times(1);
   // Enable the delayed close timer.
   server_connection->close(ConnectionCloseType::FlushWriteAndDelay);
-  EXPECT_CALL(*mocks.timer, disableTimer()).Times(1);
+  EXPECT_CALL(*mocks.timer_, disableTimer()).Times(1);
   // This close() will call closeSocket(), which should disable the timer to avoid triggering it
   // after the connection's data structures have been reset.
   server_connection->close(ConnectionCloseType::NoFlush);
@@ -1173,11 +1201,12 @@ TEST_P(ConnectionImplTest, DelayedCloseTimeoutDisableOnSocketClose) {
 // Test that the delayed close timeout callback is resilient to connection teardown edge cases.
 TEST_P(ConnectionImplTest, DelayedCloseTimeoutNullStats) {
   ConnectionMocks mocks = createConnectionMocks();
+  MockTransportSocket* transport_socket = mocks.transport_socket_.get();
   IoHandlePtr io_handle = std::make_unique<IoSocketHandleImpl>(0);
   auto server_connection = std::make_unique<Network::ConnectionImpl>(
-      *mocks.dispatcher,
+      *mocks.dispatcher_,
       std::make_unique<ConnectionSocketImpl>(std::move(io_handle), nullptr, nullptr),
-      std::move(mocks.transport_socket), true);
+      std::move(mocks.transport_socket_), true);
 
   InSequence s1;
 
@@ -1190,14 +1219,24 @@ TEST_P(ConnectionImplTest, DelayedCloseTimeoutNullStats) {
   // that edge case.
 
   Buffer::OwnedImpl data("data");
+  EXPECT_CALL(*mocks.file_event_, activate(Event::FileReadyType::Write))
+      .WillOnce(Invoke(*mocks.file_ready_cb_));
+  // The buffer must be drained when write() is called on the connection to allow the close() to
+  // enable the timer.
+  EXPECT_CALL(*transport_socket, doWrite(BufferStringEqual("data"), _))
+      .WillOnce(Invoke([&](Buffer::Instance& buffer, bool) -> IoResult {
+        buffer.drain(buffer.length());
+        return IoResult{PostIoAction::KeepOpen, buffer.length(), false};
+      }));
   server_connection->write(data, false);
 
-  EXPECT_CALL(*mocks.timer, enableTimer(_)).Times(1);
+  EXPECT_CALL(*mocks.timer_, enableTimer(_)).Times(1);
   server_connection->close(ConnectionCloseType::FlushWriteAndDelay);
-  EXPECT_CALL(*mocks.timer, disableTimer()).Times(1);
+  EXPECT_CALL(*mocks.timer_, disableTimer()).Times(1);
   // Copy the callback since mocks.timer will be freed when closeSocket() is called.
-  Event::TimerCb callback = mocks.timer->callback_;
-  // The following close() will call closeSocket() and reset internal data structures such as stats.
+  Event::TimerCb callback = mocks.timer_->callback_;
+  // The following close() will call closeSocket() and reset internal data structures such as
+  // stats.
   server_connection->close(ConnectionCloseType::NoFlush);
   // Verify the onDelayedCloseTimeout() callback is resilient to the post closeSocket(), pre
   // destruction state. This should not actually happen due to the timeout disablement in