network: adding error handling for calling readDisable on a closed co…

…nnection (#3669)

Hopefully changing an outstanding tcp proxy session crash to a more minor ASSERT failure

Risk Level: Low
Testing: unit test of new code. integration tests which fail to repro the underlying bug.
Docs Changes: n/a
Release Notes: none

Hopefully ameliorates #3639

Signed-off-by: Alyssa Wilk <alyssar@chromium.org>

source/common/network/connection_impl.cc

@@ -224,6 +224,11 @@ void ConnectionImpl::enableHalfClose(bool enabled) {
 
 void ConnectionImpl::readDisable(bool disable) {
   ASSERT(state() == State::Open);
+  if (state() != State::Open || file_event_ == nullptr) {
+    // If readDisable is called on a closed connection in error, do not crash.
+    return;
+  }
+
   ENVOY_CONN_LOG(trace, "readDisable: enabled={} disable={}", *this, read_enabled_, disable);
 
   // When we disable reads, we still allow for early close notifications (the equivalent of

source/common/tcp_proxy/tcp_proxy.cc

@@ -487,6 +487,7 @@ void Filter::onUpstreamEvent(Network::ConnectionEvent event) {
 }
 
 void Filter::onIdleTimeout() {
+  ENVOY_CONN_LOG(debug, "Session timed out", read_callbacks_->connection());
   config_->stats().idle_timeout_.inc();
 
   // This results in also closing the upstream connection.

test/common/network/connection_impl_test.cc

@@ -447,6 +447,17 @@ TEST_P(ConnectionImplTest, ReadDisable) {
   disconnect(false);
 }
 
+// Regression test for (at least one failure mode of)
+// https://github.com/envoyproxy/envoy/issues/3639 where readDisable on a close
+// connection caused a crash.
+TEST_P(ConnectionImplTest, ReadDisableAfterClose) {
+  setUpBasicConnection();
+  disconnect(false);
+
+  EXPECT_DEBUG_DEATH(client_connection_->readDisable(true), "");
+  EXPECT_DEBUG_DEATH(client_connection_->readDisable(false), "");
+}
+
 TEST_P(ConnectionImplTest, EarlyCloseOnReadDisabledConnection) {
 #ifdef __APPLE__
   // On our current OSX build, the client connection does not get the early

test/integration/integration.cc

@@ -162,9 +162,15 @@ void IntegrationTcpClient::waitForData(const std::string& data) {
   connection_->dispatcher().run(Event::Dispatcher::RunType::Block);
 }
 
-void IntegrationTcpClient::waitForDisconnect() {
-  connection_->dispatcher().run(Event::Dispatcher::RunType::Block);
-  EXPECT_TRUE(disconnected_);
+void IntegrationTcpClient::waitForDisconnect(bool ignore_spurious_events) {
+  if (ignore_spurious_events) {
+    while (!disconnected_) {
+      connection_->dispatcher().run(Event::Dispatcher::RunType::Block);
+    }
+  } else {
+    connection_->dispatcher().run(Event::Dispatcher::RunType::Block);
+    EXPECT_TRUE(disconnected_);
+  }
 }
 
 void IntegrationTcpClient::waitForHalfClose() {

test/integration/integration.h

@@ -77,7 +77,7 @@ class IntegrationTcpClient {
 
   void close();
   void waitForData(const std::string& data);
-  void waitForDisconnect();
+  void waitForDisconnect(bool ignore_spurious_events = false);
   void waitForHalfClose();
   void readDisable(bool disabled);
   void write(const std::string& data, bool end_stream = false, bool verify = true);

test/integration/tcp_proxy_integration_test.cc

@@ -293,6 +293,56 @@ TEST_P(TcpProxyIntegrationTest, ShutdownWithOpenConnections) {
   // Success criteria is that no ASSERTs fire and there are no leaks.
 }
 
+TEST_P(TcpProxyIntegrationTest, TestIdletimeoutWithNoData) {
+  enable_half_close_ = false;
+  config_helper_.addConfigModifier([&](envoy::config::bootstrap::v2::Bootstrap& bootstrap) -> void {
+    auto* listener = bootstrap.mutable_static_resources()->mutable_listeners(0);
+    auto* filter_chain = listener->mutable_filter_chains(0);
+    auto* config_blob = filter_chain->mutable_filters(0)->mutable_config();
+
+    envoy::config::filter::network::tcp_proxy::v2::TcpProxy tcp_proxy_config;
+    MessageUtil::jsonConvert(*config_blob, tcp_proxy_config);
+    tcp_proxy_config.mutable_idle_timeout()->set_nanos(
+        std::chrono::duration_cast<std::chrono::nanoseconds>(std::chrono::milliseconds(100))
+            .count());
+    MessageUtil::jsonConvert(tcp_proxy_config, *config_blob);
+  });
+
+  initialize();
+  IntegrationTcpClientPtr tcp_client = makeTcpConnection(lookupPort("tcp_proxy"));
+  FakeRawConnectionPtr fake_upstream_connection = fake_upstreams_[0]->waitForRawConnection();
+  tcp_client->waitForDisconnect(true);
+  fake_upstream_connection->waitForDisconnect(true);
+}
+
+TEST_P(TcpProxyIntegrationTest, TestIdletimeoutWithLargeOutstandingData) {
+  config_helper_.setBufferLimits(1024, 1024);
+  enable_half_close_ = false;
+  config_helper_.addConfigModifier([&](envoy::config::bootstrap::v2::Bootstrap& bootstrap) -> void {
+    auto* listener = bootstrap.mutable_static_resources()->mutable_listeners(0);
+    auto* filter_chain = listener->mutable_filter_chains(0);
+    auto* config_blob = filter_chain->mutable_filters(0)->mutable_config();
+
+    envoy::config::filter::network::tcp_proxy::v2::TcpProxy tcp_proxy_config;
+    MessageUtil::jsonConvert(*config_blob, tcp_proxy_config);
+    tcp_proxy_config.mutable_idle_timeout()->set_nanos(
+        std::chrono::duration_cast<std::chrono::nanoseconds>(std::chrono::milliseconds(500))
+            .count());
+    MessageUtil::jsonConvert(tcp_proxy_config, *config_blob);
+  });
+
+  initialize();
+  IntegrationTcpClientPtr tcp_client = makeTcpConnection(lookupPort("tcp_proxy"));
+  FakeRawConnectionPtr fake_upstream_connection = fake_upstreams_[0]->waitForRawConnection();
+
+  std::string data(1024 * 16, 'a');
+  tcp_client->write(data);
+  fake_upstream_connection->write(data);
+
+  tcp_client->waitForDisconnect(true);
+  fake_upstream_connection->waitForDisconnect(true);
+}
+
 void TcpProxySslIntegrationTest::initialize() {
   config_helper_.addSslConfig();
   TcpProxyIntegrationTest::initialize();