Accept PUSH_PROMISE before response headers

Summary:
During TGC CONNECT canary validation, Cloudflare-hosted domains consistently failed over HTTP/2 with `Invalid ingress state transition, state=0, event=3`, while HTTP/1.1 worked. We reproduced this on `fwdproxy.devtest.ftw` through Fastly, Cloudflare, and direct CONNECT; a libcurl control succeeded. Representative URLs included:

- `https://rohitai.com/blog/atria-dawn-preview-open-weight-agentic-model`
- `https://www.thescore.com/nfl/news/3600712/browns-place-rb-sampson-on-ir-with-knee-injury`
- `https://app.warehouserunner.com/costco/1748800-casper-cooling-select-12-memory-foam-queen-mattress`
- `https://www.reflector.com/news/national/kaplan-fox-notifies-equipmentshare-com-inc-nasdaq-eqpt-investors-of-a-pending-securities/article_3f5265d9-de9a-508d-8928-72edfad09e0a.html`

TGC instrumentation showed Cloudflare sending a legal `PUSH_PROMISE` before the parent response headers. `HTTPStreamSource` classified push promises as body events, so the ingress state machine rejected this ordering while still in `Start`. Add an explicit `onPushPromise` ingress event that preserves the current response state, and queue early pushes without waking a coroutine waiting for response headers.

Reviewed By: mjoras

Differential Revision: D120570583

fbshipit-source-id: 1952c0f30f724560ef7d6baeeaa9faf14e5b4259

proxygen/lib/http/coro/HTTPStreamSource.cpp

