Http::FilterHeadersStatus Filter::decodeHeaders(Http::RequestHeaderMap& headers, bool end_stream) {
  downstream_headers_ = &headers;

  // Extract debug configuration from filter state. This is used further along to determine whether
  // we should append cluster and host headers to the response, and whether to forward the request
  // upstream.
  const StreamInfo::FilterStateSharedPtr& filter_state = callbacks_->streamInfo().filterState();
  const DebugConfig* debug_config =
      filter_state->hasData<DebugConfig>(DebugConfig::key())
          ? &(filter_state->getDataReadOnly<DebugConfig>(DebugConfig::key()))
          : nullptr;

  // TODO: Maybe add a filter API for this.
  grpc_request_ = Grpc::Common::isGrpcRequestHeaders(headers);
  exclude_http_code_stats_ = grpc_request_ && config_.suppress_grpc_request_failure_code_stats_;

  // Only increment rq total stat if we actually decode headers here. This does not count requests
  // that get handled by earlier filters.
  config_.stats_.rq_total_.inc();

  // Initialize the `modify_headers` function as a no-op (so we don't have to remember to check it
  // against nullptr before calling it), and feed it behavior later if/when we have cluster info
  // headers to append.
  std::function<void(Http::ResponseHeaderMap&)> modify_headers = [](Http::ResponseHeaderMap&) {};

  // Determine if there is a route entry or a direct response for the request.
  route_ = callbacks_->route();
  if (!route_) {
    config_.stats_.no_route_.inc();
    ENVOY_STREAM_LOG(debug, "no cluster match for URL '{}'", *callbacks_, headers.getPathValue());

    callbacks_->streamInfo().setResponseFlag(StreamInfo::ResponseFlag::NoRouteFound);
    callbacks_->sendLocalReply(Http::Code::NotFound, "", modify_headers, absl::nullopt,
                               StreamInfo::ResponseCodeDetails::get().RouteNotFound);
    return Http::FilterHeadersStatus::StopIteration;
  }

  // Determine if there is a direct response for the request.
  const auto* direct_response = route_->directResponseEntry();
  if (direct_response != nullptr) {
    config_.stats_.rq_direct_response_.inc();
    direct_response->rewritePathHeader(headers, !config_.suppress_envoy_headers_);
    callbacks_->streamInfo().setRouteName(direct_response->routeName());
    callbacks_->sendLocalReply(
        direct_response->responseCode(), direct_response->responseBody(),
        [this, direct_response,
         &request_headers = headers](Http::ResponseHeaderMap& response_headers) -> void {
          std::string new_path;
          if (request_headers.Path()) {
            new_path = direct_response->newPath(request_headers);
          }
          // See https://tools.ietf.org/html/rfc7231#section-7.1.2.
          const auto add_location =
              direct_response->responseCode() == Http::Code::Created ||
              Http::CodeUtility::is3xx(enumToInt(direct_response->responseCode()));
          if (!new_path.empty() && add_location) {
            response_headers.addReferenceKey(Http::Headers::get().Location, new_path);
          }
          direct_response->finalizeResponseHeaders(response_headers, callbacks_->streamInfo());
        },
        absl::nullopt, StreamInfo::ResponseCodeDetails::get().DirectResponse);
    return Http::FilterHeadersStatus::StopIteration;
  }

  // A route entry matches for the request.
  route_entry_ = route_->routeEntry();
  // If there's a route specific limit and it's smaller than general downstream
  // limits, apply the new cap.
  retry_shadow_buffer_limit_ =
      std::min(retry_shadow_buffer_limit_, route_entry_->retryShadowBufferLimit());
  callbacks_->streamInfo().setRouteName(route_entry_->routeName());
  if (debug_config && debug_config->append_cluster_) {
    // The cluster name will be appended to any local or upstream responses from this point.
    modify_headers = [this, debug_config](Http::ResponseHeaderMap& headers) {
      headers.addCopy(debug_config->cluster_header_.value_or(Http::Headers::get().EnvoyCluster),
                      route_entry_->clusterName());
    };
  }
  Upstream::ThreadLocalCluster* cluster =
      config_.cm_.getThreadLocalCluster(route_entry_->clusterName());
  if (!cluster) {
    config_.stats_.no_cluster_.inc();
    ENVOY_STREAM_LOG(debug, "unknown cluster '{}'", *callbacks_, route_entry_->clusterName());

    callbacks_->streamInfo().setResponseFlag(StreamInfo::ResponseFlag::NoClusterFound);
    callbacks_->sendLocalReply(route_entry_->clusterNotFoundResponseCode(), "", modify_headers,
                               absl::nullopt,
                               StreamInfo::ResponseCodeDetails::get().ClusterNotFound);
    return Http::FilterHeadersStatus::StopIteration;
  }
  cluster_ = cluster->info();

  // Set up stat prefixes, etc.
  request_vcluster_ = route_entry_->virtualCluster(headers);
  ENVOY_STREAM_LOG(debug, "cluster '{}' match for URL '{}'", *callbacks_,
                   route_entry_->clusterName(), headers.getPathValue());

  if (config_.strict_check_headers_ != nullptr) {
    for (const auto& header : *config_.strict_check_headers_) {
      const auto res = FilterUtility::StrictHeaderChecker::checkHeader(headers, header);
      if (!res.valid_) {
        callbacks_->streamInfo().setResponseFlag(
            StreamInfo::ResponseFlag::InvalidEnvoyRequestHeaders);
        const std::string body = fmt::format("invalid header '{}' with value '{}'",
                                             std::string(res.entry_->key().getStringView()),
                                             std::string(res.entry_->value().getStringView()));
        const std::string details =
            absl::StrCat(StreamInfo::ResponseCodeDetails::get().InvalidEnvoyRequestHeaders, "{",
                         res.entry_->key().getStringView(), "}");
        callbacks_->sendLocalReply(Http::Code::BadRequest, body, nullptr, absl::nullopt, details);
        return Http::FilterHeadersStatus::StopIteration;
      }
    }
  }

  const Http::HeaderEntry* request_alt_name = headers.EnvoyUpstreamAltStatName();
  if (request_alt_name) {
    alt_stat_prefix_ = std::make_unique<Stats::StatNameDynamicStorage>(
        request_alt_name->value().getStringView(), config_.scope_.symbolTable());
    headers.removeEnvoyUpstreamAltStatName();
  }

  // See if we are supposed to immediately kill some percentage of this cluster's traffic.
  if (cluster_->maintenanceMode()) {
    callbacks_->streamInfo().setResponseFlag(StreamInfo::ResponseFlag::UpstreamOverflow);
    chargeUpstreamCode(Http::Code::ServiceUnavailable, nullptr, true);
    callbacks_->sendLocalReply(
        Http::Code::ServiceUnavailable, "maintenance mode",
        [modify_headers, this](Http::ResponseHeaderMap& headers) {
          if (!config_.suppress_envoy_headers_) {
            headers.addReference(Http::Headers::get().EnvoyOverloaded,
                                 Http::Headers::get().EnvoyOverloadedValues.True);
          }
          // Note: append_cluster_info does not respect suppress_envoy_headers.
          modify_headers(headers);
        },
        absl::nullopt, StreamInfo::ResponseCodeDetails::get().MaintenanceMode);
    cluster_->stats().upstream_rq_maintenance_mode_.inc();
    return Http::FilterHeadersStatus::StopIteration;
  }

  // Fetch a connection pool for the upstream cluster.
  const auto& upstream_http_protocol_options = cluster_->upstreamHttpProtocolOptions();

  if (upstream_http_protocol_options.has_value() &&
      (upstream_http_protocol_options.value().auto_sni() ||
       upstream_http_protocol_options.value().auto_san_validation())) {
    // Default the header to Host/Authority header.
    absl::string_view header_value = headers.getHostValue();

    // Check whether `override_auto_sni_header` is specified.
    const auto override_auto_sni_header =
        upstream_http_protocol_options.value().override_auto_sni_header();
    if (!override_auto_sni_header.empty()) {
      // Use the header value from `override_auto_sni_header` to set the SNI value.
      const auto overridden_header_value = Http::HeaderUtility::getAllOfHeaderAsString(
          headers, Http::LowerCaseString(override_auto_sni_header));
      if (overridden_header_value.result().has_value() &&
          !overridden_header_value.result().value().empty()) {
        header_value = overridden_header_value.result().value();
      }
    }
    const auto parsed_authority = Http::Utility::parseAuthority(header_value);
    bool should_set_sni = !parsed_authority.is_ip_address_;
    // `host_` returns a string_view so doing this should be safe.
    absl::string_view sni_value = parsed_authority.host_;

    if (should_set_sni && upstream_http_protocol_options.value().auto_sni()) {
      callbacks_->streamInfo().filterState()->setData(
          Network::UpstreamServerName::key(),
          std::make_unique<Network::UpstreamServerName>(sni_value),
          StreamInfo::FilterState::StateType::Mutable);
    }

    if (upstream_http_protocol_options.value().auto_san_validation()) {
      callbacks_->streamInfo().filterState()->setData(
          Network::UpstreamSubjectAltNames::key(),
          std::make_unique<Network::UpstreamSubjectAltNames>(
              std::vector<std::string>{std::string(sni_value)}),
          StreamInfo::FilterState::StateType::Mutable);
    }
  }

  transport_socket_options_ = Network::TransportSocketOptionsUtility::fromFilterState(
      *callbacks_->streamInfo().filterState());

  auto has_options_from_downstream =
      downstreamConnection() && downstreamConnection()
                                    ->streamInfo()
                                    .filterState()
                                    .hasData<Network::UpstreamSocketOptionsFilterState>(
                                        Network::UpstreamSocketOptionsFilterState::key());

  if (has_options_from_downstream) {
    auto downstream_options = downstreamConnection()
                                  ->streamInfo()
                                  .filterState()
                                  .getDataReadOnly<Network::UpstreamSocketOptionsFilterState>(
                                      Network::UpstreamSocketOptionsFilterState::key())
                                  .value();
    if (!upstream_options_) {
      upstream_options_ = std::make_shared<Network::Socket::Options>();
    }
    Network::Socket::appendOptions(upstream_options_, downstream_options);
  }

  if (upstream_options_ && callbacks_->getUpstreamSocketOptions()) {
    Network::Socket::appendOptions(upstream_options_, callbacks_->getUpstreamSocketOptions());
  }

  std::unique_ptr<GenericConnPool> generic_conn_pool = createConnPool(*cluster);

  if (!generic_conn_pool) {
    sendNoHealthyUpstreamResponse();
    return Http::FilterHeadersStatus::StopIteration;
  }
  Upstream::HostDescriptionConstSharedPtr host = generic_conn_pool->host();

  if (debug_config && debug_config->append_upstream_host_) {
    // The hostname and address will be appended to any local or upstream responses from this point,
    // possibly in addition to the cluster name.
    modify_headers = [modify_headers, debug_config, host](Http::ResponseHeaderMap& headers) {
      modify_headers(headers);
      headers.addCopy(
          debug_config->hostname_header_.value_or(Http::Headers::get().EnvoyUpstreamHostname),
          host->hostname());
      headers.addCopy(debug_config->host_address_header_.value_or(
                          Http::Headers::get().EnvoyUpstreamHostAddress),
                      host->address()->asString());
    };
  }

  // If we've been instructed not to forward the request upstream, send an empty local response.
  if (debug_config && debug_config->do_not_forward_) {
    modify_headers = [modify_headers, debug_config](Http::ResponseHeaderMap& headers) {
      modify_headers(headers);
      headers.addCopy(
          debug_config->not_forwarded_header_.value_or(Http::Headers::get().EnvoyNotForwarded),
          "true");
    };
    callbacks_->sendLocalReply(Http::Code::NoContent, "", modify_headers, absl::nullopt, "");
    return Http::FilterHeadersStatus::StopIteration;
  }

  hedging_params_ = FilterUtility::finalHedgingParams(*route_entry_, headers);

  timeout_ = FilterUtility::finalTimeout(*route_entry_, headers, !config_.suppress_envoy_headers_,
                                         grpc_request_, hedging_params_.hedge_on_per_try_timeout_,
                                         config_.respect_expected_rq_timeout_);

  const Http::HeaderEntry* header_max_stream_duration_entry =
      headers.EnvoyUpstreamStreamDurationMs();
  if (header_max_stream_duration_entry) {
    dynamic_max_stream_duration_ =
        FilterUtility::tryParseHeaderTimeout(*header_max_stream_duration_entry);
    headers.removeEnvoyUpstreamStreamDurationMs();
  }

  // If this header is set with any value, use an alternate response code on timeout
  if (headers.EnvoyUpstreamRequestTimeoutAltResponse()) {
    timeout_response_code_ = Http::Code::NoContent;
    headers.removeEnvoyUpstreamRequestTimeoutAltResponse();
  }

  include_attempt_count_in_request_ = route_entry_->includeAttemptCountInRequest();
  if (include_attempt_count_in_request_) {
    headers.setEnvoyAttemptCount(attempt_count_);
  }

  // The router has reached a point where it is going to try to send a request upstream,
  // so now modify_headers should attach x-envoy-attempt-count to the downstream response if the
  // config flag is true.
  if (route_entry_->includeAttemptCountInResponse()) {
    modify_headers = [modify_headers, this](Http::ResponseHeaderMap& headers) {
      modify_headers(headers);

      // This header is added without checking for config_.suppress_envoy_headers_ to mirror what is
      // done for upstream requests.
      headers.setEnvoyAttemptCount(attempt_count_);
    };
  }
  callbacks_->streamInfo().setAttemptCount(attempt_count_);

  // Inject the active span's tracing context into the request headers.
  callbacks_->activeSpan().injectContext(headers);

  route_entry_->finalizeRequestHeaders(headers, callbacks_->streamInfo(),
                                       !config_.suppress_envoy_headers_);
  FilterUtility::setUpstreamScheme(
      headers, callbacks_->streamInfo().downstreamAddressProvider().sslConnection() != nullptr,
      host->transportSocketFactory().implementsSecureTransport());

  // Ensure an http transport scheme is selected before continuing with decoding.
  ASSERT(headers.Scheme());

  retry_state_ = createRetryState(
      route_entry_->retryPolicy(), headers, *cluster_, request_vcluster_, config_.runtime_,
      config_.random_, callbacks_->dispatcher(), config_.timeSource(), route_entry_->priority());

  // Determine which shadow policies to use. It's possible that we don't do any shadowing due to
  // runtime keys.
  for (const auto& shadow_policy : route_entry_->shadowPolicies()) {
    const auto& policy_ref = *shadow_policy;
    if (FilterUtility::shouldShadow(policy_ref, config_.runtime_, callbacks_->streamId())) {
      active_shadow_policies_.push_back(std::cref(policy_ref));
    }
  }

  internal_redirects_with_body_enabled_ =
      Runtime::runtimeFeatureEnabled("envoy.reloadable_features.internal_redirects_with_body");

  ENVOY_STREAM_LOG(debug, "router decoding headers:\n{}", *callbacks_, headers);

  // Hang onto the modify_headers function for later use in handling upstream responses.
  modify_headers_ = modify_headers;

  UpstreamRequestPtr upstream_request =
      std::make_unique<UpstreamRequest>(*this, std::move(generic_conn_pool));
  LinkedList::moveIntoList(std::move(upstream_request), upstream_requests_);
  upstream_requests_.front()->encodeHeaders(end_stream);
  if (end_stream) {
    onRequestComplete();
  }

  return Http::FilterHeadersStatus::StopIteration;
}