thrift proxy: fix crash when using payload_passthrough (#14723)

We started seeing crashes triggered by ConnectionManager::passthroughEnabled()
once we enabled `payload_passthrough`. That code assumes that there will
_always_ be an active RPC. However, this is not true after a local response
has been sent (e.g.: no healthy upstream, no cluster, no route, etc.).

Risk Level: low
Testing: unit tests added
Doc Changes: n/a
Release Notes: n/a
Signed-off-by: Raul Gutierrez Segales <rgs@pinterest.com>

source/extensions/filters/network/thrift_proxy/conn_manager.cc

@@ -181,9 +181,15 @@ bool ConnectionManager::passthroughEnabled() const {
     return false;
   }
 
-  // This is called right after the metadata has been parsed, and the ActiveRpc being processed must
-  // be in the rpcs_ list.
-  ASSERT(!rpcs_.empty());
+  // If the rpcs list is empty, a local response happened.
+  //
+  // TODO(rgs1): we actually could still enable passthrough for local
+  // responses as long as the transport is framed and the protocol is
+  // not Twitter.
+  if (rpcs_.empty()) {
+    return false;
+  }
+
   return (*rpcs_.begin())->passthroughSupported();
 }
 

test/extensions/filters/network/thrift_proxy/conn_manager_test.cc

@@ -1641,6 +1641,55 @@ stat_prefix: test
   filter_callbacks_.connection_.dispatcher_.clearDeferredDeleteList();
 }
 
+// When a local reply was sent, payload passthrough is disabled because there's no
+// active RPC left.
+TEST_F(ThriftConnectionManagerTest, NoPayloadPassthroughOnLocalReply) {
+  const std::string yaml = R"EOF(
+transport: FRAMED
+protocol: BINARY
+payload_passthrough: true
+stat_prefix: test
+route_config:
+  name: "routes"
+  routes:
+    - match:
+        method_name: not_handled
+      route:
+        cluster: cluster
+)EOF";
+
+  initializeFilter(yaml);
+  writeFramedBinaryMessage(buffer_, MessageType::Oneway, 0x0F);
+
+  EXPECT_CALL(*decoder_filter_, passthroughSupported()).WillRepeatedly(Return(true));
+  EXPECT_CALL(*decoder_filter_, passthroughData(_)).Times(0);
+
+  ThriftFilters::DecoderFilterCallbacks* callbacks{};
+  EXPECT_CALL(*decoder_filter_, setDecoderFilterCallbacks(_))
+      .WillOnce(
+          Invoke([&](ThriftFilters::DecoderFilterCallbacks& cb) -> void { callbacks = &cb; }));
+
+  NiceMock<MockDirectResponse> direct_response;
+  EXPECT_CALL(direct_response, encode(_, _, _))
+      .WillOnce(Invoke([&](MessageMetadata&, Protocol&,
+                           Buffer::Instance& buffer) -> DirectResponse::ResponseType {
+        buffer.add("response");
+        return DirectResponse::ResponseType::ErrorReply;
+      }));
+
+  EXPECT_CALL(*decoder_filter_, messageBegin(_))
+      .WillOnce(Invoke([&](MessageMetadataSharedPtr) -> FilterStatus {
+        callbacks->sendLocalReply(direct_response, false);
+        return FilterStatus::StopIteration;
+      }));
+
+  EXPECT_EQ(filter_->onData(buffer_, false), Network::FilterStatus::StopIteration);
+  EXPECT_EQ(1U, store_.counter("test.request").value());
+
+  Router::RouteConstSharedPtr route = callbacks->route();
+  EXPECT_EQ(nullptr, route);
+}
+
 } // namespace ThriftProxy
 } // namespace NetworkFilters
 } // namespace Extensions