contrib/golang: address race condition in Go HTTP filter

addresses #44320

the issue here arises when the client sends data and the data is pending
to be dispatched. Previously, the stream could have been destroyed
already by the time `continueStatusInternal` was invoked. This PR adds
safety-checks in the Go shim, and on the C++ filter side to avoid it

this was tested by building Envoy + a test client that sends HTTP/1.1 +
POST with a body, and verifying the crash happens before consistently,
and no longer crashes with the patch

Signed-off-by: Henry Wang <henry.wang@datadoghq.com>

Author: henrymwang

contrib/golang/filters/http/source/go/pkg/http/shim.go

@@ -157,6 +157,9 @@ func getRequest(r *C.httpRequest) *httpRequest {
 func getState(s *C.processState) *processState {
 	r := s.req
 	req := getRequest(r)
+	if req == nil {
+		return nil
+	}
 	if s.is_encoding == 0 {
 		return &req.decodingState.processState
 	}
@@ -237,6 +240,9 @@ func envoyGoFilterOnHttpHeader(s *C.processState, endStream, headerNum, headerBy
 //export envoyGoFilterOnHttpData
 func envoyGoFilterOnHttpData(s *C.processState, endStream, buffer, length uint64) uint64 {
 	state := getState(s)
+	if state == nil {
+		return uint64(api.Continue)
+	}
 
 	req := state.request
 	if req.pInfo.paniced {
@@ -389,6 +395,9 @@ func envoyGoFilterOnHttpDestroy(r *C.httpRequest, reason uint64) {
 //export envoyGoRequestSemaDec
 func envoyGoRequestSemaDec(r *C.httpRequest) {
 	req := getRequest(r)
+	if req == nil {
+		return
+	}
 	defer req.recoverPanic()
 	req.resumeWaitCallback()
 }

contrib/golang/filters/http/source/golang_filter.cc

@@ -404,12 +404,19 @@ void Filter::continueStatusInternal(ProcessorState& state, GolangStatus status)
 
     case FilterState::ProcessingTrailer:
       state.continueDoData();
+      if (hasDestroyed()) {
+        return;
+      }
       state.continueProcessing();
       break;
 
     default:
       ASSERT(0, "unexpected state");
     }
+
+    if (hasDestroyed()) {
+      return;
+    }
   }
 
   ENVOY_LOG(debug,
@@ -426,6 +433,9 @@ void Filter::continueStatusInternal(ProcessorState& state, GolangStatus status)
     auto done = doDataGo(state, state.getBufferData(), state.getEndStream());
     if (done) {
       state.continueDoData();
+      if (hasDestroyed()) {
+        return;
+      }
     } else {
       // do not process trailers when data is not finished
       return;

contrib/golang/filters/http/test/golang_integration_test.cc

@@ -2128,4 +2128,31 @@ TEST_P(GolangIntegrationTest, DrainConnectionUponCompletion) {
   cleanupUpstreamAndDownstream();
 }
 
+// Regression test for https://github.com/envoyproxy/envoy/issues/44320.
+TEST_P(GolangIntegrationTest, AsyncContinueWithNoHealthyUpstream) {
+  config_helper_.addConfigModifier([](envoy::config::bootstrap::v3::Bootstrap& bootstrap) {
+    auto* cluster = bootstrap.mutable_static_resources()->mutable_clusters(0);
+    cluster->mutable_load_assignment()->mutable_endpoints(0)->clear_lb_endpoints();
+  });
+
+  initializeBasicFilter(BASIC, "test.com");
+
+  codec_client_ = makeHttpConnection(makeClientConnection(lookupPort("http")));
+
+  Http::TestRequestHeaderMapImpl request_headers{
+      {":method", "POST"},        {":path", "/test?async=1&sleep=1"}, {":scheme", "http"},
+      {":authority", "test.com"}, {"x-test-header-0", "foo"},
+  };
+
+  auto encoder_decoder = codec_client_->startRequest(request_headers);
+  Http::RequestEncoder& request_encoder = encoder_decoder.first;
+  auto response = std::move(encoder_decoder.second);
+  codec_client_->sendData(request_encoder, "request body", true);
+
+  ASSERT_TRUE(response->waitForEndStream());
+  EXPECT_EQ("503", response->headers().getStatusValue());
+
+  cleanup();
+}
+
 } // namespace Envoy