Guard unset ingress source during CONNECT teardown

Summary:
HTTPConnectStream constructs the transport before awaiting sendRequest(), but ingressSource_ is assigned only after that await completes. If CONNECT setup is cancelled or throws first, destruction calls close() -> shutdownRead() -> canRead() and dereferences the unset shared_ptr.

Guard ingressSource_ in canRead() so partially constructed streams report that they are not readable. Apply the change to both the fbcode and xplat mirrors.

Keep the existing ConnectCancelledBeforeIngressAssigned regression test. It reproduces the production destructor stack without the guard and passes once the guard is present.

This fixes the recurring crash tracked by T277272947, T283037317, and T284870008.

Reviewed By: jbeshay

Differential Revision: D114422937

fbshipit-source-id: e064f038a3034a51f8c220b61a207d9c90fdb9f3

Author: sandarsh

proxygen/lib/http/coro/transport/HTTPConnectStream.cpp

@@ -167,7 +167,7 @@ void HTTPConnectStream::close() {
 }
 
 bool HTTPConnectStream::canRead() const {
-  return *ingressSource_;
+  return ingressSource_ && *ingressSource_;
 }
 
 bool HTTPConnectStream::canWrite() const {

proxygen/lib/http/coro/transport/test/HTTPConnectTransportTest.cpp

@@ -289,6 +289,25 @@ TEST_F(HTTPConnectTransportTest, ConnectCancelled) {
   });
 }
 
+TEST_F(HTTPConnectTransportTest, ConnectCancelledBeforeIngressAssigned) {
+  run([&]() -> Task<> {
+    auto session = co_await HTTPCoroConnector::connect(
+        &evb, *srv->address(), 0ms, getConnParams());
+    auto reservation = session->reserveRequest();
+    // Cancel after the session exists so connectUnique constructs the stream.
+    cancelSource.requestCancellation();
+    auto result = co_await folly::coro::co_awaitTry(co_withCancellation(
+        cancelSource.getToken(),
+        HTTPConnectStream::connectUnique(session,
+                                         std::move(*reservation),
+                                         kAuthority,
+                                         std::chrono::seconds(1),
+                                         {{"Foo", "Bar"}},
+                                         egressBufferSize)));
+    EXPECT_TRUE(result.hasException());
+  });
+}
+
 TEST_F(HTTPConnectTransportTest, SimpleReadWrite) {
   run([&]() -> Task<> {
     // Should fill window