dynamic modules: keep C++ SDK response callbacks after a local reply

changelogs/current/bug_fixes/dynamic_modules__cpp-sdk-response-callbacks-after-local-reply.rst

@@ -0,0 +1,5 @@
+dynamic modules: fixed a bug in the C++ SDK where a filter stopped receiving
+``onResponseHeaders``, ``onResponseBody`` and ``onResponseTrailers`` after any local reply on the
+stream, including local replies the module did not send. A module could not observe or modify the
+response for a ``direct_response`` route, a local reply from another filter, or an
+Envoy-generated error. The Rust and Go SDKs were unaffected.

source/extensions/dynamic_modules/sdk/cpp/sdk_internal.cc

@@ -1530,7 +1530,6 @@ envoy_dynamic_module_on_http_filter_local_reply(
   if (plugin_handle == nullptr) {
     return envoy_dynamic_module_type_on_http_filter_local_reply_status_Continue;
   }
-  plugin_handle->local_reply_sent_ = true;
   return static_cast<envoy_dynamic_module_type_on_http_filter_local_reply_status>(
       plugin_handle->plugin_->onLocalReply(
           response_code,

test/extensions/dynamic_modules/http/integration_test.cc

@@ -259,6 +259,48 @@ TEST_P(DynamicModulesIntegrationTest, LogLevel) {
                         .getStringView());
 }
 
+// A `direct_response` route is served as a local reply. The module did not send it, so its
+// response callbacks must still run. Regression test for the C++ SDK, which set the same
+// `local_reply_sent_` flag on the local-reply notification that it sets when the module itself
+// calls `sendLocalResponse()`, and then skipped every response callback for the stream.
+TEST_P(DynamicModulesIntegrationTest, ResponseCallbacksOnLocalReply) {
+#ifdef __APPLE__
+  if (GetParam() == "go") {
+    // Not this test: with a Go module loaded, ~IntegrationTestServer never returns, because the
+    // exiting server thread runs macOS pthread TSD destructors and one of them enters the Go
+    // runtime and does not come back. The request itself succeeds. See #46905. Scoped to Apple
+    // platforms because Go is the only SDK here whose runtime installs TSD destructors and this
+    // has not been seen on Linux.
+    GTEST_SKIP() << "Go module deadlocks server teardown on macOS, see #46905";
+  }
+#endif
+  config_helper_.addConfigModifier(
+      [](envoy::extensions::filters::network::http_connection_manager::v3::HttpConnectionManager&
+             hcm) {
+        auto* route = hcm.mutable_route_config()->mutable_virtual_hosts(0)->mutable_routes(0);
+        route->clear_route();
+        auto* direct_response = route->mutable_direct_response();
+        direct_response->set_status(200);
+        direct_response->mutable_body()->set_inline_string("ok");
+      });
+  initializeFilter("local_reply_response_headers");
+  codec_client_ = makeHttpConnection(makeClientConnection(lookupPort("http")));
+
+  auto response = codec_client_->makeHeaderOnlyRequest(
+      Http::TestRequestHeaderMapImpl{{":method", "GET"},
+                                     {":path", "/test/long/url"},
+                                     {":scheme", "http"},
+                                     {":authority", "host"}});
+  ASSERT_TRUE(response->waitForEndStream());
+
+  EXPECT_TRUE(response->complete());
+  EXPECT_EQ("200", response->headers().Status()->value().getStringView());
+  EXPECT_EQ("called", response->headers()
+                          .get(Http::LowerCaseString("on-response-headers"))[0]
+                          ->value()
+                          .getStringView());
+}
+
 TEST_P(DynamicModulesIntegrationTest, HeaderCallbacks) { runHeaderCallbacksTest(false); }
 
 TEST_P(DynamicModulesIntegrationTest, HeaderCallbacksWithUpstreamFilter) {

test/extensions/dynamic_modules/test_data/cpp/http_integration_test.cc

@@ -170,6 +170,40 @@ class PassthroughConfigFactory : public HttpFilterConfigFactory {
 
 REGISTER_HTTP_FILTER_CONFIG_FACTORY(PassthroughConfigFactory, "passthrough");
 
+// Only records that its response-headers callback ran. Used to check that the callback still fires
+// when the response is a local reply the module did not send, such as a `direct_response` route.
+class LocalReplyResponseHeadersFilter : public HttpFilter {
+public:
+  HeadersStatus onRequestHeaders(HeaderMap&, bool) override { return HeadersStatus::Continue; }
+  HeadersStatus onResponseHeaders(HeaderMap& headers, bool) override {
+    headers.set("on-response-headers", "called");
+    return HeadersStatus::Continue;
+  }
+  BodyStatus onRequestBody(BodyBuffer&, bool) override { return BodyStatus::Continue; }
+  BodyStatus onResponseBody(BodyBuffer&, bool) override { return BodyStatus::Continue; }
+  TrailersStatus onRequestTrailers(HeaderMap&) override { return TrailersStatus::Continue; }
+  TrailersStatus onResponseTrailers(HeaderMap&) override { return TrailersStatus::Continue; }
+  void onStreamComplete() override {}
+  void onDestroy() override {}
+};
+
+class LocalReplyResponseHeadersFilterFactory : public HttpFilterFactory {
+public:
+  std::unique_ptr<HttpFilter> create(HttpFilterHandle&) override {
+    return std::make_unique<LocalReplyResponseHeadersFilter>();
+  }
+};
+
+class LocalReplyResponseHeadersConfigFactory : public HttpFilterConfigFactory {
+public:
+  std::unique_ptr<HttpFilterFactory> create(HttpFilterConfigHandle&, std::string_view) override {
+    return std::make_unique<LocalReplyResponseHeadersFilterFactory>();
+  }
+};
+
+REGISTER_HTTP_FILTER_CONFIG_FACTORY(LocalReplyResponseHeadersConfigFactory,
+                                    "local_reply_response_headers");
+
 // -----------------------------------------------------------------------------
 // HeaderCallbacks
 // -----------------------------------------------------------------------------

test/extensions/dynamic_modules/test_data/go/http_integration_test/http_integration_test.go

@@ -18,6 +18,7 @@ import (
 func init() {
 	sdk.RegisterHttpFilterConfigFactories(map[string]shared.HttpFilterConfigFactory{
 		"passthrough":                  &PassthroughConfigFactory{},
+		"local_reply_response_headers": &LocalReplyResponseHeadersConfigFactory{},
 		"header_callbacks_on_creation": &HeaderCallbacksOnCreationConfigFactory{},
 		"header_callbacks":             &HeaderCallbacksConfigFactory{},
 		"per_route_config":             &PerRouteConfigFactory{},
@@ -101,6 +102,36 @@ func (p *ConfigSchedulerFilter) OnRequestHeaders(headers shared.HeaderMap,
 // Passthrough
 // -----------------------------------------------------------------------------
 
+// LocalReplyResponseHeadersConfigFactory builds a filter that only records that its
+// response-headers callback ran. Used to check that the callback still fires when the response is a
+// local reply the module did not send, such as a `direct_response` route.
+type LocalReplyResponseHeadersConfigFactory struct {
+	shared.EmptyHttpFilterConfigFactory
+}
+
+func (f *LocalReplyResponseHeadersConfigFactory) Create(_ shared.HttpFilterConfigHandle,
+	_ []byte) (shared.HttpFilterFactory, error) {
+	return &LocalReplyResponseHeadersFilterFactory{}, nil
+}
+
+type LocalReplyResponseHeadersFilterFactory struct {
+	shared.EmptyHttpFilterFactory
+}
+
+func (f *LocalReplyResponseHeadersFilterFactory) Create(shared.HttpFilterHandle) shared.HttpFilter {
+	return &LocalReplyResponseHeadersFilter{}
+}
+
+type LocalReplyResponseHeadersFilter struct {
+	shared.EmptyHttpFilter
+}
+
+func (f *LocalReplyResponseHeadersFilter) OnResponseHeaders(headers shared.HeaderMap,
+	_ bool) shared.HeadersStatus {
+	headers.Set("on-response-headers", "called")
+	return shared.HeadersStatusContinue
+}
+
 type PassthroughConfigFactory struct {
 	shared.EmptyHttpFilterConfigFactory
 }

test/extensions/dynamic_modules/test_data/rust/http_integration_test.rs

@@ -25,6 +25,7 @@ fn new_http_filter_config_fn<EC: EnvoyHttpFilterConfig, EHF: EnvoyHttpFilter>(
 ) -> Option<Box<dyn HttpFilterConfig<EHF>>> {
   match name {
     "passthrough" => Some(Box::new(PassthroughHttpFilterConfig {})),
+    "local_reply_response_headers" => Some(Box::new(LocalReplyResponseHeadersConfig {})),
     "header_callbacks" => Some(Box::new(HeadersHttpFilterConfig {
       headers_to_add: String::from_utf8(config.to_owned()).unwrap(),
     })),
@@ -477,6 +478,30 @@ impl<EHF: EnvoyHttpFilter> HttpFilter<EHF> for ConfigStreamFilter {
   }
 }
 
+// Only records that its response-headers callback ran. Used to check that the
+// callback still fires when the response is a local reply the module did not
+// send, such as a `direct_response` route.
+struct LocalReplyResponseHeadersConfig {}
+
+impl<EHF: EnvoyHttpFilter> HttpFilterConfig<EHF> for LocalReplyResponseHeadersConfig {
+  fn new_http_filter(&self, _envoy: &mut EHF) -> Box<dyn HttpFilter<EHF>> {
+    Box::new(LocalReplyResponseHeadersFilter {})
+  }
+}
+
+struct LocalReplyResponseHeadersFilter {}
+
+impl<EHF: EnvoyHttpFilter> HttpFilter<EHF> for LocalReplyResponseHeadersFilter {
+  fn on_response_headers(
+    &self,
+    envoy_filter: &mut EHF,
+    _end_of_stream: bool,
+  ) -> envoy_dynamic_module_type_on_http_filter_response_headers_status {
+    envoy_filter.set_response_header("on-response-headers", b"called");
+    envoy_dynamic_module_type_on_http_filter_response_headers_status::Continue
+  }
+}
+
 struct PassthroughHttpFilterConfig {}
 
 impl<EHF: EnvoyHttpFilter> HttpFilterConfig<EHF> for PassthroughHttpFilterConfig {