@@ -117,16 +117,17 @@ void HTTPStreamSource::datagram(std::unique_ptr<folly::IOBuf> datagram) {
 void HTTPStreamSource::pushPromise(std::unique_ptr<HTTPMessage> promise,
                                    HTTPSource* pushSource,
                                    bool eom) {
-  // Treat this like a body event for state machine purposes.  Legal
-  // anytime after headers and before eom
   XCHECK(pushSource);
-  if (!validateStateTransition(HTTPTransactionIngressSM::Event::onBody, eom)) {
+  if (!validateStateTransition(HTTPTransactionIngressSM::Event::onPushPromise,
+                               eom)) {
     pushSource->stopReading();
     return;
   }
   if (!sinkMode_) {
     bodyQueue_.emplace_back(std::move(promise), pushSource, eom);
-    event_.signal();
+    if (readState_ == ReadState::BodyEvents) {
+      event_.signal();
+    }
   } else {
     pushSource->stopReading();
   }

proxygen/lib/http/coro/test/HTTPSourceTests.cpp

@@ -224,6 +224,30 @@ TEST_F(HTTPStreamSourceTest, Body) {
   EXPECT_EQ(bodyEvents_[4].eom, true);
 }
 
+TEST_F(HTTPStreamSourceTest, PushPromiseBeforeResponseHeaders) {
+  auto promiseSource = HTTPFixedSource::makeFixedResponse(
+      200, folly::IOBuf::copyBuffer("push body"));
+  auto promise = std::make_unique<HTTPMessage>();
+  promise->setURL("/push");
+  stream_.pushPromise(std::move(promise), promiseSource);
+
+  auto response = std::make_unique<HTTPMessage>();
+  response->setStatusCode(200);
+  stream_.headers(std::move(response));
+  stream_.eom();
+
+  co_withExecutor(&evb_, drainSource()).start();
+  run();
+
+  EXPECT_FALSE(error_.hasValue());
+  ASSERT_EQ(headerEvents_.size(), 1);
+  EXPECT_EQ(headerEvents_[0].headers->getStatusCode(), 200);
+  ASSERT_EQ(bodyEvents_.size(), 1);
+  EXPECT_EQ(bodyEvents_[0].eventType, HTTPBodyEvent::PUSH_PROMISE);
+  EXPECT_EQ(bodyEvents_[0].event.push.promise->getPathAsStringPiece(), "/push");
+  EXPECT_TRUE(bodyEvents_[0].eom);
+}
+
 TEST_F(HTTPStreamSourceTest, ContentLength0) {
   auto req = makePostRequest(0);
   stream_.headers(std::move(req), false);

proxygen/lib/http/session/HTTPTransactionIngressSM.cpp

@@ -35,8 +35,12 @@ HTTPTransactionIngressSMData::find(HTTPTransactionIngressSMData::State s,
           static_cast<uint64_t>(Event::NumEvents),
           {{{State::Start, Event::onFinalHeaders}, State::FinalHeadersReceived},
 
+           {{State::Start, Event::onPushPromise}, State::Start},
+
            {{State::Start, Event::onNonFinalHeaders},
             State::NonFinalHeadersReceived},
+           {{State::NonFinalHeadersReceived, Event::onPushPromise},
+            State::NonFinalHeadersReceived},
            {{State::NonFinalHeadersReceived, Event::onNonFinalHeaders},
             State::NonFinalHeadersReceived},
            {{State::NonFinalHeadersReceived, Event::onFinalHeaders},
@@ -46,6 +50,8 @@ HTTPTransactionIngressSMData::find(HTTPTransactionIngressSMData::State s,
 
            {{State::FinalHeadersReceived, Event::onBody},
             State::RegularBodyReceived},
+           {{State::FinalHeadersReceived, Event::onPushPromise},
+            State::FinalHeadersReceived},
            {{State::FinalHeadersReceived, Event::onDatagram},
             State::FinalHeadersReceived},
            {{State::FinalHeadersReceived, Event::onChunkHeader},
@@ -59,6 +65,8 @@ HTTPTransactionIngressSMData::find(HTTPTransactionIngressSMData::State s,
 
            {{State::RegularBodyReceived, Event::onBody},
             State::RegularBodyReceived},
+           {{State::RegularBodyReceived, Event::onPushPromise},
+            State::RegularBodyReceived},
            {{State::RegularBodyReceived, Event::onDatagram},
             State::RegularBodyReceived},
            // HTTP2 supports trailers and doesn't handle body as chunked events
@@ -163,6 +171,9 @@ std::ostream& operator<<(std::ostream& os,
     case HTTPTransactionIngressSMData::Event::onEOM:
       os << "onEOM";
       break;
+    case HTTPTransactionIngressSMData::Event::onPushPromise:
+      os << "onPushPromise";
+      break;
     case HTTPTransactionIngressSMData::Event::eomFlushed:
       os << "eomFlushed";
       break;

proxygen/lib/http/session/HTTPTransactionIngressSM.h

@@ -44,6 +44,7 @@ class HTTPTransactionIngressSMData {
     onTrailers,
     onUpgrade,
     onEOM,
+    onPushPromise,
     // Internal state transitions
     eomFlushed,
 

proxygen/lib/http/session/test/HTTPTransactionSMTest.cpp

@@ -227,6 +227,18 @@ TEST_F(IngressStateMachineFixture, NormalIngressTransitions) {
   follow(HTTPTransactionIngressSM::Event::onEOM);
 }
 
+TEST_F(IngressStateMachineFixture, PushPromiseTransitions) {
+  follow(HTTPTransactionIngressSM::Event::onPushPromise);
+  follow(HTTPTransactionIngressSM::Event::onNonFinalHeaders);
+  follow(HTTPTransactionIngressSM::Event::onPushPromise);
+  follow(HTTPTransactionIngressSM::Event::onFinalHeaders);
+  follow(HTTPTransactionIngressSM::Event::onPushPromise);
+  follow(HTTPTransactionIngressSM::Event::onBody);
+  follow(HTTPTransactionIngressSM::Event::onPushPromise);
+  follow(HTTPTransactionIngressSM::Event::onEOM);
+  fail(HTTPTransactionIngressSM::Event::onPushPromise);
+}
+
 TEST_F(IngressStateMachineFixture, WeirdIngressTransitions) {
   follow(HTTPTransactionIngressSM::Event::onNonFinalHeaders);
   follow(HTTPTransactionIngressSM::Event::onNonFinalHeaders);