diff --git a/include/envoy/router/router.h b/include/envoy/router/router.h
index ddc1820240f7..91d8feb4267e 100644
--- a/include/envoy/router/router.h
+++ b/include/envoy/router/router.h
@@ -14,6 +14,7 @@
 #include "envoy/config/core/v3/base.pb.h"
 #include "envoy/config/route/v3/route_components.pb.h"
 #include "envoy/config/typed_metadata.h"
+#include "envoy/event/deferred_deletable.h"
 #include "envoy/http/codec.h"
 #include "envoy/http/codes.h"
 #include "envoy/http/conn_pool.h"
@@ -1254,9 +1255,8 @@ class GenericConnectionPoolCallbacks {
  *
  * It is similar logically to RequestEncoder, only without the getStream interface.
  */
-class GenericUpstream {
+class GenericUpstream : public Event::DeferredDeletable {
 public:
-  virtual ~GenericUpstream() = default;
   /**
    * Encode a data frame.
    * @param data supplies the data to encode. The data may be moved by the encoder.
diff --git a/source/common/http/status.cc b/source/common/http/status.cc
index 1204a9ac9e4f..af8d9ecd90f7 100644
--- a/source/common/http/status.cc
+++ b/source/common/http/status.cc
@@ -25,6 +25,8 @@ absl::string_view statusCodeToString(StatusCode code) {
     return "CodecClientError";
   case StatusCode::InboundFramesWithEmptyPayload:
     return "InboundFramesWithEmptyPayloadError";
+  case StatusCode::StreamAlreadyReset:
+    return "StreamAlreadyReset";
   }
   NOT_REACHED_GCOVR_EXCL_LINE;
 }
@@ -113,6 +115,13 @@ Status inboundFramesWithEmptyPayloadError() {
   return status;
 }
 
+Status streamAlreadyReset() {
+  absl::Status status(absl::StatusCode::kInternal,
+                      "Attempted to proxy headers after stream has been reset.");
+  storePayload(status, EnvoyStatusPayload(StatusCode::StreamAlreadyReset));
+  return status;
+}
+
 // Methods for checking and extracting error information
 StatusCode getStatusCode(const Status& status) {
   return status.ok() ? StatusCode::Ok : getPayload(status).status_code_;
diff --git a/source/common/http/status.h b/source/common/http/status.h
index 97c7dac96ca1..6bb73851b9f2 100644
--- a/source/common/http/status.h
+++ b/source/common/http/status.h
@@ -74,6 +74,11 @@ enum class StatusCode : int {
    * Indicates that peer sent too many consecutive DATA frames with empty payload.
    */
   InboundFramesWithEmptyPayload = 5,
+
+  /**
+   * Indicates that we attempted to proxy upstream headers to a stream that has already been reset.
+   */
+  StreamAlreadyReset = 6,
 };
 
 using Status = absl::Status;
@@ -94,6 +99,7 @@ Status bufferFloodError(absl::string_view message);
 Status prematureResponseError(absl::string_view message, Http::Code http_code);
 Status codecClientError(absl::string_view message);
 Status inboundFramesWithEmptyPayloadError();
+Status streamAlreadyReset();
 
 /**
  * Returns Envoy::StatusCode of the given status object.
@@ -109,6 +115,7 @@ ABSL_MUST_USE_RESULT bool isBufferFloodError(const Status& status);
 ABSL_MUST_USE_RESULT bool isPrematureResponseError(const Status& status);
 ABSL_MUST_USE_RESULT bool isCodecClientError(const Status& status);
 ABSL_MUST_USE_RESULT bool isInboundFramesWithEmptyPayloadError(const Status& status);
+ABSL_MUST_USE_RESULT bool isStreamAlreadyReset(const Status& status);
 
 /**
  * Returns Http::Code value of the PrematureResponseError status.
diff --git a/source/common/router/upstream_request.cc b/source/common/router/upstream_request.cc
index bf2536449df5..f44825323bf3 100644
--- a/source/common/router/upstream_request.cc
+++ b/source/common/router/upstream_request.cc
@@ -481,8 +481,8 @@ void UpstreamRequest::clearRequestEncoder() {
   // Before clearing the encoder, unsubscribe from callbacks.
   if (upstream_) {
     parent_.callbacks()->removeDownstreamWatermarkCallbacks(downstream_watermark_manager_);
+    parent_.callbacks()->dispatcher().deferredDelete(std::move(upstream_));
   }
-  upstream_.reset();
 }
 
 void UpstreamRequest::DownstreamWatermarkManager::onAboveWriteBufferHighWatermark() {
diff --git a/source/extensions/upstreams/http/tcp/upstream_request.cc b/source/extensions/upstreams/http/tcp/upstream_request.cc
index bc2f1b76f4b2..8d0d4b25a4b5 100644
--- a/source/extensions/upstreams/http/tcp/upstream_request.cc
+++ b/source/extensions/upstreams/http/tcp/upstream_request.cc
@@ -11,6 +11,7 @@
 #include "common/http/header_map_impl.h"
 #include "common/http/headers.h"
 #include "common/http/message_impl.h"
+#include "common/http/status.h"
 #include "common/network/transport_socket_options_impl.h"
 #include "common/router/router.h"
 
@@ -45,6 +46,10 @@ void TcpUpstream::encodeData(Buffer::Instance& data, bool end_stream) {
 
 Envoy::Http::Status TcpUpstream::encodeHeaders(const Envoy::Http::RequestHeaderMap&,
                                                bool end_stream) {
+  if (!upstream_request_) {
+    return Envoy::Http::streamAlreadyReset();
+  }
+
   // Headers should only happen once, so use this opportunity to add the proxy
   // proto header, if configured.
   ASSERT(upstream_request_->routeEntry().connectConfig().has_value());
@@ -86,7 +91,16 @@ void TcpUpstream::resetStream() {
 }
 
 void TcpUpstream::onUpstreamData(Buffer::Instance& data, bool end_stream) {
+  if (!upstream_request_) {
+    return;
+  }
+
   upstream_request_->decodeData(data, end_stream);
+  // This ensures that if we get a reset after end_stream we won't propagate two
+  // "end streams" to the upstream_request_.
+  if (end_stream) {
+    upstream_request_ = nullptr;
+  }
 }
 
 void TcpUpstream::onEvent(Network::ConnectionEvent event) {
diff --git a/test/common/router/router_test.cc b/test/common/router/router_test.cc
index 4596d81a8f29..a710720e450b 100644
--- a/test/common/router/router_test.cc
+++ b/test/common/router/router_test.cc
@@ -115,6 +115,10 @@ class RouterTestBase : public testing::Test {
     // Allow any number of setTrackedObject calls for the dispatcher strict mock.
     EXPECT_CALL(callbacks_.dispatcher_, setTrackedObject(_)).Times(AnyNumber());
   }
+  ~RouterTestBase() override {
+    EXPECT_CALL(callbacks_.dispatcher_, clearDeferredDeleteList());
+    callbacks_.dispatcher_.clearDeferredDeleteList();
+  }
 
   void expectResponseTimerCreate() {
     response_timeout_ = new Event::MockTimer(&callbacks_.dispatcher_);
@@ -272,6 +276,7 @@ class RouterTestBase : public testing::Test {
         .WillOnce(Invoke([expected_count](Http::ResponseHeaderMap& headers, bool) {
           EXPECT_EQ(expected_count, atoi(std::string(headers.getEnvoyAttemptCountValue()).c_str()));
         }));
+    EXPECT_CALL(callbacks_.dispatcher_, deferredDelete_(_));
     response_decoder->decodeHeaders(std::move(response_headers), true);
     EXPECT_TRUE(verifyHostUpstreamStats(1, 0));
     EXPECT_EQ(1U,
@@ -492,6 +497,7 @@ TEST_F(RouterTest, MissingRequiredHeaders) {
                              "filter_removed_required_headers{missing required header: :method}"))
       .WillOnce(testing::InvokeWithoutArgs([] {}));
   router_.decodeHeaders(headers, true);
+  EXPECT_CALL(callbacks_.dispatcher_, deferredDelete_(_));
   router_.onDestroy();
 }
 
@@ -709,6 +715,7 @@ TEST_F(RouterTest, AddCookie) {
 
   Http::ResponseHeaderMapPtr response_headers(
       new Http::TestResponseHeaderMapImpl{{":status", "200"}});
+  EXPECT_CALL(callbacks_.dispatcher_, deferredDelete_(_));
   response_decoder->decodeHeaders(std::move(response_headers), true);
   EXPECT_EQ(callbacks_.details(), "via_upstream");
   // When the router filter gets reset we should cancel the pool request.
@@ -762,6 +769,7 @@ TEST_F(RouterTest, AddCookieNoDuplicate) {
 
   Http::ResponseHeaderMapPtr response_headers(
       new Http::TestResponseHeaderMapImpl{{":status", "200"}, {"set-cookie", "foo=baz"}});
+  EXPECT_CALL(callbacks_.dispatcher_, deferredDelete_(_));
   response_decoder->decodeHeaders(std::move(response_headers), true);
   // When the router filter gets reset we should cancel the pool request.
   router_.onDestroy();
@@ -822,6 +830,7 @@ TEST_F(RouterTest, AddMultipleCookies) {
 
   Http::ResponseHeaderMapPtr response_headers(
       new Http::TestResponseHeaderMapImpl{{":status", "200"}});
+  EXPECT_CALL(callbacks_.dispatcher_, deferredDelete_(_));
   response_decoder->decodeHeaders(std::move(response_headers), true);
   router_.onDestroy();
 }
@@ -981,6 +990,7 @@ TEST_F(RouterTest, ResponseCodeDetailsSetByUpstream) {
 
   Http::ResponseHeaderMapPtr response_headers(
       new Http::TestResponseHeaderMapImpl{{":status", "200"}});
+  EXPECT_CALL(callbacks_.dispatcher_, deferredDelete_(_));
   response_decoder->decodeHeaders(std::move(response_headers), true);
   EXPECT_TRUE(verifyHostUpstreamStats(1, 0));
 }
@@ -1012,6 +1022,7 @@ TEST_F(RouterTest, EnvoyUpstreamServiceTime) {
       .WillOnce(Invoke([](Http::HeaderMap& headers, bool) {
         EXPECT_FALSE(headers.get(Http::Headers::get().EnvoyUpstreamServiceTime).empty());
       }));
+  EXPECT_CALL(callbacks_.dispatcher_, deferredDelete_(_));
   response_decoder->decodeHeaders(std::move(response_headers), true);
   EXPECT_TRUE(verifyHostUpstreamStats(1, 0));
 }
@@ -1072,6 +1083,7 @@ TEST_F(RouterTest, EnvoyAttemptCountInRequestUpdatedInRetries) {
   Http::ResponseHeaderMapPtr response_headers1(
       new Http::TestResponseHeaderMapImpl{{":status", "503"}});
   EXPECT_CALL(cm_.conn_pool_.host_->outlier_detector_, putHttpResponseCode(503));
+  EXPECT_CALL(callbacks_.dispatcher_, deferredDelete_(_));
   response_decoder->decodeHeaders(std::move(response_headers1), true);
   EXPECT_TRUE(verifyHostUpstreamStats(0, 1));
 
@@ -1100,6 +1112,7 @@ TEST_F(RouterTest, EnvoyAttemptCountInRequestUpdatedInRetries) {
   Http::ResponseHeaderMapPtr response_headers2(
       new Http::TestResponseHeaderMapImpl{{":status", "200"}});
   EXPECT_CALL(cm_.conn_pool_.host_->outlier_detector_, putHttpResponseCode(200));
+  EXPECT_CALL(callbacks_.dispatcher_, deferredDelete_(_));
   response_decoder->decodeHeaders(std::move(response_headers2), true);
   EXPECT_TRUE(verifyHostUpstreamStats(1, 1));
 }
@@ -1203,6 +1216,7 @@ TEST_F(RouterTest, EnvoyAttemptCountInResponseWithRetries) {
 
   Http::TestRequestHeaderMapImpl headers{{"x-envoy-retry-on", "5xx"}, {"x-envoy-internal", "true"}};
   HttpTestUtility::addDefaultHeaders(headers);
+  EXPECT_CALL(callbacks_.dispatcher_, deferredDelete_(_));
   router_.decodeHeaders(headers, true);
   EXPECT_EQ(1U,
             callbacks_.route_->route_entry_.virtual_cluster_.stats().upstream_rq_total_.value());
@@ -1242,6 +1256,7 @@ TEST_F(RouterTest, EnvoyAttemptCountInResponseWithRetries) {
         // Because a retry happened the number of attempts in the response headers should be 2.
         EXPECT_EQ(2, atoi(std::string(headers.getEnvoyAttemptCountValue()).c_str()));
       }));
+  EXPECT_CALL(callbacks_.dispatcher_, deferredDelete_(_));
   response_decoder->decodeHeaders(std::move(response_headers2), true);
   EXPECT_TRUE(verifyHostUpstreamStats(1, 1));
 }
@@ -1289,6 +1304,7 @@ void RouterTestBase::testAppendCluster(absl::optional<Http::LowerCaseString> clu
         EXPECT_FALSE(cluster_header.empty());
         EXPECT_EQ("fake_cluster", cluster_header[0]->value().getStringView());
       }));
+  EXPECT_CALL(callbacks_.dispatcher_, deferredDelete_(_));
   response_decoder->decodeHeaders(std::move(response_headers), true);
   EXPECT_TRUE(verifyHostUpstreamStats(1, 0));
 }
@@ -1353,6 +1369,7 @@ void RouterTestBase::testAppendUpstreamHost(
         EXPECT_FALSE(host_address_header.empty());
         EXPECT_EQ("10.0.0.5:9211", host_address_header[0]->value().getStringView());
       }));
+  EXPECT_CALL(callbacks_.dispatcher_, deferredDelete_(_));
   response_decoder->decodeHeaders(std::move(response_headers), true);
   EXPECT_TRUE(verifyHostUpstreamStats(1, 0));
 }
@@ -1466,6 +1483,7 @@ TEST_F(RouterTestSuppressEnvoyHeaders, EnvoyUpstreamServiceTime) {
 
   Http::TestRequestHeaderMapImpl headers;
   HttpTestUtility::addDefaultHeaders(headers);
+  EXPECT_CALL(callbacks_.dispatcher_, deferredDelete_(_));
   router_.decodeHeaders(headers, true);
   EXPECT_EQ(1U,
             callbacks_.route_->route_entry_.virtual_cluster_.stats().upstream_rq_total_.value());
@@ -1508,6 +1526,7 @@ TEST_F(RouterTest, NoRetriesOverflow) {
   Http::ResponseHeaderMapPtr response_headers1(
       new Http::TestResponseHeaderMapImpl{{":status", "503"}});
   EXPECT_CALL(cm_.conn_pool_.host_->outlier_detector_, putHttpResponseCode(503));
+  EXPECT_CALL(callbacks_.dispatcher_, deferredDelete_(_));
   response_decoder->decodeHeaders(std::move(response_headers1), true);
   EXPECT_TRUE(verifyHostUpstreamStats(0, 1));
 
@@ -1535,6 +1554,7 @@ TEST_F(RouterTest, NoRetriesOverflow) {
   Http::ResponseHeaderMapPtr response_headers2(
       new Http::TestResponseHeaderMapImpl{{":status", "503"}});
   EXPECT_CALL(cm_.conn_pool_.host_->outlier_detector_, putHttpResponseCode(503));
+  EXPECT_CALL(callbacks_.dispatcher_, deferredDelete_(_));
   response_decoder->decodeHeaders(std::move(response_headers2), true);
   EXPECT_TRUE(verifyHostUpstreamStats(0, 2));
 }
@@ -1568,6 +1588,7 @@ TEST_F(RouterTest, ResetDuringEncodeHeaders) {
                         absl::optional<uint64_t>(absl::nullopt)));
   EXPECT_CALL(cm_.conn_pool_.host_->outlier_detector_,
               putResult(Upstream::Outlier::Result::LocalOriginConnectFailed, _));
+  EXPECT_CALL(callbacks_.dispatcher_, deferredDelete_(_));
   router_.decodeHeaders(headers, true);
   EXPECT_EQ(1U,
             callbacks_.route_->route_entry_.virtual_cluster_.stats().upstream_rq_total_.value());
@@ -1611,6 +1632,7 @@ TEST_F(RouterTest, UpstreamTimeout) {
   EXPECT_CALL(*router_.retry_state_, shouldRetryReset(_, _)).Times(0);
   EXPECT_CALL(cm_.conn_pool_.host_->outlier_detector_,
               putResult(Upstream::Outlier::Result::LocalOriginTimeout, _));
+  EXPECT_CALL(callbacks_.dispatcher_, deferredDelete_(_));
   response_timeout_->invokeCallback();
 
   EXPECT_EQ(1U,
@@ -1663,6 +1685,7 @@ TEST_F(RouterTest, TimeoutBudgetHistogramStat) {
       new Http::TestResponseHeaderMapImpl{{":status", "200"}});
   response_decoder->decodeHeaders(std::move(response_headers), false);
   test_time_.advanceTimeWait(std::chrono::milliseconds(80));
+  EXPECT_CALL(callbacks_.dispatcher_, deferredDelete_(_));
   response_decoder->decodeData(data, true);
 }
 
@@ -1707,6 +1730,7 @@ TEST_F(RouterTest, TimeoutBudgetHistogramStatFailure) {
       new Http::TestResponseHeaderMapImpl{{":status", "500"}});
   response_decoder->decodeHeaders(std::move(response_headers), false);
   test_time_.advanceTimeWait(std::chrono::milliseconds(80));
+  EXPECT_CALL(callbacks_.dispatcher_, deferredDelete_(_));
   response_decoder->decodeData(data, true);
 }
 
@@ -1748,6 +1772,7 @@ TEST_F(RouterTest, TimeoutBudgetHistogramStatOnlyGlobal) {
       new Http::TestResponseHeaderMapImpl{{":status", "200"}});
   response_decoder->decodeHeaders(std::move(response_headers), false);
   test_time_.advanceTimeWait(std::chrono::milliseconds(80));
+  EXPECT_CALL(callbacks_.dispatcher_, deferredDelete_(_));
   response_decoder->decodeData(data, true);
 }
 
@@ -1794,6 +1819,7 @@ TEST_F(RouterTest, TimeoutBudgetHistogramStatDuringRetries) {
   Http::ResponseHeaderMapPtr response_headers1(
       new Http::TestResponseHeaderMapImpl{{":status", "504"}});
   EXPECT_CALL(cm_.conn_pool_.host_->outlier_detector_, putHttpResponseCode(504));
+  EXPECT_CALL(callbacks_.dispatcher_, deferredDelete_(_)).Times(2);
   response_decoder1->decodeHeaders(std::move(response_headers1), true);
   EXPECT_TRUE(verifyHostUpstreamStats(0, 1));
 
@@ -1868,6 +1894,7 @@ TEST_F(RouterTest, TimeoutBudgetHistogramStatDuringGlobalTimeout) {
                                          {"x-envoy-upstream-rq-timeout-ms", "400"},
                                          {"x-envoy-upstream-rq-per-try-timeout-ms", "320"}};
   HttpTestUtility::addDefaultHeaders(headers);
+  EXPECT_CALL(callbacks_.dispatcher_, deferredDelete_(_)).Times(2);
   router_.decodeHeaders(headers, false);
   Buffer::OwnedImpl data;
   router_.decodeData(data, true);
@@ -1963,6 +1990,7 @@ TEST_F(RouterTest, GrpcOkTrailersOnly) {
   Http::TestRequestHeaderMapImpl headers{{"content-type", "application/grpc"},
                                          {"grpc-timeout", "20S"}};
   HttpTestUtility::addDefaultHeaders(headers);
+  EXPECT_CALL(callbacks_.dispatcher_, deferredDelete_(_));
   router_.decodeHeaders(headers, true);
   EXPECT_EQ(1U,
             callbacks_.route_->route_entry_.virtual_cluster_.stats().upstream_rq_total_.value());
@@ -1992,6 +2020,7 @@ TEST_F(RouterTest, GrpcAlreadyExistsTrailersOnly) {
   Http::TestRequestHeaderMapImpl headers{{"content-type", "application/grpc"},
                                          {"grpc-timeout", "20S"}};
   HttpTestUtility::addDefaultHeaders(headers);
+  EXPECT_CALL(callbacks_.dispatcher_, deferredDelete_(_));
   router_.decodeHeaders(headers, true);
   EXPECT_EQ(1U,
             callbacks_.route_->route_entry_.virtual_cluster_.stats().upstream_rq_total_.value());
@@ -2021,6 +2050,7 @@ TEST_F(RouterTest, GrpcOutlierDetectionUnavailableStatusCode) {
   Http::TestRequestHeaderMapImpl headers{{"content-type", "application/grpc"},
                                          {"grpc-timeout", "20S"}};
   HttpTestUtility::addDefaultHeaders(headers);
+  EXPECT_CALL(callbacks_.dispatcher_, deferredDelete_(_));
   router_.decodeHeaders(headers, true);
   EXPECT_EQ(1U,
             callbacks_.route_->route_entry_.virtual_cluster_.stats().upstream_rq_total_.value());
@@ -2058,6 +2088,7 @@ TEST_F(RouterTest, GrpcInternalTrailersOnly) {
   Http::ResponseHeaderMapPtr response_headers(
       new Http::TestResponseHeaderMapImpl{{":status", "200"}, {"grpc-status", "13"}});
   EXPECT_CALL(cm_.conn_pool_.host_->outlier_detector_, putHttpResponseCode(500));
+  EXPECT_CALL(callbacks_.dispatcher_, deferredDelete_(_));
   response_decoder->decodeHeaders(std::move(response_headers), true);
   EXPECT_TRUE(verifyHostUpstreamStats(0, 1));
 }
@@ -2081,6 +2112,7 @@ TEST_F(RouterTest, GrpcDataEndStream) {
   Http::TestRequestHeaderMapImpl headers{{"content-type", "application/grpc"},
                                          {"grpc-timeout", "20S"}};
   HttpTestUtility::addDefaultHeaders(headers);
+  EXPECT_CALL(callbacks_.dispatcher_, deferredDelete_(_));
   router_.decodeHeaders(headers, true);
   EXPECT_EQ(1U,
             callbacks_.route_->route_entry_.virtual_cluster_.stats().upstream_rq_total_.value());
@@ -2114,6 +2146,7 @@ TEST_F(RouterTest, GrpcReset) {
   Http::TestRequestHeaderMapImpl headers{{"content-type", "application/grpc"},
                                          {"grpc-timeout", "20S"}};
   HttpTestUtility::addDefaultHeaders(headers);
+  EXPECT_CALL(callbacks_.dispatcher_, deferredDelete_(_));
   router_.decodeHeaders(headers, true);
   EXPECT_EQ(1U,
             callbacks_.route_->route_entry_.virtual_cluster_.stats().upstream_rq_total_.value());
@@ -2148,6 +2181,7 @@ TEST_F(RouterTest, GrpcOk) {
   Http::TestRequestHeaderMapImpl headers{{"content-type", "application/grpc"},
                                          {"grpc-timeout", "20S"}};
   HttpTestUtility::addDefaultHeaders(headers);
+  EXPECT_CALL(callbacks_.dispatcher_, deferredDelete_(_));
   router_.decodeHeaders(headers, true);
   EXPECT_EQ(1U,
             callbacks_.route_->route_entry_.virtual_cluster_.stats().upstream_rq_total_.value());
@@ -2184,6 +2218,7 @@ TEST_F(RouterTest, GrpcInternal) {
   Http::TestRequestHeaderMapImpl headers{{"content-type", "application/grpc"},
                                          {"grpc-timeout", "20S"}};
   HttpTestUtility::addDefaultHeaders(headers);
+  EXPECT_CALL(callbacks_.dispatcher_, deferredDelete_(_));
   router_.decodeHeaders(headers, true);
   EXPECT_EQ(1U,
             callbacks_.route_->route_entry_.virtual_cluster_.stats().upstream_rq_total_.value());
@@ -2236,6 +2271,7 @@ TEST_F(RouterTest, UpstreamTimeoutWithAltResponse) {
   EXPECT_CALL(
       cm_.conn_pool_.host_->outlier_detector_,
       putResult(Upstream::Outlier::Result::LocalOriginTimeout, absl::optional<uint64_t>(204)));
+  EXPECT_CALL(callbacks_.dispatcher_, deferredDelete_(_));
   response_timeout_->invokeCallback();
 
   EXPECT_EQ(1U,
@@ -2288,6 +2324,7 @@ TEST_F(RouterTest, UpstreamPerTryTimeout) {
   EXPECT_CALL(
       cm_.conn_pool_.host_->outlier_detector_,
       putResult(Upstream::Outlier::Result::LocalOriginTimeout, absl::optional<uint64_t>(504)));
+  EXPECT_CALL(callbacks_.dispatcher_, deferredDelete_(_));
   per_try_timeout_->invokeCallback();
 
   EXPECT_EQ(1U, cm_.thread_local_cluster_.cluster_.info_->stats_store_
@@ -2345,6 +2382,7 @@ TEST_F(RouterTest, UpstreamPerTryTimeoutDelayedPoolReady) {
   EXPECT_CALL(callbacks_, encodeData(_, true));
   EXPECT_CALL(cm_.conn_pool_.host_->outlier_detector_,
               putResult(Upstream::Outlier::Result::LocalOriginTimeout, _));
+  EXPECT_CALL(callbacks_.dispatcher_, deferredDelete_(_));
   per_try_timeout_->invokeCallback();
 
   EXPECT_EQ(1U, cm_.thread_local_cluster_.cluster_.info_->stats_store_
@@ -2396,6 +2434,7 @@ TEST_F(RouterTest, UpstreamPerTryTimeoutExcludesNewStream) {
             callbacks_.route_->route_entry_.virtual_cluster_.stats().upstream_rq_total_.value());
 
   EXPECT_CALL(encoder.stream_, resetStream(Http::StreamResetReason::LocalReset));
+  EXPECT_CALL(callbacks_.dispatcher_, deferredDelete_(_));
   EXPECT_CALL(cm_.conn_pool_.host_->outlier_detector_,
               putResult(Upstream::Outlier::Result::LocalOriginTimeout, _));
   EXPECT_CALL(*per_try_timeout_, disableTimer());
@@ -2487,6 +2526,7 @@ TEST_F(RouterTest, HedgedPerTryTimeoutFirstRequestSucceeds) {
         EXPECT_EQ(headers.Status()->value(), "200");
         EXPECT_TRUE(end_stream);
       }));
+  EXPECT_CALL(callbacks_.dispatcher_, deferredDelete_(_)).Times(2);
   response_decoder1->decodeHeaders(std::move(response_headers), true);
   EXPECT_TRUE(verifyHostUpstreamStats(1, 0));
 
@@ -2524,6 +2564,7 @@ TEST_F(RouterTest, HedgedPerTryTimeoutResetsOnBadHeaders) {
 
   Http::TestRequestHeaderMapImpl headers{{"x-envoy-upstream-rq-per-try-timeout-ms", "5"}};
   HttpTestUtility::addDefaultHeaders(headers);
+  EXPECT_CALL(callbacks_.dispatcher_, deferredDelete_(_)).Times(2);
   router_.decodeHeaders(headers, true);
   EXPECT_EQ(1U,
             callbacks_.route_->route_entry_.virtual_cluster_.stats().upstream_rq_total_.value());
@@ -2608,6 +2649,7 @@ TEST_F(RouterTest, HedgedPerTryTimeoutThirdRequestSucceeds) {
 
   Http::TestRequestHeaderMapImpl headers{{"x-envoy-upstream-rq-per-try-timeout-ms", "5"}};
   HttpTestUtility::addDefaultHeaders(headers);
+  EXPECT_CALL(callbacks_.dispatcher_, deferredDelete_(_)).Times(3);
   router_.decodeHeaders(headers, true);
   EXPECT_EQ(1U,
             callbacks_.route_->route_entry_.virtual_cluster_.stats().upstream_rq_total_.value());
@@ -2727,6 +2769,7 @@ TEST_F(RouterTest, RetryOnlyOnceForSameUpstreamRequest) {
       cm_.conn_pool_.host_->outlier_detector_,
       putResult(Upstream::Outlier::Result::LocalOriginTimeout, absl::optional<uint64_t>(504)));
   router_.retry_state_->expectHedgedPerTryTimeoutRetry();
+  EXPECT_CALL(callbacks_.dispatcher_, deferredDelete_(_)).Times(2);
   per_try_timeout_->invokeCallback();
 
   NiceMock<Http::MockRequestEncoder> encoder2;
@@ -2788,6 +2831,7 @@ TEST_F(RouterTest, BadHeadersDroppedIfPreviousRetryScheduled) {
 
   Http::TestRequestHeaderMapImpl headers{{"x-envoy-upstream-rq-per-try-timeout-ms", "5"}};
   HttpTestUtility::addDefaultHeaders(headers);
+  EXPECT_CALL(callbacks_.dispatcher_, deferredDelete_(_)).Times(2);
   router_.decodeHeaders(headers, true);
 
   EXPECT_CALL(encoder1.stream_, resetStream(_)).Times(0);
@@ -2856,6 +2900,7 @@ TEST_F(RouterTest, RetryRequestBeforeBody) {
   Http::TestRequestHeaderMapImpl headers{
       {"x-envoy-retry-on", "5xx"}, {"x-envoy-internal", "true"}, {"myheader", "present"}};
   HttpTestUtility::addDefaultHeaders(headers);
+  EXPECT_CALL(callbacks_.dispatcher_, deferredDelete_(_)).Times(2);
   router_.decodeHeaders(headers, false);
 
   router_.retry_state_->expectResetRetry();
@@ -2918,6 +2963,7 @@ TEST_F(RouterTest, RetryRequestDuringBody) {
   Http::TestRequestHeaderMapImpl headers{
       {"x-envoy-retry-on", "5xx"}, {"x-envoy-internal", "true"}, {"myheader", "present"}};
   HttpTestUtility::addDefaultHeaders(headers);
+  EXPECT_CALL(callbacks_.dispatcher_, deferredDelete_(_)).Times(2);
   router_.decodeHeaders(headers, false);
   const std::string body1("body1");
   Buffer::OwnedImpl buf1(body1);
@@ -2995,6 +3041,7 @@ TEST_F(RouterTest, RetryRequestDuringBodyDataBetweenAttemptsNotEndStream) {
   router_.decodeData(buf1, false);
 
   router_.retry_state_->expectResetRetry();
+  EXPECT_CALL(callbacks_.dispatcher_, deferredDelete_(_));
   encoder1.stream_.resetStream(Http::StreamResetReason::RemoteReset);
 
   const std::string body2("body2");
@@ -3023,6 +3070,7 @@ TEST_F(RouterTest, RetryRequestDuringBodyDataBetweenAttemptsNotEndStream) {
   const std::string body3("body3");
   EXPECT_CALL(encoder2, encodeData(BufferStringEqual(body3), true));
   Buffer::OwnedImpl buf3(body3);
+  EXPECT_CALL(callbacks_.dispatcher_, deferredDelete_(_));
   router_.decodeData(buf3, true);
 
   // Send successful response, verify success.
@@ -3067,11 +3115,13 @@ TEST_F(RouterTest, RetryRequestDuringBodyCompleteBetweenAttempts) {
   router_.decodeData(buf1, false);
 
   router_.retry_state_->expectResetRetry();
+  EXPECT_CALL(callbacks_.dispatcher_, deferredDelete_(_));
   encoder1.stream_.resetStream(Http::StreamResetReason::RemoteReset);
 
   // Complete request while there is no upstream request.
   const std::string body2("body2");
   Buffer::OwnedImpl buf2(body2);
+  EXPECT_CALL(callbacks_.dispatcher_, deferredDelete_(_));
   router_.decodeData(buf2, true);
 
   NiceMock<Http::MockRequestEncoder> encoder2;
@@ -3134,10 +3184,12 @@ TEST_F(RouterTest, RetryRequestDuringBodyTrailerBetweenAttempts) {
   router_.decodeData(buf1, false);
 
   router_.retry_state_->expectResetRetry();
+  EXPECT_CALL(callbacks_.dispatcher_, deferredDelete_(_));
   encoder1.stream_.resetStream(Http::StreamResetReason::RemoteReset);
 
   // Complete request while there is no upstream request.
   Http::TestRequestTrailerMapImpl trailers{{"some", "trailer"}};
+  EXPECT_CALL(callbacks_.dispatcher_, deferredDelete_(_));
   router_.decodeTrailers(trailers);
 
   NiceMock<Http::MockRequestEncoder> encoder2;
@@ -3202,6 +3254,7 @@ TEST_F(RouterTest, RetryRequestDuringBodyBufferLimitExceeded) {
   router_.decodeData(buf1, false);
 
   router_.retry_state_->expectResetRetry();
+  EXPECT_CALL(callbacks_.dispatcher_, deferredDelete_(_));
   encoder1.stream_.resetStream(Http::StreamResetReason::RemoteReset);
 
   // Complete request while there is no upstream request.
@@ -3284,6 +3337,7 @@ TEST_F(RouterTest, HedgedPerTryTimeoutGlobalTimeout) {
       .WillOnce(Invoke([&](Http::ResponseHeaderMap& headers, bool) -> void {
         EXPECT_EQ(headers.Status()->value(), "504");
       }));
+  EXPECT_CALL(callbacks_.dispatcher_, deferredDelete_(_)).Times(2);
   response_timeout_->invokeCallback();
   EXPECT_TRUE(verifyHostUpstreamStats(0, 2));
   EXPECT_EQ(2, cm_.conn_pool_.host_->stats_.rq_timeout_.value());
@@ -3316,6 +3370,7 @@ TEST_F(RouterTest, HedgingRetriesExhaustedBadResponse) {
 
   Http::TestRequestHeaderMapImpl headers{{"x-envoy-upstream-rq-per-try-timeout-ms", "5"}};
   HttpTestUtility::addDefaultHeaders(headers);
+  EXPECT_CALL(callbacks_.dispatcher_, deferredDelete_(_)).Times(2);
   router_.decodeHeaders(headers, true);
   EXPECT_EQ(1U,
             callbacks_.route_->route_entry_.virtual_cluster_.stats().upstream_rq_total_.value());
@@ -3410,6 +3465,7 @@ TEST_F(RouterTest, HedgingRetriesProceedAfterReset) {
 
   Http::TestRequestHeaderMapImpl headers{{"x-envoy-upstream-rq-per-try-timeout-ms", "5"}};
   HttpTestUtility::addDefaultHeaders(headers);
+  EXPECT_CALL(callbacks_.dispatcher_, deferredDelete_(_)).Times(2);
   router_.decodeHeaders(headers, true);
   EXPECT_EQ(1U,
             callbacks_.route_->route_entry_.virtual_cluster_.stats().upstream_rq_total_.value());
@@ -3492,6 +3548,7 @@ TEST_F(RouterTest, HedgingRetryImmediatelyReset) {
 
   Http::TestRequestHeaderMapImpl headers{{"x-envoy-upstream-rq-per-try-timeout-ms", "5"}};
   HttpTestUtility::addDefaultHeaders(headers);
+  EXPECT_CALL(callbacks_.dispatcher_, deferredDelete_(_));
   router_.decodeHeaders(headers, false);
 
   expectPerTryTimerCreate();
@@ -3573,6 +3630,7 @@ TEST_F(RouterTest, RetryNoneHealthy) {
   router_.retry_state_->expectResetRetry();
   EXPECT_CALL(cm_.conn_pool_.host_->outlier_detector_,
               putResult(Upstream::Outlier::Result::LocalOriginConnectFailed, _));
+  EXPECT_CALL(callbacks_.dispatcher_, deferredDelete_(_));
   encoder1.stream_.resetStream(Http::StreamResetReason::LocalReset);
 
   EXPECT_CALL(cm_, httpConnPoolForCluster(_, _, _, _)).WillOnce(Return(nullptr));
@@ -3616,6 +3674,7 @@ TEST_F(RouterTest, RetryUpstreamReset) {
   router_.retry_state_->expectResetRetry();
   EXPECT_CALL(cm_.conn_pool_.host_->outlier_detector_,
               putResult(Upstream::Outlier::Result::LocalOriginConnectFailed, _));
+  EXPECT_CALL(callbacks_.dispatcher_, deferredDelete_(_)).Times(2);
   encoder1.stream_.resetStream(Http::StreamResetReason::RemoteReset);
 
   // We expect this reset to kick off a new request.
@@ -3663,6 +3722,7 @@ TEST_F(RouterTest, NoRetryWithBodyLimit) {
   EXPECT_CALL(callbacks_.route_->route_entry_, retryShadowBufferLimit()).WillOnce(Return(0));
   Http::TestRequestHeaderMapImpl headers{{"x-envoy-retry-on", "5xx"}, {"x-envoy-internal", "true"}};
   HttpTestUtility::addDefaultHeaders(headers);
+  EXPECT_CALL(callbacks_.dispatcher_, deferredDelete_(_));
   router_.decodeHeaders(headers, false);
   // Unlike RetryUpstreamReset above the data won't be buffered as the body exceeds the buffer limit
   EXPECT_CALL(*router_.retry_state_, enabled()).WillOnce(Return(true));
@@ -3700,6 +3760,7 @@ TEST_F(RouterTest, RetryUpstreamPerTryTimeout) {
                                          {"x-envoy-internal", "true"},
                                          {"x-envoy-upstream-rq-per-try-timeout-ms", "5"}};
   HttpTestUtility::addDefaultHeaders(headers);
+  EXPECT_CALL(callbacks_.dispatcher_, deferredDelete_(_)).Times(2);
   router_.decodeHeaders(headers, true);
   EXPECT_EQ(1U,
             callbacks_.route_->route_entry_.virtual_cluster_.stats().upstream_rq_total_.value());
@@ -3753,6 +3814,7 @@ TEST_F(RouterTest, RetryUpstreamConnectionFailure) {
 
   Http::TestRequestHeaderMapImpl headers{{"x-envoy-retry-on", "5xx"}, {"x-envoy-internal", "true"}};
   HttpTestUtility::addDefaultHeaders(headers);
+  EXPECT_CALL(callbacks_.dispatcher_, deferredDelete_(_));
   router_.decodeHeaders(headers, true);
 
   EXPECT_CALL(*router_.retry_state_, onHostAttempted(_)).Times(0);
@@ -3809,6 +3871,7 @@ TEST_F(RouterTest, DontResetStartedResponseOnUpstreamPerTryTimeout) {
   Http::TestRequestHeaderMapImpl headers{{"x-envoy-internal", "true"},
                                          {"x-envoy-upstream-rq-per-try-timeout-ms", "5"}};
   HttpTestUtility::addDefaultHeaders(headers);
+  EXPECT_CALL(callbacks_.dispatcher_, deferredDelete_(_));
   router_.decodeHeaders(headers, true);
   EXPECT_EQ(1U,
             callbacks_.route_->route_entry_.virtual_cluster_.stats().upstream_rq_total_.value());
@@ -3866,6 +3929,7 @@ TEST_F(RouterTest, RetryUpstreamResetResponseStarted) {
   // the encoder again.
   EXPECT_CALL(callbacks_, sendLocalReply(_, _, _, _, _)).WillOnce(testing::InvokeWithoutArgs([] {
   }));
+  EXPECT_CALL(callbacks_.dispatcher_, deferredDelete_(_));
   encoder1.stream_.resetStream(Http::StreamResetReason::RemoteReset);
   // For normal HTTP, once we have a 200 we consider this a success, even if a
   // later reset occurs.
@@ -3915,6 +3979,7 @@ TEST_F(RouterTest, Coalesce100ContinueHeaders) {
         new Http::TestResponseHeaderMapImpl{{":status", "100"}});
     response_decoder->decode100ContinueHeaders(std::move(continue_headers));
   }
+
   EXPECT_EQ(
       2U,
       cm_.thread_local_cluster_.cluster_.info_->stats_store_.counter("upstream_rq_100").value());
@@ -3922,6 +3987,7 @@ TEST_F(RouterTest, Coalesce100ContinueHeaders) {
   // Reset stream and cleanup.
   EXPECT_CALL(cm_.conn_pool_.host_->outlier_detector_,
               putResult(Upstream::Outlier::Result::LocalOriginConnectFailed, _));
+  EXPECT_CALL(callbacks_.dispatcher_, deferredDelete_(_));
   encoder1.stream_.resetStream(Http::StreamResetReason::RemoteReset);
   EXPECT_EQ(1U,
             callbacks_.route_->route_entry_.virtual_cluster_.stats().upstream_rq_total_.value());
@@ -3960,6 +4026,7 @@ TEST_F(RouterTest, RetryUpstreamReset100ContinueResponseStarted) {
       cm_.thread_local_cluster_.cluster_.info_->stats_store_.counter("upstream_rq_100").value());
   EXPECT_CALL(cm_.conn_pool_.host_->outlier_detector_,
               putResult(Upstream::Outlier::Result::LocalOriginConnectFailed, _));
+  EXPECT_CALL(callbacks_.dispatcher_, deferredDelete_(_));
   encoder1.stream_.resetStream(Http::StreamResetReason::RemoteReset);
   EXPECT_EQ(1U,
             callbacks_.route_->route_entry_.virtual_cluster_.stats().upstream_rq_total_.value());
@@ -3981,6 +4048,7 @@ TEST_F(RouterTest, RetryUpstream5xx) {
 
   Http::TestRequestHeaderMapImpl headers{{"x-envoy-retry-on", "5xx"}, {"x-envoy-internal", "true"}};
   HttpTestUtility::addDefaultHeaders(headers);
+  EXPECT_CALL(callbacks_.dispatcher_, deferredDelete_(_)).Times(2);
   router_.decodeHeaders(headers, true);
   EXPECT_EQ(1U,
             callbacks_.route_->route_entry_.virtual_cluster_.stats().upstream_rq_total_.value());
@@ -4035,6 +4103,7 @@ TEST_F(RouterTest, RetryTimeoutDuringRetryDelay) {
 
   Http::TestRequestHeaderMapImpl headers{{"x-envoy-retry-on", "5xx"}, {"x-envoy-internal", "true"}};
   HttpTestUtility::addDefaultHeaders(headers);
+  EXPECT_CALL(callbacks_.dispatcher_, deferredDelete_(_));
   router_.decodeHeaders(headers, true);
   EXPECT_EQ(1U,
             callbacks_.route_->route_entry_.virtual_cluster_.stats().upstream_rq_total_.value());
@@ -4078,6 +4147,7 @@ TEST_F(RouterTest, MaxStreamDurationValidlyConfiguredWithoutRetryPolicy) {
   Http::TestRequestHeaderMapImpl headers;
   HttpTestUtility::addDefaultHeaders(headers);
   router_.decodeHeaders(headers, false);
+  EXPECT_CALL(callbacks_.dispatcher_, deferredDelete_(_));
   max_stream_duration_timer_->invokeCallback();
 
   router_.onDestroy();
@@ -4105,6 +4175,7 @@ TEST_F(RouterTest, MaxStreamDurationDisabledIfSetToZero) {
   HttpTestUtility::addDefaultHeaders(headers);
   router_.decodeHeaders(headers, false);
 
+  EXPECT_CALL(callbacks_.dispatcher_, deferredDelete_(_));
   router_.onDestroy();
   EXPECT_TRUE(verifyHostUpstreamStats(0, 0));
 }
@@ -4128,6 +4199,7 @@ TEST_F(RouterTest, MaxStreamDurationCallbackNotCalled) {
   HttpTestUtility::addDefaultHeaders(headers);
   router_.decodeHeaders(headers, false);
 
+  EXPECT_CALL(callbacks_.dispatcher_, deferredDelete_(_));
   router_.onDestroy();
   EXPECT_TRUE(verifyHostUpstreamStats(0, 0));
 }
@@ -4153,6 +4225,7 @@ TEST_F(RouterTest, MaxStreamDurationWhenDownstreamAlreadyStartedWithoutRetryPoli
   Http::ResponseHeaderMapPtr response_headers(
       new Http::TestResponseHeaderMapImpl{{":status", "200"}});
   response_decoder->decodeHeaders(std::move(response_headers), false);
+  EXPECT_CALL(callbacks_.dispatcher_, deferredDelete_(_));
   max_stream_duration_timer_->invokeCallback();
 
   router_.onDestroy();
@@ -4178,6 +4251,8 @@ TEST_F(RouterTest, MaxStreamDurationWithRetryPolicy) {
   Http::TestRequestHeaderMapImpl headers{{"x-envoy-retry-on", "reset"},
                                          {"x-envoy-internal", "true"}};
   HttpTestUtility::addDefaultHeaders(headers);
+
+  EXPECT_CALL(callbacks_.dispatcher_, deferredDelete_(_)).Times(2);
   router_.decodeHeaders(headers, false);
 
   router_.retry_state_->expectResetRetry();
@@ -4221,6 +4296,7 @@ TEST_F(RouterTest, RetryTimeoutDuringRetryDelayWithUpstreamRequestNoHost) {
 
   Http::TestRequestHeaderMapImpl headers{{"x-envoy-retry-on", "5xx"}, {"x-envoy-internal", "true"}};
   HttpTestUtility::addDefaultHeaders(headers);
+  EXPECT_CALL(callbacks_.dispatcher_, deferredDelete_(_));
   router_.decodeHeaders(headers, true);
   EXPECT_EQ(1U,
             callbacks_.route_->route_entry_.virtual_cluster_.stats().upstream_rq_total_.value());
@@ -4278,6 +4354,7 @@ TEST_F(RouterTest, RetryTimeoutDuringRetryDelayWithUpstreamRequestNoHostAltRespo
                                          {"x-envoy-internal", "true"},
                                          {"x-envoy-upstream-rq-timeout-alt-response", "204"}};
   HttpTestUtility::addDefaultHeaders(headers);
+  EXPECT_CALL(callbacks_.dispatcher_, deferredDelete_(_));
   router_.decodeHeaders(headers, true);
   EXPECT_EQ(1U,
             callbacks_.route_->route_entry_.virtual_cluster_.stats().upstream_rq_total_.value());
@@ -4338,6 +4415,7 @@ TEST_F(RouterTest, RetryUpstream5xxNotComplete) {
   EXPECT_EQ(Http::FilterDataStatus::StopIterationNoBuffer, router_.decodeData(*body_data, false));
 
   Http::TestRequestTrailerMapImpl trailers{{"some", "trailer"}};
+  EXPECT_CALL(callbacks_.dispatcher_, deferredDelete_(_)).Times(2);
   router_.decodeTrailers(trailers);
   EXPECT_EQ(1U,
             callbacks_.route_->route_entry_.virtual_cluster_.stats().upstream_rq_total_.value());
@@ -4414,6 +4492,7 @@ TEST_F(RouterTest, RetryUpstreamGrpcCancelled) {
                                          {"content-type", "application/grpc"},
                                          {"grpc-timeout", "20S"}};
   HttpTestUtility::addDefaultHeaders(headers);
+  EXPECT_CALL(callbacks_.dispatcher_, deferredDelete_(_)).Times(2);
   router_.decodeHeaders(headers, true);
   EXPECT_EQ(1U,
             callbacks_.route_->route_entry_.virtual_cluster_.stats().upstream_rq_total_.value());
@@ -4493,6 +4572,7 @@ TEST_F(RouterTest, RetryRespsectsMaxHostSelectionCount) {
       new Http::TestResponseHeaderMapImpl{{":status", "503"}});
   EXPECT_CALL(encoder1.stream_, resetStream(Http::StreamResetReason::LocalReset));
   EXPECT_CALL(cm_.conn_pool_.host_->outlier_detector_, putHttpResponseCode(503));
+  EXPECT_CALL(callbacks_.dispatcher_, deferredDelete_(_)).Times(2);
   response_decoder->decodeHeaders(std::move(response_headers1), false);
   EXPECT_TRUE(verifyHostUpstreamStats(0, 1));
 
@@ -4570,6 +4650,7 @@ TEST_F(RouterTest, RetryRespectsRetryHostPredicate) {
       new Http::TestResponseHeaderMapImpl{{":status", "503"}});
   EXPECT_CALL(encoder1.stream_, resetStream(Http::StreamResetReason::LocalReset));
   EXPECT_CALL(cm_.conn_pool_.host_->outlier_detector_, putHttpResponseCode(503));
+  EXPECT_CALL(callbacks_.dispatcher_, deferredDelete_(_)).Times(2);
   response_decoder->decodeHeaders(std::move(response_headers1), false);
   EXPECT_TRUE(verifyHostUpstreamStats(0, 1));
 
@@ -4615,6 +4696,7 @@ TEST_F(RouterTest, InternalRedirectRejectedWhenReachingMaxInternalRedirect) {
   response_decoder_->decodeHeaders(std::move(redirect_headers_), false);
 
   Buffer::OwnedImpl data("1234567890");
+  EXPECT_CALL(callbacks_.dispatcher_, deferredDelete_(_));
   response_decoder_->decodeData(data, true);
   EXPECT_EQ(1U, cm_.thread_local_cluster_.cluster_.info_->stats_store_
                     .counter("upstream_internal_redirect_failed_total")
@@ -4634,6 +4716,7 @@ TEST_F(RouterTest, InternalRedirectRejectedWithEmptyLocation) {
   response_decoder_->decodeHeaders(std::move(redirect_headers_), false);
 
   Buffer::OwnedImpl data("1234567890");
+  EXPECT_CALL(callbacks_.dispatcher_, deferredDelete_(_));
   response_decoder_->decodeData(data, true);
   EXPECT_EQ(1U, cm_.thread_local_cluster_.cluster_.info_->stats_store_
                     .counter("upstream_internal_redirect_failed_total")
@@ -4652,6 +4735,7 @@ TEST_F(RouterTest, InternalRedirectRejectedWithInvalidLocation) {
   response_decoder_->decodeHeaders(std::move(redirect_headers_), false);
 
   Buffer::OwnedImpl data("1234567890");
+  EXPECT_CALL(callbacks_.dispatcher_, deferredDelete_(_));
   response_decoder_->decodeData(data, true);
   EXPECT_EQ(1U, cm_.thread_local_cluster_.cluster_.info_->stats_store_
                     .counter("upstream_internal_redirect_failed_total")
@@ -4669,6 +4753,7 @@ TEST_F(RouterTest, InternalRedirectRejectedWithoutCompleteRequest) {
   response_decoder_->decodeHeaders(std::move(redirect_headers_), false);
 
   Buffer::OwnedImpl data("1234567890");
+  EXPECT_CALL(callbacks_.dispatcher_, deferredDelete_(_));
   response_decoder_->decodeData(data, true);
   EXPECT_EQ(1U, cm_.thread_local_cluster_.cluster_.info_->stats_store_
                     .counter("upstream_internal_redirect_failed_total")
@@ -4686,6 +4771,7 @@ TEST_F(RouterTest, InternalRedirectRejectedWithoutLocation) {
 
   response_decoder_->decodeHeaders(std::move(redirect_headers_), false);
   Buffer::OwnedImpl data("1234567890");
+  EXPECT_CALL(callbacks_.dispatcher_, deferredDelete_(_));
   response_decoder_->decodeData(data, true);
   EXPECT_EQ(1U, cm_.thread_local_cluster_.cluster_.info_->stats_store_
                     .counter("upstream_internal_redirect_failed_total")
@@ -4703,6 +4789,7 @@ TEST_F(RouterTest, InternalRedirectRejectedWithBody) {
 
   response_decoder_->decodeHeaders(std::move(redirect_headers_), false);
   Buffer::OwnedImpl data("1234567890");
+  EXPECT_CALL(callbacks_.dispatcher_, deferredDelete_(_));
   response_decoder_->decodeData(data, true);
   EXPECT_EQ(1U, cm_.thread_local_cluster_.cluster_.info_->stats_store_
                     .counter("upstream_internal_redirect_failed_total")
@@ -4712,6 +4799,7 @@ TEST_F(RouterTest, InternalRedirectRejectedWithBody) {
 TEST_F(RouterTest, CrossSchemeRedirectRejectedByPolicy) {
   enableRedirects();
 
+  EXPECT_CALL(callbacks_.dispatcher_, deferredDelete_(_));
   sendRequest();
 
   redirect_headers_->setLocation("https://www.foo.com");
@@ -4729,6 +4817,7 @@ TEST_F(RouterTest, CrossSchemeRedirectRejectedByPolicy) {
 TEST_F(RouterTest, InternalRedirectRejectedByPredicate) {
   enableRedirects();
 
+  EXPECT_CALL(callbacks_.dispatcher_, deferredDelete_(_));
   sendRequest();
 
   redirect_headers_->setLocation("http://www.foo.com/some/path");
@@ -4771,6 +4860,7 @@ TEST_F(RouterTest, HttpInternalRedirectSucceeded) {
                     .value());
 
   // In production, the HCM recreateStream would have called this.
+  EXPECT_CALL(callbacks_.dispatcher_, deferredDelete_(_));
   router_.onDestroy();
   EXPECT_EQ(3, callbacks_.streamInfo()
                    .filterState()
@@ -4796,6 +4886,7 @@ TEST_F(RouterTest, HttpsInternalRedirectSucceeded) {
                     .value());
 
   // In production, the HCM recreateStream would have called this.
+  EXPECT_CALL(callbacks_.dispatcher_, deferredDelete_(_));
   router_.onDestroy();
 }
 
@@ -4819,6 +4910,7 @@ TEST_F(RouterTest, CrossSchemeRedirectAllowedByPolicy) {
                     .value());
 
   // In production, the HCM recreateStream would have called this.
+  EXPECT_CALL(callbacks_.dispatcher_, deferredDelete_(_));
   router_.onDestroy();
 }
 
@@ -4848,6 +4940,7 @@ TEST_F(RouterTest, Shadow) {
 
   Http::TestRequestHeaderMapImpl headers;
   HttpTestUtility::addDefaultHeaders(headers);
+  EXPECT_CALL(callbacks_.dispatcher_, deferredDelete_(_));
   router_.decodeHeaders(headers, false);
 
   Buffer::InstancePtr body_data(new Buffer::OwnedImpl("hello"));
@@ -4905,6 +4998,7 @@ TEST_F(RouterTest, AltStatName) {
   Http::TestRequestHeaderMapImpl headers{{"x-envoy-upstream-alt-stat-name", "alt_stat"},
                                          {"x-envoy-internal", "true"}};
   HttpTestUtility::addDefaultHeaders(headers);
+  EXPECT_CALL(callbacks_.dispatcher_, deferredDelete_(_));
   router_.decodeHeaders(headers, true);
   EXPECT_EQ(1U,
             callbacks_.route_->route_entry_.virtual_cluster_.stats().upstream_rq_total_.value());
@@ -5118,6 +5212,7 @@ TEST_F(RouterTest, PropagatesUpstreamFilterState) {
 
   Http::TestRequestHeaderMapImpl headers{};
   HttpTestUtility::addDefaultHeaders(headers);
+  EXPECT_CALL(callbacks_.dispatcher_, deferredDelete_(_));
   router_.decodeHeaders(headers, true);
 
   Http::ResponseHeaderMapPtr response_headers(
@@ -5156,6 +5251,7 @@ TEST_F(RouterTest, UpstreamSSLConnection) {
 
   Http::ResponseHeaderMapPtr response_headers(
       new Http::TestResponseHeaderMapImpl{{":status", "200"}});
+  EXPECT_CALL(callbacks_.dispatcher_, deferredDelete_(_));
   response_decoder->decodeHeaders(std::move(response_headers), true);
   EXPECT_TRUE(verifyHostUpstreamStats(1, 0));
 
@@ -5192,6 +5288,7 @@ TEST_F(RouterTest, UpstreamTimingSingleRequest) {
 
   test_time_.advanceTimeWait(std::chrono::milliseconds(32));
   Buffer::OwnedImpl data;
+  EXPECT_CALL(callbacks_.dispatcher_, deferredDelete_(_));
   router_.decodeData(data, true);
   EXPECT_EQ(1U,
             callbacks_.route_->route_entry_.virtual_cluster_.stats().upstream_rq_total_.value());
@@ -5247,6 +5344,7 @@ TEST_F(RouterTest, UpstreamTimingRetry) {
   // Check that upstream timing is updated after the first request.
   Http::TestRequestHeaderMapImpl headers{{"x-envoy-retry-on", "5xx"}};
   HttpTestUtility::addDefaultHeaders(headers);
+  EXPECT_CALL(callbacks_.dispatcher_, deferredDelete_(_)).Times(2);
   router_.decodeHeaders(headers, false);
 
   router_.retry_state_->expectHeadersRetry();
@@ -5353,6 +5451,7 @@ TEST_F(RouterTest, UpstreamTimingTimeout) {
   response_decoder->decodeHeaders(std::move(response_headers), false);
 
   test_time_.advanceTimeWait(std::chrono::milliseconds(99));
+  EXPECT_CALL(callbacks_.dispatcher_, deferredDelete_(_));
   response_timeout_->invokeCallback();
 
   EXPECT_TRUE(stream_info.firstUpstreamTxByteSent().has_value());
@@ -5905,6 +6004,7 @@ TEST_F(RouterTest, CanaryStatusTrue) {
                                           {"x-envoy-upstream-canary", "false"},
                                           {"x-envoy-virtual-cluster", "hello"}});
   ON_CALL(*cm_.conn_pool_.host_, canary()).WillByDefault(Return(true));
+  EXPECT_CALL(callbacks_.dispatcher_, deferredDelete_(_));
   response_decoder->decodeHeaders(std::move(response_headers), true);
   EXPECT_TRUE(verifyHostUpstreamStats(1, 0));
 
@@ -5941,6 +6041,7 @@ TEST_F(RouterTest, CanaryStatusFalse) {
       new Http::TestResponseHeaderMapImpl{{":status", "200"},
                                           {"x-envoy-upstream-canary", "false"},
                                           {"x-envoy-virtual-cluster", "hello"}});
+  EXPECT_CALL(callbacks_.dispatcher_, deferredDelete_(_));
   response_decoder->decodeHeaders(std::move(response_headers), true);
   EXPECT_TRUE(verifyHostUpstreamStats(1, 0));
 
@@ -5986,6 +6087,7 @@ TEST_F(RouterTest, AutoHostRewriteEnabled) {
         EXPECT_EQ(host_address_, host->address());
       }));
   EXPECT_CALL(callbacks_.route_->route_entry_, autoHostRewrite()).WillOnce(Return(true));
+  EXPECT_CALL(callbacks_.dispatcher_, deferredDelete_(_));
   router_.decodeHeaders(incoming_headers, true);
   EXPECT_EQ(1U,
             callbacks_.route_->route_entry_.virtual_cluster_.stats().upstream_rq_total_.value());
@@ -6025,6 +6127,7 @@ TEST_F(RouterTest, AutoHostRewriteDisabled) {
         EXPECT_EQ(host_address_, host->address());
       }));
   EXPECT_CALL(callbacks_.route_->route_entry_, autoHostRewrite()).WillOnce(Return(false));
+  EXPECT_CALL(callbacks_.dispatcher_, deferredDelete_(_));
   router_.decodeHeaders(incoming_headers, true);
   EXPECT_EQ(1U,
             callbacks_.route_->route_entry_.virtual_cluster_.stats().upstream_rq_total_.value());
@@ -6110,6 +6213,7 @@ TEST_F(RouterTest, ConnectPauseAndResume) {
   // Make sure any early data does not go upstream.
   EXPECT_CALL(encoder, encodeData(_, _)).Times(0);
   Buffer::OwnedImpl data;
+  EXPECT_CALL(callbacks_.dispatcher_, deferredDelete_(_));
   router_.decodeData(data, true);
 
   // Now send the response headers, and ensure the deferred payload is proxied.
@@ -6151,6 +6255,7 @@ TEST_F(RouterTest, ConnectPauseNoResume) {
   // Make sure any early data does not go upstream.
   EXPECT_CALL(encoder, encodeData(_, _)).Times(0);
   Buffer::OwnedImpl data;
+  EXPECT_CALL(callbacks_.dispatcher_, deferredDelete_(_));
   router_.decodeData(data, true);
 
   // Now send the response headers, and ensure the deferred payload is not proxied.
@@ -6213,6 +6318,7 @@ class WatermarkTest : public RouterTest {
     }
   }
   void sendResponse() {
+    EXPECT_CALL(callbacks_.dispatcher_, deferredDelete_(_));
     response_decoder_->decodeHeaders(
         Http::ResponseHeaderMapPtr{new Http::TestResponseHeaderMapImpl{{":status", "200"}}}, true);
   }
@@ -6266,11 +6372,12 @@ TEST_F(WatermarkTest, UpstreamWatermarks) {
 
   Buffer::OwnedImpl data;
   EXPECT_CALL(encoder_, getStream()).Times(2).WillRepeatedly(ReturnRef(stream_));
+  EXPECT_CALL(callbacks_.dispatcher_, deferredDelete_(_));
   response_decoder_->decodeData(data, true);
 }
 
 TEST_F(WatermarkTest, FilterWatermarks) {
-  EXPECT_CALL(callbacks_, decoderBufferLimit()).Times(3).WillRepeatedly(Return(10));
+  EXPECT_CALL(callbacks_, decoderBufferLimit()).WillRepeatedly(Return(10));
   router_.setDecoderFilterCallbacks(callbacks_);
   // Send the headers sans-fin, and don't flag the pool as ready.
   sendRequest(false, false);
@@ -6366,6 +6473,7 @@ TEST_F(WatermarkTest, RetryRequestNotComplete) {
   // This should not trigger a retry as the retry state has been deleted.
   EXPECT_CALL(cm_.conn_pool_.host_->outlier_detector_,
               putResult(Upstream::Outlier::Result::LocalOriginConnectFailed, _));
+  EXPECT_CALL(callbacks_.dispatcher_, deferredDelete_(_));
   encoder1.stream_.resetStream(Http::StreamResetReason::RemoteReset);
   EXPECT_EQ(callbacks_.details(), "upstream_reset_before_response_started{remote reset}");
 }
@@ -6401,6 +6509,7 @@ TEST_F(RouterTestChildSpan, BasicFlow) {
   EXPECT_CALL(callbacks_.active_span_, spawnChild_(_, "router fake_cluster egress", _))
       .WillOnce(Return(child_span));
   EXPECT_CALL(callbacks_, tracingConfig());
+  EXPECT_CALL(callbacks_.dispatcher_, deferredDelete_(_));
   router_.decodeHeaders(headers, true);
   EXPECT_EQ(1U,
             callbacks_.route_->route_entry_.virtual_cluster_.stats().upstream_rq_total_.value());
@@ -6445,6 +6554,7 @@ TEST_F(RouterTestChildSpan, ResetFlow) {
   EXPECT_CALL(callbacks_.active_span_, spawnChild_(_, "router fake_cluster egress", _))
       .WillOnce(Return(child_span));
   EXPECT_CALL(callbacks_, tracingConfig());
+  EXPECT_CALL(callbacks_.dispatcher_, deferredDelete_(_));
   router_.decodeHeaders(headers, true);
   EXPECT_EQ(1U,
             callbacks_.route_->route_entry_.virtual_cluster_.stats().upstream_rq_total_.value());
@@ -6510,6 +6620,7 @@ TEST_F(RouterTestChildSpan, CancelFlow) {
   EXPECT_CALL(*child_span,
               setTag(Eq(Tracing::Tags::get().Canceled), Eq(Tracing::Tags::get().True)));
   EXPECT_CALL(*child_span, finishSpan());
+  EXPECT_CALL(callbacks_.dispatcher_, deferredDelete_(_));
   router_.onDestroy();
 }
 
@@ -6538,6 +6649,7 @@ TEST_F(RouterTestChildSpan, ResetRetryFlow) {
   EXPECT_CALL(callbacks_.active_span_, spawnChild_(_, "router fake_cluster egress", _))
       .WillOnce(Return(child_span_1));
   EXPECT_CALL(callbacks_, tracingConfig());
+  EXPECT_CALL(callbacks_.dispatcher_, deferredDelete_(_)).Times(2);
   router_.decodeHeaders(headers, true);
   EXPECT_EQ(1U,
             callbacks_.route_->route_entry_.virtual_cluster_.stats().upstream_rq_total_.value());
diff --git a/test/common/router/router_upstream_log_test.cc b/test/common/router/router_upstream_log_test.cc
index 632eca512b22..15b91e422791 100644
--- a/test/common/router/router_upstream_log_test.cc
+++ b/test/common/router/router_upstream_log_test.cc
@@ -145,6 +145,7 @@ class RouterUpstreamLogTest : public testing::Test {
 
     Http::TestRequestHeaderMapImpl headers(request_headers_init);
     HttpTestUtility::addDefaultHeaders(headers);
+    EXPECT_CALL(callbacks_.dispatcher_, deferredDelete_(_));
     router_->decodeHeaders(headers, true);
 
     EXPECT_CALL(*router_->retry_state_, shouldRetryHeaders(_, _)).WillOnce(Return(RetryStatus::No));
@@ -186,6 +187,7 @@ class RouterUpstreamLogTest : public testing::Test {
                                            {"x-envoy-internal", "true"},
                                            {"x-envoy-upstream-rq-per-try-timeout-ms", "5"}};
     HttpTestUtility::addDefaultHeaders(headers);
+    EXPECT_CALL(callbacks_.dispatcher_, deferredDelete_(_)).Times(2);
     router_->decodeHeaders(headers, true);
 
     router_->retry_state_->expectResetRetry();
diff --git a/test/extensions/upstreams/http/tcp/upstream_request_test.cc b/test/extensions/upstreams/http/tcp/upstream_request_test.cc
index ea7b55f8b2b1..889fed89a652 100644
--- a/test/extensions/upstreams/http/tcp/upstream_request_test.cc
+++ b/test/extensions/upstreams/http/tcp/upstream_request_test.cc
@@ -185,6 +185,28 @@ TEST_F(TcpUpstreamTest, V2Header) {
   tcp_upstream_->encodeData(buffer, false);
 }
 
+// Verifies that a reset after end_stream=true doesn't trigger a callback
+// on the router filter.
+TEST_F(TcpUpstreamTest, ResetAfterEndStream) {
+  Buffer::OwnedImpl buffer("something");
+  EXPECT_CALL(mock_router_filter_, onUpstreamData(BufferStringEqual("something"), _, true));
+  tcp_upstream_->onUpstreamData(buffer, true);
+  tcp_upstream_->onEvent(Network::ConnectionEvent::RemoteClose);
+}
+
+// Verifies that if we send data after the upstream has been reset nothing crashes.
+TEST_F(TcpUpstreamTest, DataAfterReset) {
+  tcp_upstream_->resetStream();
+  Buffer::OwnedImpl buffer("something");
+  tcp_upstream_->onUpstreamData(buffer, true);
+}
+
+// Verifies that if we send headers after the upstream has been reset nothing crashes.
+TEST_F(TcpUpstreamTest, HeadersAfterReset) {
+  tcp_upstream_->resetStream();
+  EXPECT_FALSE(tcp_upstream_->encodeHeaders(request_, false).ok());
+}
+
 TEST_F(TcpUpstreamTest, TrailersEndStream) {
   // Swallow the headers.
   EXPECT_TRUE(tcp_upstream_->encodeHeaders(request_, false).ok());