[metadata] Add an experiment to ensure a unique refcount on parsed slice strings (#33205)

The intuition here is that these strings may end up in the hpack table,
and then unnecessarily extend the lifetime of the read blocks.

Instead, take a copy of these short strings when we need to and allow
the incoming large memory object to be discarded.

---------

Co-authored-by: ctiller <ctiller@users.noreply.github.com>

bazel/experiments.bzl

@@ -28,6 +28,7 @@ EXPERIMENTS = {
             "event_engine_listener",
             "promise_based_client_call",
             "promise_based_server_call",
+            "unique_metadata_strings",
             "work_stealing",
         ],
         "cpp_end2end_test": [

src/core/ext/transport/chttp2/transport/hpack_parser.cc

@@ -983,7 +983,7 @@ class HPackParser::Parser {
     const auto transport_size =
         key_string.size() + value.wire_size + hpack_constants::kEntryOverhead;
     auto md = grpc_metadata_batch::Parse(
-        key_string, std::move(value_slice), transport_size,
+        key_string, std::move(value_slice), state_.add_to_table, transport_size,
         [key_string, &status, this](absl::string_view message, const Slice&) {
           if (!status.ok()) return;
           input_->SetErrorAndContinueParsing(

src/core/ext/transport/chttp2/transport/hpack_parser_table.cc

@@ -245,7 +245,7 @@ HPackTable::Memento MakeMemento(size_t i) {
   auto sm = kStaticTable[i];
   return HPackTable::Memento{
       grpc_metadata_batch::Parse(
-          sm.key, Slice::FromStaticString(sm.value),
+          sm.key, Slice::FromStaticString(sm.value), true,
           strlen(sm.key) + strlen(sm.value) + hpack_constants::kEntryOverhead,
           [](absl::string_view, const Slice&) {
             abort();  // not expecting to see this

src/core/lib/experiments/experiments.cc

@@ -88,6 +88,11 @@ const char* const description_canary_client_privacy =
 const char* const additional_constraints_canary_client_privacy = "{}";
 const char* const description_server_privacy = "If set, server privacy";
 const char* const additional_constraints_server_privacy = "{}";
+const char* const description_unique_metadata_strings =
+    "Ensure a unique copy of strings from parsed metadata are taken. The "
+    "hypothesis here is that ref counting these are causing read buffer "
+    "lifetimes to be extended leading to memory bloat.";
+const char* const additional_constraints_unique_metadata_strings = "{}";
 }  // namespace
 
 namespace grpc_core {
@@ -134,6 +139,8 @@ const ExperimentMetadata g_experiment_metadata[] = {
      additional_constraints_canary_client_privacy, false, false},
     {"server_privacy", description_server_privacy,
      additional_constraints_server_privacy, false, false},
+    {"unique_metadata_strings", description_unique_metadata_strings,
+     additional_constraints_unique_metadata_strings, false, true},
 };
 
 }  // namespace grpc_core
@@ -206,6 +213,11 @@ const char* const description_canary_client_privacy =
 const char* const additional_constraints_canary_client_privacy = "{}";
 const char* const description_server_privacy = "If set, server privacy";
 const char* const additional_constraints_server_privacy = "{}";
+const char* const description_unique_metadata_strings =
+    "Ensure a unique copy of strings from parsed metadata are taken. The "
+    "hypothesis here is that ref counting these are causing read buffer "
+    "lifetimes to be extended leading to memory bloat.";
+const char* const additional_constraints_unique_metadata_strings = "{}";
 }  // namespace
 
 namespace grpc_core {
@@ -252,6 +264,8 @@ const ExperimentMetadata g_experiment_metadata[] = {
      additional_constraints_canary_client_privacy, false, false},
     {"server_privacy", description_server_privacy,
      additional_constraints_server_privacy, false, false},
+    {"unique_metadata_strings", description_unique_metadata_strings,
+     additional_constraints_unique_metadata_strings, false, true},
 };
 
 }  // namespace grpc_core
@@ -324,6 +338,11 @@ const char* const description_canary_client_privacy =
 const char* const additional_constraints_canary_client_privacy = "{}";
 const char* const description_server_privacy = "If set, server privacy";
 const char* const additional_constraints_server_privacy = "{}";
+const char* const description_unique_metadata_strings =
+    "Ensure a unique copy of strings from parsed metadata are taken. The "
+    "hypothesis here is that ref counting these are causing read buffer "
+    "lifetimes to be extended leading to memory bloat.";
+const char* const additional_constraints_unique_metadata_strings = "{}";
 }  // namespace
 
 namespace grpc_core {
@@ -370,6 +389,8 @@ const ExperimentMetadata g_experiment_metadata[] = {
      additional_constraints_canary_client_privacy, false, false},
     {"server_privacy", description_server_privacy,
      additional_constraints_server_privacy, false, false},
+    {"unique_metadata_strings", description_unique_metadata_strings,
+     additional_constraints_unique_metadata_strings, false, true},
 };
 
 }  // namespace grpc_core

src/core/lib/experiments/experiments.h

@@ -79,6 +79,7 @@ inline bool IsWorkStealingEnabled() { return false; }
 inline bool IsClientPrivacyEnabled() { return false; }
 inline bool IsCanaryClientPrivacyEnabled() { return false; }
 inline bool IsServerPrivacyEnabled() { return false; }
+inline bool IsUniqueMetadataStringsEnabled() { return false; }
 
 #elif defined(GPR_WINDOWS)
 inline bool IsTcpFrameSizeTuningEnabled() { return false; }
@@ -101,6 +102,7 @@ inline bool IsWorkStealingEnabled() { return false; }
 inline bool IsClientPrivacyEnabled() { return false; }
 inline bool IsCanaryClientPrivacyEnabled() { return false; }
 inline bool IsServerPrivacyEnabled() { return false; }
+inline bool IsUniqueMetadataStringsEnabled() { return false; }
 
 #else
 inline bool IsTcpFrameSizeTuningEnabled() { return false; }
@@ -123,6 +125,7 @@ inline bool IsWorkStealingEnabled() { return false; }
 inline bool IsClientPrivacyEnabled() { return false; }
 inline bool IsCanaryClientPrivacyEnabled() { return false; }
 inline bool IsServerPrivacyEnabled() { return false; }
+inline bool IsUniqueMetadataStringsEnabled() { return false; }
 #endif
 
 #else
@@ -172,8 +175,10 @@ inline bool IsClientPrivacyEnabled() { return IsExperimentEnabled(16); }
 inline bool IsCanaryClientPrivacyEnabled() { return IsExperimentEnabled(17); }
 #define GRPC_EXPERIMENT_IS_INCLUDED_SERVER_PRIVACY
 inline bool IsServerPrivacyEnabled() { return IsExperimentEnabled(18); }
+#define GRPC_EXPERIMENT_IS_INCLUDED_UNIQUE_METADATA_STRINGS
+inline bool IsUniqueMetadataStringsEnabled() { return IsExperimentEnabled(19); }
 
-constexpr const size_t kNumExperiments = 19;
+constexpr const size_t kNumExperiments = 20;
 extern const ExperimentMetadata g_experiment_metadata[kNumExperiments];
 
 #endif

src/core/lib/experiments/experiments.yaml

@@ -149,3 +149,12 @@
   owner: alishananda@google.com
   test_tags: []
   allow_in_fuzzing_config: false
+- name: unique_metadata_strings
+  description:
+    Ensure a unique copy of strings from parsed metadata are taken.
+    The hypothesis here is that ref counting these are causing read buffer
+    lifetimes to be extended leading to memory bloat.
+  expiry: 2023/11/01
+  owner: ctiller@google.com
+  test_tags: ["core_end2end_test"]
+  allow_in_fuzzing_config: true

src/core/lib/experiments/rollouts.yaml

@@ -74,3 +74,5 @@
   default: false
 - name: server_privacy
   default: false
+- name: unique_metadata_strings
+  default: false

src/core/lib/slice/slice.h

@@ -342,6 +342,22 @@ class GPR_MSVC_EMPTY_BASE_CLASS_WORKAROUND Slice
     return Slice(TakeCSlice());
   }
 
+  // As per TakeOwned, but if the slice is refcounted and there are other refs
+  // then it will copy instead of ref-counting, to ensure the returned slice is
+  // not shared.
+  Slice TakeUniquelyOwned() {
+    if (c_slice().refcount == nullptr) {
+      return Slice(c_slice());
+    }
+    if (c_slice().refcount == grpc_slice_refcount::NoopRefcount()) {
+      return Slice(grpc_slice_copy(c_slice()));
+    }
+    if (c_slice().refcount->IsUnique()) {
+      return Slice(TakeCSlice());
+    }
+    return Slice(grpc_slice_copy(c_slice()));
+  }
+
   // AsOwned returns an owned slice but does not mutate the current slice,
   // meaning that it may add a reference to the underlying slice.
   Slice AsOwned() const {

src/core/lib/transport/metadata_batch.cc

@@ -63,7 +63,7 @@ absl::optional<absl::string_view> UnknownMap::GetStringValue(
 }  // namespace metadata_detail
 
 ContentTypeMetadata::MementoType ContentTypeMetadata::ParseMemento(
-    Slice value, MetadataParseErrorFn on_error) {
+    Slice value, bool, MetadataParseErrorFn on_error) {
   auto out = kInvalid;
   auto value_string = value.as_string_view();
   if (value_string == "application/grpc") {
@@ -105,7 +105,7 @@ const char* ContentTypeMetadata::DisplayValue(ValueType content_type) {
 }
 
 GrpcTimeoutMetadata::MementoType GrpcTimeoutMetadata::ParseMemento(
-    Slice value, MetadataParseErrorFn on_error) {
+    Slice value, bool, MetadataParseErrorFn on_error) {
   auto timeout = ParseTimeout(value);
   if (!timeout.has_value()) {
     on_error("invalid value", value);
@@ -127,7 +127,7 @@ Slice GrpcTimeoutMetadata::Encode(ValueType x) {
 }
 
 TeMetadata::MementoType TeMetadata::ParseMemento(
-    Slice value, MetadataParseErrorFn on_error) {
+    Slice value, bool, MetadataParseErrorFn on_error) {
   auto out = kInvalid;
   if (value == "trailers") {
     out = kTrailers;
@@ -191,7 +191,7 @@ const char* HttpSchemeMetadata::DisplayValue(ValueType content_type) {
 }
 
 HttpMethodMetadata::MementoType HttpMethodMetadata::ParseMemento(
-    Slice value, MetadataParseErrorFn on_error) {
+    Slice value, bool, MetadataParseErrorFn on_error) {
   auto out = kInvalid;
   auto value_string = value.as_string_view();
   if (value_string == "POST") {
@@ -236,7 +236,7 @@ const char* HttpMethodMetadata::DisplayValue(ValueType content_type) {
 }
 
 CompressionAlgorithmBasedMetadata::MementoType
-CompressionAlgorithmBasedMetadata::ParseMemento(Slice value,
+CompressionAlgorithmBasedMetadata::ParseMemento(Slice value, bool,
                                                 MetadataParseErrorFn on_error) {
   auto algorithm = ParseCompressionAlgorithm(value.as_string_view());
   if (!algorithm.has_value()) {
@@ -247,7 +247,7 @@ CompressionAlgorithmBasedMetadata::ParseMemento(Slice value,
 }
 
 Duration GrpcRetryPushbackMsMetadata::ParseMemento(
-    Slice value, MetadataParseErrorFn on_error) {
+    Slice value, bool, MetadataParseErrorFn on_error) {
   int64_t out;
   if (!absl::SimpleAtoi(value.as_string_view(), &out)) {
     on_error("not an integer", value);
@@ -269,7 +269,7 @@ std::string LbCostBinMetadata::DisplayValue(ValueType x) {
 }
 
 LbCostBinMetadata::MementoType LbCostBinMetadata::ParseMemento(
-    Slice value, MetadataParseErrorFn on_error) {
+    Slice value, bool, MetadataParseErrorFn on_error) {
   if (value.length() < sizeof(double)) {
     on_error("too short", value);
     return {0, ""};

src/core/lib/transport/metadata_batch.h

@@ -80,7 +80,9 @@ struct GrpcTimeoutMetadata {
   using MementoType = Duration;
   using CompressionTraits = TimeoutCompressor;
   static absl::string_view key() { return "grpc-timeout"; }
-  static MementoType ParseMemento(Slice value, MetadataParseErrorFn on_error);
+  static MementoType ParseMemento(Slice value,
+                                  bool will_keep_past_request_lifetime,
+                                  MetadataParseErrorFn on_error);
   static ValueType MementoToValue(MementoType timeout);
   static Slice Encode(ValueType x);
   static std::string DisplayValue(ValueType x) { return x.ToString(); }
@@ -100,7 +102,9 @@ struct TeMetadata {
   using MementoType = ValueType;
   using CompressionTraits = KnownValueCompressor<ValueType, kTrailers>;
   static absl::string_view key() { return "te"; }
-  static MementoType ParseMemento(Slice value, MetadataParseErrorFn on_error);
+  static MementoType ParseMemento(Slice value,
+                                  bool will_keep_past_request_lifetime,
+                                  MetadataParseErrorFn on_error);
   static ValueType MementoToValue(MementoType te) { return te; }
   static StaticSlice Encode(ValueType x) {
     GPR_ASSERT(x == kTrailers);
@@ -128,7 +132,9 @@ struct ContentTypeMetadata {
   using MementoType = ValueType;
   using CompressionTraits = KnownValueCompressor<ValueType, kApplicationGrpc>;
   static absl::string_view key() { return "content-type"; }
-  static MementoType ParseMemento(Slice value, MetadataParseErrorFn on_error);
+  static MementoType ParseMemento(Slice value,
+                                  bool will_keep_past_request_lifetime,
+                                  MetadataParseErrorFn on_error);
   static ValueType MementoToValue(MementoType content_type) {
     return content_type;
   }
@@ -151,7 +157,8 @@ struct HttpSchemeMetadata {
   using MementoType = ValueType;
   using CompressionTraits = HttpSchemeCompressor;
   static absl::string_view key() { return ":scheme"; }
-  static MementoType ParseMemento(Slice value, MetadataParseErrorFn on_error) {
+  static MementoType ParseMemento(Slice value, bool,
+                                  MetadataParseErrorFn on_error) {
     return Parse(value.as_string_view(), on_error);
   }
   static ValueType Parse(absl::string_view value,
@@ -180,7 +187,9 @@ struct HttpMethodMetadata {
   using MementoType = ValueType;
   using CompressionTraits = HttpMethodCompressor;
   static absl::string_view key() { return ":method"; }
-  static MementoType ParseMemento(Slice value, MetadataParseErrorFn on_error);
+  static MementoType ParseMemento(Slice value,
+                                  bool will_keep_past_request_lifetime,
+                                  MetadataParseErrorFn on_error);
   static ValueType MementoToValue(MementoType content_type) {
     return content_type;
   }
@@ -196,7 +205,9 @@ struct HttpMethodMetadata {
 struct CompressionAlgorithmBasedMetadata {
   using ValueType = grpc_compression_algorithm;
   using MementoType = ValueType;
-  static MementoType ParseMemento(Slice value, MetadataParseErrorFn on_error);
+  static MementoType ParseMemento(Slice value,
+                                  bool will_keep_past_request_lifetime,
+                                  MetadataParseErrorFn on_error);
   static ValueType MementoToValue(MementoType x) { return x; }
   static Slice Encode(ValueType x) {
     GPR_ASSERT(x != GRPC_COMPRESS_ALGORITHMS_COUNT);
@@ -234,7 +245,7 @@ struct GrpcAcceptEncodingMetadata {
   using ValueType = CompressionAlgorithmSet;
   using MementoType = ValueType;
   using CompressionTraits = StableValueCompressor;
-  static MementoType ParseMemento(Slice value, MetadataParseErrorFn) {
+  static MementoType ParseMemento(Slice value, bool, MetadataParseErrorFn) {
     return CompressionAlgorithmSet::FromString(value.as_string_view());
   }
   static ValueType MementoToValue(MementoType x) { return x; }
@@ -324,7 +335,7 @@ struct SimpleIntBasedMetadataBase {
 template <typename Int, Int kInvalidValue>
 struct SimpleIntBasedMetadata : public SimpleIntBasedMetadataBase<Int> {
   static constexpr Int invalid_value() { return kInvalidValue; }
-  static Int ParseMemento(Slice value, MetadataParseErrorFn on_error) {
+  static Int ParseMemento(Slice value, bool, MetadataParseErrorFn on_error) {
     Int out;
     if (!absl::SimpleAtoi(value.as_string_view(), &out)) {
       on_error("not an integer", value);
@@ -361,7 +372,9 @@ struct GrpcRetryPushbackMsMetadata {
   static Slice Encode(Duration x) { return Slice::FromInt64(x.millis()); }
   static int64_t DisplayValue(Duration x) { return x.millis(); }
   static int64_t DisplayMemento(Duration x) { return DisplayValue(x); }
-  static Duration ParseMemento(Slice value, MetadataParseErrorFn on_error);
+  static Duration ParseMemento(Slice value,
+                               bool will_keep_past_request_lifetime,
+                               MetadataParseErrorFn on_error);
 };
 
 // :status metadata trait.
@@ -388,7 +401,7 @@ struct GrpcLbClientStatsMetadata {
   static const char* DisplayMemento(MementoType) {
     return "<internal-lb-stats>";
   }
-  static MementoType ParseMemento(Slice, MetadataParseErrorFn) {
+  static MementoType ParseMemento(Slice, bool, MetadataParseErrorFn) {
     return nullptr;
   }
 };
@@ -419,7 +432,9 @@ struct LbCostBinMetadata {
   static Slice Encode(const ValueType& x);
   static std::string DisplayValue(ValueType x);
   static std::string DisplayMemento(MementoType x) { return DisplayValue(x); }
-  static MementoType ParseMemento(Slice value, MetadataParseErrorFn on_error);
+  static MementoType ParseMemento(Slice value,
+                                  bool will_keep_past_request_lifetime,
+                                  MetadataParseErrorFn on_error);
 };
 
 // Annotation added by a transport to note whether a failed request was never
@@ -574,9 +589,9 @@ struct ParseValue {
   template <ParseMementoFn parse_memento, MementoToValueFn memento_to_value>
   static GPR_ATTRIBUTE_NOINLINE auto Parse(Slice* value,
                                            MetadataParseErrorFn on_error)
-      -> decltype(memento_to_value(parse_memento(std::move(*value),
+      -> decltype(memento_to_value(parse_memento(std::move(*value), false,
                                                  on_error))) {
-    return memento_to_value(parse_memento(std::move(*value), on_error));
+    return memento_to_value(parse_memento(std::move(*value), false, on_error));
   }
 };
 
@@ -586,8 +601,10 @@ struct ParseValue {
 template <typename Container>
 class ParseHelper {
  public:
-  ParseHelper(Slice value, MetadataParseErrorFn on_error, size_t transport_size)
+  ParseHelper(Slice value, bool will_keep_past_request_lifetime,
+              MetadataParseErrorFn on_error, size_t transport_size)
       : value_(std::move(value)),
+        will_keep_past_request_lifetime_(will_keep_past_request_lifetime),
         on_error_(on_error),
         transport_size_(transport_size) {}
 
@@ -607,12 +624,14 @@ class ParseHelper {
   }
 
  private:
-  template <typename T, T (*parse_memento)(Slice, MetadataParseErrorFn)>
+  template <typename T, T (*parse_memento)(Slice, bool, MetadataParseErrorFn)>
   GPR_ATTRIBUTE_NOINLINE T ParseValueToMemento() {
-    return parse_memento(std::move(value_), on_error_);
+    return parse_memento(std::move(value_), will_keep_past_request_lifetime_,
+                         on_error_);
   }
 
   Slice value_;
+  const bool will_keep_past_request_lifetime_;
   MetadataParseErrorFn on_error_;
   const size_t transport_size_;
 };
@@ -1107,8 +1126,14 @@ MetadataValueAsSlice(typename Which::ValueType value) {
 //   static absl::string_view key() { return "grpc-xyz"; }
 //   // Parse a memento from a slice
 //   // Takes ownership of value
+//   // If will_keep_past_request_lifetime is true, expect that the returned
+//   // memento will be kept for a long time, and so try not to keep a ref to
+//   // the input slice.
 //   // Calls fn in the case of an error that should be reported to the user
-//   static MementoType ParseMemento(Slice value, MementoParseErrorFn fn) {
+//   static MementoType ParseMemento(
+//       Slice value,
+//       bool will_keep_past_request_lifetime,
+//       MementoParseErrorFn fn) {
 //   ...
 //   }
 //   // Convert a memento to a value
@@ -1335,10 +1360,12 @@ class MetadataMap {
   // Parse metadata from a key/value pair, and return an object representing
   // that result.
   static ParsedMetadata<Derived> Parse(absl::string_view key, Slice value,
+                                       bool will_keep_past_request_lifetime,
                                        uint32_t transport_size,
                                        MetadataParseErrorFn on_error) {
-    metadata_detail::ParseHelper<Derived> helper(value.TakeOwned(), on_error,
-                                                 transport_size);
+    metadata_detail::ParseHelper<Derived> helper(
+        value.TakeOwned(), will_keep_past_request_lifetime, on_error,
+        transport_size);
     return metadata_detail::NameLookup<Traits...>::Lookup(key, &helper);
   }
 

src/core/lib/transport/parsed_metadata.h

@@ -33,6 +33,7 @@
 
 #include <grpc/slice.h>
 
+#include "src/core/lib/experiments/experiments.h"
 #include "src/core/lib/gprpp/time.h"
 #include "src/core/lib/slice/slice.h"
 
@@ -193,14 +194,16 @@ class ParsedMetadata {
   // HTTP2 defined storage size of this metadatum.
   uint32_t transport_size() const { return transport_size_; }
   // Create a new parsed metadata with the same key but a different value.
-  ParsedMetadata WithNewValue(Slice value, uint32_t value_wire_size,
+  ParsedMetadata WithNewValue(Slice value, bool will_keep_past_request_lifetime,
+                              uint32_t value_wire_size,
                               MetadataParseErrorFn on_error) const {
     ParsedMetadata result;
     result.vtable_ = vtable_;
     result.value_ = value_;
     result.transport_size_ =
         TransportSize(static_cast<uint32_t>(key().length()), value_wire_size);
-    vtable_->with_new_value(&value, on_error, &result);
+    vtable_->with_new_value(&value, will_keep_past_request_lifetime, on_error,
+                            &result);
     return result;
   }
   std::string DebugString() const { return vtable_->debug_string(value_); }
@@ -224,6 +227,7 @@ class ParsedMetadata {
     void (*const set)(const Buffer& value, MetadataContainer* container);
     // result is a bitwise copy of the originating ParsedMetadata.
     void (*const with_new_value)(Slice* new_value,
+                                 bool will_keep_past_request_lifetime,
                                  MetadataParseErrorFn on_error,
                                  ParsedMetadata* result);
     std::string (*const debug_string)(const Buffer& value);
@@ -241,17 +245,22 @@ class ParsedMetadata {
   template <typename Which>
   static const VTable* SliceTraitVTable();
 
-  template <Slice (*ParseMemento)(Slice, MetadataParseErrorFn)>
+  template <Slice (*ParseMemento)(Slice, bool, MetadataParseErrorFn)>
   GPR_ATTRIBUTE_NOINLINE static void WithNewValueSetSlice(
-      Slice* slice, MetadataParseErrorFn on_error, ParsedMetadata* result) {
+      Slice* slice, bool will_keep_past_request_lifetime,
+      MetadataParseErrorFn on_error, ParsedMetadata* result) {
     result->value_.slice =
-        ParseMemento(std::move(*slice), on_error).TakeCSlice();
+        ParseMemento(std::move(*slice), will_keep_past_request_lifetime,
+                     on_error)
+            .TakeCSlice();
   }
 
-  template <typename T, T (*ParseMemento)(Slice, MetadataParseErrorFn)>
+  template <typename T, T (*ParseMemento)(Slice, bool, MetadataParseErrorFn)>
   GPR_ATTRIBUTE_NOINLINE static void WithNewValueSetTrivial(
-      Slice* slice, MetadataParseErrorFn on_error, ParsedMetadata* result) {
-    T memento = ParseMemento(std::move(*slice), on_error);
+      Slice* slice, bool will_keep_past_request_lifetime,
+      MetadataParseErrorFn on_error, ParsedMetadata* result) {
+    T memento = ParseMemento(std::move(*slice), will_keep_past_request_lifetime,
+                             on_error);
     memcpy(result->value_.trivial, &memento, sizeof(memento));
   }
 
@@ -272,7 +281,7 @@ ParsedMetadata<MetadataContainer>::EmptyVTable() {
       // set
       [](const Buffer&, MetadataContainer*) {},
       // with_new_value
-      [](Slice*, MetadataParseErrorFn, ParsedMetadata*) {},
+      [](Slice*, bool, MetadataParseErrorFn, ParsedMetadata*) {},
       // debug_string
       [](const Buffer&) -> std::string { return "empty"; },
       // key
@@ -330,9 +339,11 @@ ParsedMetadata<MetadataContainer>::NonTrivialTraitVTable() {
         map->Set(Which(), Which::MementoToValue(*p));
       },
       // with_new_value
-      [](Slice* value, MetadataParseErrorFn on_error, ParsedMetadata* result) {
-        result->value_.pointer = new typename Which::MementoType(
-            Which::ParseMemento(std::move(*value), on_error));
+      [](Slice* value, bool will_keep_past_request_lifetime,
+         MetadataParseErrorFn on_error, ParsedMetadata* result) {
+        result->value_.pointer =
+            new typename Which::MementoType(Which::ParseMemento(
+                std::move(*value), will_keep_past_request_lifetime, on_error));
       },
       // debug_string
       [](const Buffer& value) {
@@ -387,14 +398,17 @@ ParsedMetadata<MetadataContainer>::KeyValueVTable(absl::string_view key) {
     auto* p = static_cast<KV*>(value.pointer);
     map->unknown_.Append(p->first.as_string_view(), p->second.Ref());
   };
-  static const auto with_new_value = [](Slice* value, MetadataParseErrorFn,
-                                        ParsedMetadata* result) {
-    auto* p = new KV{
-        static_cast<KV*>(result->value_.pointer)->first.Ref(),
-        std::move(*value),
-    };
-    result->value_.pointer = p;
-  };
+  static const auto with_new_value =
+      [](Slice* value, bool will_keep_past_request_lifetime,
+         MetadataParseErrorFn, ParsedMetadata* result) {
+        auto* p = new KV{
+            static_cast<KV*>(result->value_.pointer)->first.Ref(),
+            will_keep_past_request_lifetime && IsUniqueMetadataStringsEnabled()
+                ? value->TakeUniquelyOwned()
+                : std::move(*value),
+        };
+        result->value_.pointer = p;
+      };
   static const auto debug_string = [](const Buffer& value) {
     auto* p = static_cast<KV*>(value.pointer);
     return absl::StrCat(p->first.as_string_view(), ": ",

src/core/lib/transport/simple_slice_based_metadata.h

@@ -19,6 +19,7 @@
 
 #include "absl/strings/string_view.h"
 
+#include "src/core/lib/experiments/experiments.h"
 #include "src/core/lib/slice/slice.h"
 #include "src/core/lib/transport/parsed_metadata.h"
 
@@ -30,8 +31,14 @@ namespace grpc_core {
 struct SimpleSliceBasedMetadata {
   using ValueType = Slice;
   using MementoType = Slice;
-  static MementoType ParseMemento(Slice value, MetadataParseErrorFn) {
-    return value.TakeOwned();
+  static MementoType ParseMemento(Slice value,
+                                  bool will_keep_past_request_lifetime,
+                                  MetadataParseErrorFn) {
+    if (will_keep_past_request_lifetime && IsUniqueMetadataStringsEnabled()) {
+      return value.TakeUniquelyOwned();
+    } else {
+      return value.TakeOwned();
+    }
   }
   static ValueType MementoToValue(MementoType value) { return value; }
   static Slice Encode(const ValueType& x) { return x.Ref(); }

test/core/filters/filter_test.cc

@@ -339,7 +339,7 @@ ClientMetadataHandle FilterTestBase::Call::NewClientMetadata(
   auto md = impl_->arena()->MakePooled<ClientMetadata>(impl_->arena());
   for (auto& p : init) {
     auto parsed = ClientMetadata::Parse(
-        p.first, Slice::FromCopiedString(p.second),
+        p.first, Slice::FromCopiedString(p.second), false,
         p.first.length() + p.second.length() + 32,
         [p](absl::string_view, const Slice&) {
           Crash(absl::StrCat("Illegal metadata value: ", p.first, ": ",
@@ -356,7 +356,7 @@ ServerMetadataHandle FilterTestBase::Call::NewServerMetadata(
   auto md = impl_->arena()->MakePooled<ClientMetadata>(impl_->arena());
   for (auto& p : init) {
     auto parsed = ServerMetadata::Parse(
-        p.first, Slice::FromCopiedString(p.second),
+        p.first, Slice::FromCopiedString(p.second), false,
         p.first.length() + p.second.length() + 32,
         [p](absl::string_view, const Slice&) {
           Crash(absl::StrCat("Illegal metadata value: ", p.first, ": ",

test/core/slice/slice_test.cc

@@ -24,6 +24,7 @@
 #include <string.h>
 
 #include <algorithm>
+#include <functional>
 #include <memory>
 #include <random>
 #include <string>
@@ -36,7 +37,9 @@
 #include <grpc/support/log.h>
 
 #include "src/core/lib/gprpp/memory.h"
+#include "src/core/lib/gprpp/no_destruct.h"
 #include "src/core/lib/slice/slice_internal.h"
+#include "src/core/lib/slice/slice_refcount.h"
 #include "test/core/util/build.h"
 
 TEST(GrpcSliceTest, MallocReturnsSomethingSensible) {
@@ -351,6 +354,36 @@ INSTANTIATE_TEST_SUITE_P(SliceSizedTest, SliceSizedTest,
                            return std::to_string(info.param);
                          });
 
+class TakeUniquelyOwnedTest
+    : public ::testing::TestWithParam<std::function<Slice()>> {};
+
+TEST_P(TakeUniquelyOwnedTest, TakeUniquelyOwned) {
+  auto owned = GetParam()().TakeUniquelyOwned();
+  auto* refcount = owned.c_slice().refcount;
+  if (refcount != nullptr && refcount != grpc_slice_refcount::NoopRefcount()) {
+    EXPECT_TRUE(refcount->IsUnique());
+  }
+}
+
+INSTANTIATE_TEST_SUITE_P(
+    TakeUniquelyOwnedTest, TakeUniquelyOwnedTest,
+    ::testing::Values(
+        []() {
+          static const NoDestruct<std::string> big('a', 1024);
+          return Slice::FromStaticBuffer(big->data(), big->size());
+        },
+        []() {
+          static const NoDestruct<std::string> big('a', 1024);
+          return Slice::FromCopiedBuffer(big->data(), big->size());
+        },
+        []() {
+          static const NoDestruct<std::string> big('a', 1024);
+          static const NoDestruct<Slice> big_slice(
+              Slice::FromCopiedBuffer(big->data(), big->size()));
+          return big_slice->Ref();
+        },
+        []() { return Slice::FromStaticString("hello"); }));
+
 size_t SumSlice(const Slice& slice) {
   size_t x = 0;
   for (size_t i = 0; i < slice.size(); ++i) {

test/core/transport/parsed_metadata_test.cc

@@ -37,7 +37,7 @@ struct CharTrait {
   static char test_value() { return 'a'; }
   static size_t test_memento_transport_size() { return 34; }
   static char MementoToValue(char memento) { return memento; }
-  static char ParseMemento(Slice slice, MetadataParseErrorFn) {
+  static char ParseMemento(Slice slice, bool, MetadataParseErrorFn) {
     return slice[0];
   }
   static std::string DisplayValue(char value) { return std::string(1, value); }
@@ -53,7 +53,7 @@ struct Int32Trait {
   static int32_t test_value() { return -1; }
   static size_t test_memento_transport_size() { return 478; }
   static int32_t MementoToValue(int32_t memento) { return memento; }
-  static int32_t ParseMemento(Slice slice, MetadataParseErrorFn) {
+  static int32_t ParseMemento(Slice slice, bool, MetadataParseErrorFn) {
     int32_t out;
     GPR_ASSERT(absl::SimpleAtoi(slice.as_string_view(), &out));
     return out;
@@ -73,7 +73,7 @@ struct Int64Trait {
   static int64_t test_value() { return -83481847284179298; }
   static size_t test_memento_transport_size() { return 87; }
   static int64_t MementoToValue(int64_t memento) { return -memento; }
-  static int64_t ParseMemento(Slice slice, MetadataParseErrorFn) {
+  static int64_t ParseMemento(Slice slice, bool, MetadataParseErrorFn) {
     int64_t out;
     GPR_ASSERT(absl::SimpleAtoi(slice.as_string_view(), &out));
     return out;
@@ -93,7 +93,7 @@ struct IntptrTrait {
   static intptr_t test_value() { return test_memento() / 2; }
   static size_t test_memento_transport_size() { return 800; }
   static intptr_t MementoToValue(intptr_t memento) { return memento / 2; }
-  static intptr_t ParseMemento(Slice slice, MetadataParseErrorFn) {
+  static intptr_t ParseMemento(Slice slice, bool, MetadataParseErrorFn) {
     intptr_t out;
     GPR_ASSERT(absl::SimpleAtoi(slice.as_string_view(), &out));
     return out;
@@ -115,7 +115,7 @@ struct StringTrait {
   static std::string MementoToValue(std::string memento) {
     return "hi " + memento;
   }
-  static std::string ParseMemento(Slice slice, MetadataParseErrorFn) {
+  static std::string ParseMemento(Slice slice, bool, MetadataParseErrorFn) {
     auto view = slice.as_string_view();
     return std::string(view.begin(), view.end());
   }
@@ -228,12 +228,13 @@ TEST(KeyValueTest, Simple) {
                            Slice::FromCopiedString("value"), 40);
   EXPECT_EQ(p->DebugString(), "key: value");
   EXPECT_EQ(p->transport_size(), 40);
-  PM p2 = p->WithNewValue(
-      Slice::FromCopiedString("some_other_value"), strlen("some_other_value"),
-      [](absl::string_view msg, const Slice& value) {
-        ASSERT_TRUE(false) << "Should not be called: msg=" << msg
-                           << ", value=" << value.as_string_view();
-      });
+  PM p2 = p->WithNewValue(Slice::FromCopiedString("some_other_value"), true,
+                          strlen("some_other_value"),
+                          [](absl::string_view msg, const Slice& value) {
+                            ASSERT_TRUE(false)
+                                << "Should not be called: msg=" << msg
+                                << ", value=" << value.as_string_view();
+                          });
   EXPECT_EQ(p->DebugString(), "key: value");
   EXPECT_EQ(p2.DebugString(), "key: some_other_value");
   EXPECT_EQ(p2.transport_size(), 51);
@@ -255,12 +256,13 @@ TEST(KeyValueTest, LongKey) {
       p->DebugString(),
       "aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa: value");
   EXPECT_EQ(p->transport_size(), 97);
-  PM p2 = p->WithNewValue(
-      Slice::FromCopiedString("some_other_value"), strlen("some_other_value"),
-      [](absl::string_view msg, const Slice& value) {
-        ASSERT_TRUE(false) << "Should not be called: msg=" << msg
-                           << ", value=" << value.as_string_view();
-      });
+  PM p2 = p->WithNewValue(Slice::FromCopiedString("some_other_value"), true,
+                          strlen("some_other_value"),
+                          [](absl::string_view msg, const Slice& value) {
+                            ASSERT_TRUE(false)
+                                << "Should not be called: msg=" << msg
+                                << ", value=" << value.as_string_view();
+                          });
   EXPECT_EQ(
       p->DebugString(),
       "aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa: value");