[metadata] Separate compression algorithm from metadata key. (#32860)

This change allows metadata keys to set the appropriate compression
algorithm in the hpack encoder, without needing to change the source
text of the hpack encoder.

We'll leverage this with #32650 to allow some important but
Google-internal metadata to be compressed appropriately.

---------

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

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

@@ -109,11 +109,22 @@ void HPackCompressor::Frame(const EncodeHeaderOptions& options,
   }
 }
 
-void HPackCompressor::Encoder::EmitIndexed(uint32_t elem_index) {
+void HPackCompressor::SetMaxUsableSize(uint32_t max_table_size) {
-  VarintWriter<1> w(elem_index);
+  max_usable_size_ = max_table_size;
-  w.Write(0x80, output_.AddTiny(w.length()));
+  SetMaxTableSize(std::min(table_.max_size(), max_table_size));
+}
+
+void HPackCompressor::SetMaxTableSize(uint32_t max_table_size) {
+  if (table_.SetMaxSize(std::min(max_usable_size_, max_table_size))) {
+    advertise_table_size_change_ = true;
+    if (GRPC_TRACE_FLAG_ENABLED(grpc_http_trace)) {
+      gpr_log(GPR_INFO, "set max table size from encoder to %d",
+              max_table_size);
+    }
+  }
 }
 
+namespace {
 struct WireValue {
   WireValue(uint8_t huffman_prefix, bool insert_null_before_wire_value,
             Slice slice)
@@ -127,8 +138,7 @@ struct WireValue {
   const size_t length;
 };
 
-static WireValue GetWireValue(Slice value, bool true_binary_enabled,
+WireValue GetWireValue(Slice value, bool true_binary_enabled, bool is_bin_hdr) {
-                              bool is_bin_hdr) {
   if (is_bin_hdr) {
     if (true_binary_enabled) {
       return WireValue(0x00, true, std::move(value));
@@ -214,9 +224,16 @@ class StringKey {
   Slice key_;
   VarintWriter<1> len_key_;
 };
+}  // namespace
+
+namespace hpack_encoder_detail {
+void Encoder::EmitIndexed(uint32_t elem_index) {
+  VarintWriter<1> w(elem_index);
+  w.Write(0x80, output_.AddTiny(w.length()));
+}
 
-void HPackCompressor::Encoder::EmitLitHdrWithNonBinaryStringKeyIncIdx(
+void Encoder::EmitLitHdrWithNonBinaryStringKeyIncIdx(Slice key_slice,
-    Slice key_slice, Slice value_slice) {
+                                                     Slice value_slice) {
   StringKey key(std::move(key_slice));
   key.WritePrefix(0x40, output_.AddTiny(key.prefix_length()));
   output_.Append(key.key());
@@ -225,8 +242,8 @@ void HPackCompressor::Encoder::EmitLitHdrWithNonBinaryStringKeyIncIdx(
   output_.Append(emit.data());
 }
 
-void HPackCompressor::Encoder::EmitLitHdrWithBinaryStringKeyNotIdx(
+void Encoder::EmitLitHdrWithBinaryStringKeyNotIdx(Slice key_slice,
-    Slice key_slice, Slice value_slice) {
+                                                  Slice value_slice) {
   StringKey key(std::move(key_slice));
   key.WritePrefix(0x00, output_.AddTiny(key.prefix_length()));
   output_.Append(key.key());
@@ -235,8 +252,8 @@ void HPackCompressor::Encoder::EmitLitHdrWithBinaryStringKeyNotIdx(
   output_.Append(emit.data());
 }
 
-void HPackCompressor::Encoder::EmitLitHdrWithBinaryStringKeyIncIdx(
+void Encoder::EmitLitHdrWithBinaryStringKeyIncIdx(Slice key_slice,
-    Slice key_slice, Slice value_slice) {
+                                                  Slice value_slice) {
   StringKey key(std::move(key_slice));
   key.WritePrefix(0x40, output_.AddTiny(key.prefix_length()));
   output_.Append(key.key());
@@ -245,8 +262,8 @@ void HPackCompressor::Encoder::EmitLitHdrWithBinaryStringKeyIncIdx(
   output_.Append(emit.data());
 }
 
-void HPackCompressor::Encoder::EmitLitHdrWithBinaryStringKeyNotIdx(
+void Encoder::EmitLitHdrWithBinaryStringKeyNotIdx(uint32_t key_index,
-    uint32_t key_index, Slice value_slice) {
+                                                  Slice value_slice) {
   BinaryStringValue emit(std::move(value_slice), use_true_binary_metadata_);
   VarintWriter<4> key(key_index);
   uint8_t* data = output_.AddTiny(key.length() + emit.prefix_length());
@@ -255,8 +272,8 @@ void HPackCompressor::Encoder::EmitLitHdrWithBinaryStringKeyNotIdx(
   output_.Append(emit.data());
 }
 
-void HPackCompressor::Encoder::EmitLitHdrWithNonBinaryStringKeyNotIdx(
+void Encoder::EmitLitHdrWithNonBinaryStringKeyNotIdx(Slice key_slice,
-    Slice key_slice, Slice value_slice) {
+                                                     Slice value_slice) {
   StringKey key(std::move(key_slice));
   key.WritePrefix(0x00, output_.AddTiny(key.prefix_length()));
   output_.Append(key.key());
@@ -265,14 +282,14 @@ void HPackCompressor::Encoder::EmitLitHdrWithNonBinaryStringKeyNotIdx(
   output_.Append(emit.data());
 }
 
-void HPackCompressor::Encoder::AdvertiseTableSizeChange() {
+void Encoder::AdvertiseTableSizeChange() {
   VarintWriter<3> w(compressor_->table_.max_size());
   w.Write(0x20, output_.AddTiny(w.length()));
 }
 
-void HPackCompressor::SliceIndex::EmitTo(absl::string_view key,
+void SliceIndex::EmitTo(absl::string_view key, const Slice& value,
-                                         const Slice& value, Encoder* encoder) {
+                        Encoder* encoder) {
-  auto& table = encoder->compressor_->table_;
+  auto& table = encoder->hpack_table();
   using It = std::vector<ValueIndex>::iterator;
   It prev = values_.end();
   size_t transport_length =
@@ -316,7 +333,7 @@ void HPackCompressor::SliceIndex::EmitTo(absl::string_view key,
   values_.emplace_back(value.Ref(), index);
 }
 
-void HPackCompressor::Encoder::Encode(const Slice& key, const Slice& value) {
+void Encoder::Encode(const Slice& key, const Slice& value) {
   if (absl::EndsWith(key.as_string_view(), "-bin")) {
     EmitLitHdrWithBinaryStringKeyNotIdx(key.Ref(), value.Ref());
   } else {
@@ -324,43 +341,14 @@ void HPackCompressor::Encoder::Encode(const Slice& key, const Slice& value) {
   }
 }
 
-void HPackCompressor::Encoder::Encode(HttpPathMetadata, const Slice& value) {
+void Compressor<HttpSchemeMetadata, HttpSchemeCompressor>::EncodeWith(
-  compressor_->path_index_.EmitTo(HttpPathMetadata::key(), value, this);
+    HttpSchemeMetadata, HttpSchemeMetadata::ValueType value, Encoder* encoder) {
-}
-
-void HPackCompressor::Encoder::Encode(HttpAuthorityMetadata,
-                                      const Slice& value) {
-  compressor_->authority_index_.EmitTo(HttpAuthorityMetadata::key(), value,
-                                       this);
-}
-
-void HPackCompressor::Encoder::Encode(TeMetadata, TeMetadata::ValueType value) {
-  GPR_ASSERT(value == TeMetadata::ValueType::kTrailers);
-  EncodeAlwaysIndexed(
-      &compressor_->te_index_, "te", Slice::FromStaticString("trailers"),
-      2 /* te */ + 8 /* trailers */ + hpack_constants::kEntryOverhead);
-}
-
-void HPackCompressor::Encoder::Encode(ContentTypeMetadata,
-                                      ContentTypeMetadata::ValueType value) {
-  if (value != ContentTypeMetadata::ValueType::kApplicationGrpc) {
-    gpr_log(GPR_ERROR, "Not encoding bad content-type header");
-    return;
-  }
-  EncodeAlwaysIndexed(&compressor_->content_type_index_, "content-type",
-                      Slice::FromStaticString("application/grpc"),
-                      12 /* content-type */ + 16 /* application/grpc */ +
-                          hpack_constants::kEntryOverhead);
-}
-
-void HPackCompressor::Encoder::Encode(HttpSchemeMetadata,
-                                      HttpSchemeMetadata::ValueType value) {
   switch (value) {
     case HttpSchemeMetadata::ValueType::kHttp:
-      EmitIndexed(6);  // :scheme: http
+      encoder->EmitIndexed(6);  // :scheme: http
       break;
     case HttpSchemeMetadata::ValueType::kHttps:
-      EmitIndexed(7);  // :scheme: https
+      encoder->EmitIndexed(7);  // :scheme: https
       break;
     case HttpSchemeMetadata::ValueType::kInvalid:
       Crash("invalid http scheme encoding");
@@ -368,22 +356,10 @@ void HPackCompressor::Encoder::Encode(HttpSchemeMetadata,
   }
 }
 
-void HPackCompressor::Encoder::Encode(GrpcTraceBinMetadata,
+void Compressor<HttpStatusMetadata, HttpStatusCompressor>::EncodeWith(
-                                      const Slice& slice) {
+    HttpStatusMetadata, uint32_t status, Encoder* encoder) {
-  EncodeRepeatingSliceValue(GrpcTraceBinMetadata::key(), slice,
-                            &compressor_->grpc_trace_bin_index_,
-                            HPackEncoderTable::MaxEntrySize());
-}
-
-void HPackCompressor::Encoder::Encode(GrpcTagsBinMetadata, const Slice& slice) {
-  EncodeRepeatingSliceValue(GrpcTagsBinMetadata::key(), slice,
-                            &compressor_->grpc_tags_bin_index_,
-                            HPackEncoderTable::MaxEntrySize());
-}
-
-void HPackCompressor::Encoder::Encode(HttpStatusMetadata, uint32_t status) {
   if (status == 200) {
-    EmitIndexed(8);  // :status: 200
+    encoder->EmitIndexed(8);  // :status: 200
     return;
   }
   uint8_t index = 0;
@@ -408,27 +384,28 @@ void HPackCompressor::Encoder::Encode(HttpStatusMetadata, uint32_t status) {
       break;
   }
   if (GPR_LIKELY(index != 0)) {
-    EmitIndexed(index);
+    encoder->EmitIndexed(index);
   } else {
-    EmitLitHdrWithNonBinaryStringKeyIncIdx(Slice::FromStaticString(":status"),
+    encoder->EmitLitHdrWithNonBinaryStringKeyIncIdx(
-                                           Slice::FromInt64(status));
+        Slice::FromStaticString(":status"), Slice::FromInt64(status));
   }
 }
 
-void HPackCompressor::Encoder::Encode(HttpMethodMetadata,
+void Compressor<HttpMethodMetadata, HttpMethodCompressor>::EncodeWith(
-                                      HttpMethodMetadata::ValueType method) {
+    HttpMethodMetadata, HttpMethodMetadata::ValueType method,
+    Encoder* encoder) {
   switch (method) {
     case HttpMethodMetadata::ValueType::kPost:
-      EmitIndexed(3);  // :method: POST
+      encoder->EmitIndexed(3);  // :method: POST
       break;
     case HttpMethodMetadata::ValueType::kGet:
-      EmitIndexed(2);  // :method: GET
+      encoder->EmitIndexed(2);  // :method: GET
       break;
     case HttpMethodMetadata::ValueType::kPut:
       // Right now, we only emit PUT as a method for testing purposes, so it's
       // fine to not index it.
-      EmitLitHdrWithNonBinaryStringKeyNotIdx(Slice::FromStaticString(":method"),
+      encoder->EmitLitHdrWithNonBinaryStringKeyNotIdx(
-                                             Slice::FromStaticString("PUT"));
+          Slice::FromStaticString(":method"), Slice::FromStaticString("PUT"));
       break;
     case HttpMethodMetadata::ValueType::kInvalid:
       Crash("invalid http method encoding");
@@ -436,10 +413,8 @@ void HPackCompressor::Encoder::Encode(HttpMethodMetadata,
   }
 }
 
-void HPackCompressor::Encoder::EncodeAlwaysIndexed(uint32_t* index,
+void Encoder::EncodeAlwaysIndexed(uint32_t* index, absl::string_view key,
-                                                   absl::string_view key,
+                                  Slice value, size_t transport_length) {
-                                                   Slice value,
-                                                   size_t transport_length) {
   if (compressor_->table_.ConvertableToDynamicIndex(*index)) {
     EmitIndexed(compressor_->table_.DynamicIndex(*index));
   } else {
@@ -449,8 +424,9 @@ void HPackCompressor::Encoder::EncodeAlwaysIndexed(uint32_t* index,
   }
 }
 
-void HPackCompressor::Encoder::EncodeIndexedKeyWithBinaryValue(
+void Encoder::EncodeIndexedKeyWithBinaryValue(uint32_t* index,
-    uint32_t* index, absl::string_view key, Slice value) {
+                                              absl::string_view key,
+                                              Slice value) {
   if (compressor_->table_.ConvertableToDynamicIndex(*index)) {
     EmitLitHdrWithBinaryStringKeyNotIdx(
         compressor_->table_.DynamicIndex(*index), std::move(value));
@@ -462,9 +438,9 @@ void HPackCompressor::Encoder::EncodeIndexedKeyWithBinaryValue(
   }
 }
 
-void HPackCompressor::Encoder::EncodeRepeatingSliceValue(
+void Encoder::EncodeRepeatingSliceValue(const absl::string_view& key,
-    const absl::string_view& key, const Slice& slice, uint32_t* index,
+                                        const Slice& slice, uint32_t* index,
-    size_t max_compression_size) {
+                                        size_t max_compression_size) {
   if (hpack_constants::SizeForEntry(key.size(), slice.size()) >
       max_compression_size) {
     EmitLitHdrWithBinaryStringKeyNotIdx(Slice::FromStaticString(key),
@@ -474,141 +450,39 @@ void HPackCompressor::Encoder::EncodeRepeatingSliceValue(
   }
 }
 
-void HPackCompressor::Encoder::Encode(GrpcTimeoutMetadata, Timestamp deadline) {
+void TimeoutCompressorImpl::EncodeWith(absl::string_view key,
+                                       Timestamp deadline, Encoder* encoder) {
   Timeout timeout = Timeout::FromDuration(deadline - Timestamp::Now());
-  for (auto it = compressor_->previous_timeouts_.begin();
+  auto& table = encoder->hpack_table();
-       it != compressor_->previous_timeouts_.end(); ++it) {
+  for (auto it = previous_timeouts_.begin(); it != previous_timeouts_.end();
+       ++it) {
     double ratio = timeout.RatioVersus(it->timeout);
     // If the timeout we're sending is shorter than a previous timeout, but
     // within 3% of it, we'll consider sending it.
     if (ratio > -3 && ratio <= 0 &&
-        compressor_->table_.ConvertableToDynamicIndex(it->index)) {
+        table.ConvertableToDynamicIndex(it->index)) {
-      EmitIndexed(compressor_->table_.DynamicIndex(it->index));
+      encoder->EmitIndexed(table.DynamicIndex(it->index));
       // Put this timeout to the front of the queue - forces common timeouts to
       // be considered earlier.
-      std::swap(*it, *compressor_->previous_timeouts_.begin());
+      std::swap(*it, *previous_timeouts_.begin());
       return;
     }
   }
   // Clean out some expired timeouts.
-  while (!compressor_->previous_timeouts_.empty() &&
+  while (!previous_timeouts_.empty() &&
-         !compressor_->table_.ConvertableToDynamicIndex(
+         !table.ConvertableToDynamicIndex(previous_timeouts_.back().index)) {
-             compressor_->previous_timeouts_.back().index)) {
+    previous_timeouts_.pop_back();
-    compressor_->previous_timeouts_.pop_back();
   }
   Slice encoded = timeout.Encode();
-  uint32_t index = compressor_->table_.AllocateIndex(
+  uint32_t index = table.AllocateIndex(key.length() + encoded.length() +
-      GrpcTimeoutMetadata::key().length() + encoded.length() +
+                                       hpack_constants::kEntryOverhead);
-      hpack_constants::kEntryOverhead);
+  previous_timeouts_.push_back(PreviousTimeout{timeout, index});
-  compressor_->previous_timeouts_.push_back(PreviousTimeout{timeout, index});
+  encoder->EmitLitHdrWithNonBinaryStringKeyIncIdx(Slice::FromStaticString(key),
-  EmitLitHdrWithNonBinaryStringKeyIncIdx(
+                                                  std::move(encoded));
-      Slice::FromStaticString(GrpcTimeoutMetadata::key()), std::move(encoded));
-}
-
-void HPackCompressor::Encoder::Encode(UserAgentMetadata, const Slice& slice) {
-  if (hpack_constants::SizeForEntry(UserAgentMetadata::key().size(),
-                                    slice.size()) >
-      HPackEncoderTable::MaxEntrySize()) {
-    EmitLitHdrWithNonBinaryStringKeyNotIdx(
-        Slice::FromStaticString(UserAgentMetadata::key()), slice.Ref());
-    return;
-  }
-  if (!slice.is_equivalent(compressor_->user_agent_)) {
-    compressor_->user_agent_ = slice.Ref();
-    compressor_->user_agent_index_ = 0;
-  }
-  EncodeAlwaysIndexed(&compressor_->user_agent_index_, UserAgentMetadata::key(),
-                      slice.Ref(),
-                      hpack_constants::SizeForEntry(
-                          UserAgentMetadata::key().size(), slice.size()));
-}
-
-void HPackCompressor::Encoder::Encode(GrpcStatusMetadata,
-                                      grpc_status_code status) {
-  const uint32_t code = static_cast<uint32_t>(status);
-  uint32_t* index = nullptr;
-  if (code < kNumCachedGrpcStatusValues) {
-    index = &compressor_->cached_grpc_status_[code];
-    if (compressor_->table_.ConvertableToDynamicIndex(*index)) {
-      EmitIndexed(compressor_->table_.DynamicIndex(*index));
-      return;
-    }
-  }
-  Slice key = Slice::FromStaticString(GrpcStatusMetadata::key());
-  Slice value = Slice::FromInt64(code);
-  const size_t transport_length =
-      key.length() + value.length() + hpack_constants::kEntryOverhead;
-  if (index != nullptr) {
-    *index = compressor_->table_.AllocateIndex(transport_length);
-    EmitLitHdrWithNonBinaryStringKeyIncIdx(std::move(key), std::move(value));
-  } else {
-    EmitLitHdrWithNonBinaryStringKeyNotIdx(std::move(key), std::move(value));
-  }
-}
-
-void HPackCompressor::Encoder::Encode(GrpcEncodingMetadata,
-                                      grpc_compression_algorithm value) {
-  uint32_t* index = nullptr;
-  if (value < GRPC_COMPRESS_ALGORITHMS_COUNT) {
-    index = &compressor_->cached_grpc_encoding_[static_cast<uint32_t>(value)];
-    if (compressor_->table_.ConvertableToDynamicIndex(*index)) {
-      EmitIndexed(compressor_->table_.DynamicIndex(*index));
-      return;
-    }
-  }
-  auto key = Slice::FromStaticString(GrpcEncodingMetadata::key());
-  auto encoded_value = GrpcEncodingMetadata::Encode(value);
-  size_t transport_length =
-      key.length() + encoded_value.length() + hpack_constants::kEntryOverhead;
-  if (index != nullptr) {
-    *index = compressor_->table_.AllocateIndex(transport_length);
-    EmitLitHdrWithNonBinaryStringKeyIncIdx(std::move(key),
-                                           std::move(encoded_value));
-  } else {
-    EmitLitHdrWithNonBinaryStringKeyNotIdx(std::move(key),
-                                           std::move(encoded_value));
-  }
-}
-
-void HPackCompressor::Encoder::Encode(GrpcAcceptEncodingMetadata,
-                                      CompressionAlgorithmSet value) {
-  if (compressor_->grpc_accept_encoding_index_ != 0 &&
-      value == compressor_->grpc_accept_encoding_ &&
-      compressor_->table_.ConvertableToDynamicIndex(
-          compressor_->grpc_accept_encoding_index_)) {
-    EmitIndexed(compressor_->table_.DynamicIndex(
-        compressor_->grpc_accept_encoding_index_));
-    return;
-  }
-  auto key = Slice::FromStaticString(GrpcAcceptEncodingMetadata::key());
-  auto encoded_value = GrpcAcceptEncodingMetadata::Encode(value);
-  size_t transport_length =
-      key.length() + encoded_value.length() + hpack_constants::kEntryOverhead;
-  compressor_->grpc_accept_encoding_index_ =
-      compressor_->table_.AllocateIndex(transport_length);
-  compressor_->grpc_accept_encoding_ = value;
-  EmitLitHdrWithNonBinaryStringKeyIncIdx(std::move(key),
-                                         std::move(encoded_value));
-}
-
-void HPackCompressor::SetMaxUsableSize(uint32_t max_table_size) {
-  max_usable_size_ = max_table_size;
-  SetMaxTableSize(std::min(table_.max_size(), max_table_size));
-}
-
-void HPackCompressor::SetMaxTableSize(uint32_t max_table_size) {
-  if (table_.SetMaxSize(std::min(max_usable_size_, max_table_size))) {
-    advertise_table_size_change_ = true;
-    if (GRPC_TRACE_FLAG_ENABLED(grpc_http_trace)) {
-      gpr_log(GPR_INFO, "set max table size from encoder to %d",
-              max_table_size);
-    }
-  }
 }
 
-HPackCompressor::Encoder::Encoder(HPackCompressor* compressor,
+Encoder::Encoder(HPackCompressor* compressor, bool use_true_binary_metadata,
-                                  bool use_true_binary_metadata,
+                 SliceBuffer& output)
-                                  SliceBuffer& output)
     : use_true_binary_metadata_(use_true_binary_metadata),
       compressor_(compressor),
       output_(output) {
@@ -617,4 +491,5 @@ HPackCompressor::Encoder::Encoder(HPackCompressor* compressor,
   }
 }
 
+}  // namespace hpack_encoder_detail
 }  // namespace grpc_core

src/core/ext/transport/chttp2/transport/hpack_encoder.h

@@ -28,15 +28,14 @@
 #include <vector>
 
 #include "absl/strings/match.h"
+#include "absl/strings/str_cat.h"
 #include "absl/strings/string_view.h"
 
-#include <grpc/impl/compression_types.h>
 #include <grpc/slice.h>
-#include <grpc/status.h>
+#include <grpc/support/log.h>
 
 #include "src/core/ext/transport/chttp2/transport/hpack_constants.h"
 #include "src/core/ext/transport/chttp2/transport/hpack_encoder_table.h"
-#include "src/core/lib/compression/compression_internal.h"
 #include "src/core/lib/gprpp/time.h"
 #include "src/core/lib/slice/slice.h"
 #include "src/core/lib/slice/slice_buffer.h"
@@ -46,6 +45,280 @@
 
 namespace grpc_core {
 
+// Forward decl for encoder
+class HPackCompressor;
+
+namespace hpack_encoder_detail {
+
+class Encoder {
+ public:
+  Encoder(HPackCompressor* compressor, bool use_true_binary_metadata,
+          SliceBuffer& output);
+
+  void Encode(const Slice& key, const Slice& value);
+  template <typename MetadataTrait>
+  void Encode(MetadataTrait, const typename MetadataTrait::ValueType& value);
+
+  void AdvertiseTableSizeChange();
+  void EmitIndexed(uint32_t index);
+  void EmitLitHdrWithNonBinaryStringKeyIncIdx(Slice key_slice,
+                                              Slice value_slice);
+  void EmitLitHdrWithBinaryStringKeyIncIdx(Slice key_slice, Slice value_slice);
+  void EmitLitHdrWithBinaryStringKeyNotIdx(Slice key_slice, Slice value_slice);
+  void EmitLitHdrWithBinaryStringKeyNotIdx(uint32_t key_index,
+                                           Slice value_slice);
+  void EmitLitHdrWithNonBinaryStringKeyNotIdx(Slice key_slice,
+                                              Slice value_slice);
+
+  void EncodeAlwaysIndexed(uint32_t* index, absl::string_view key, Slice value,
+                           size_t transport_length);
+  void EncodeIndexedKeyWithBinaryValue(uint32_t* index, absl::string_view key,
+                                       Slice value);
+
+  void EncodeRepeatingSliceValue(const absl::string_view& key,
+                                 const Slice& slice, uint32_t* index,
+                                 size_t max_compression_size);
+
+  HPackEncoderTable& hpack_table();
+
+ private:
+  const bool use_true_binary_metadata_;
+  HPackCompressor* const compressor_;
+  SliceBuffer& output_;
+};
+
+// Compressor is partially specialized on CompressionTraits, but leaves
+// MetadataTrait as variable.
+// Via MetadataMap::StatefulCompressor it builds compression state for
+// HPackCompressor.
+// Each trait compressor gets to have some persistent state across the channel
+// (declared as Compressor member variables).
+// The compressors expose a single method:
+// void EncodeWith(MetadataTrait, const MetadataTrait::ValueType, Encoder*);
+// This method figures out how to encode the value, and then delegates to
+// Encoder to perform the encoding.
+template <typename MetadataTrait, typename CompressonTraits>
+class Compressor;
+
+// No compression encoder: just emit the key and value as literals.
+template <typename MetadataTrait>
+class Compressor<MetadataTrait, NoCompressionCompressor> {
+ public:
+  void EncodeWith(MetadataTrait, const typename MetadataTrait::ValueType& value,
+                  Encoder* encoder) {
+    const Slice& slice = MetadataValueAsSlice<MetadataTrait>(value);
+    if (absl::EndsWith(MetadataTrait::key(), "-bin")) {
+      encoder->EmitLitHdrWithBinaryStringKeyNotIdx(
+          Slice::FromStaticString(MetadataTrait::key()), slice.Ref());
+    } else {
+      encoder->EmitLitHdrWithNonBinaryStringKeyNotIdx(
+          Slice::FromStaticString(MetadataTrait::key()), slice.Ref());
+    }
+  }
+};
+
+// Frequent key with no value compression encoder
+template <typename MetadataTrait>
+class Compressor<MetadataTrait, FrequentKeyWithNoValueCompressionCompressor> {
+ public:
+  void EncodeWith(MetadataTrait, const typename MetadataTrait::ValueType& value,
+                  Encoder* encoder) {
+    const Slice& slice = MetadataValueAsSlice<MetadataTrait>(value);
+    encoder->EncodeRepeatingSliceValue(MetadataTrait::key(), slice,
+                                       &some_sent_value_,
+                                       HPackEncoderTable::MaxEntrySize());
+  }
+
+ private:
+  // Some previously sent value with this tag.
+  uint32_t some_sent_value_ = 0;
+};
+
+// Helper to determine if two objects have the same identity.
+// Equivalent here => equality, but equality does not imply equivalency.
+// For example, two slices with the same contents are equal, but not
+// equivalent.
+// Used as a much faster check for equality than the full equality check,
+// since many metadatum that are stable have the same root object in metadata
+// maps.
+template <typename T>
+static bool IsEquivalent(T a, T b) {
+  return a == b;
+}
+
+template <typename T>
+static bool IsEquivalent(const Slice& a, const Slice& b) {
+  return a.is_equivalent(b);
+}
+
+template <typename T>
+static void SaveCopyTo(const T& value, T& copy) {
+  copy = value;
+}
+
+static inline void SaveCopyTo(const Slice& value, Slice& copy) {
+  copy = value.Ref();
+}
+
+template <typename MetadataTrait>
+class Compressor<MetadataTrait, StableValueCompressor> {
+ public:
+  void EncodeWith(MetadataTrait, const typename MetadataTrait::ValueType& value,
+                  Encoder* encoder) {
+    auto& table = encoder->hpack_table();
+    if (previously_sent_value_ == value &&
+        table.ConvertableToDynamicIndex(previously_sent_index_)) {
+      encoder->EmitIndexed(table.DynamicIndex(previously_sent_index_));
+      return;
+    }
+    previously_sent_index_ = 0;
+    auto key = MetadataTrait::key();
+    const Slice& value_slice = MetadataValueAsSlice<MetadataTrait>(value);
+    if (hpack_constants::SizeForEntry(key.size(), value_slice.size()) >
+        HPackEncoderTable::MaxEntrySize()) {
+      encoder->EmitLitHdrWithNonBinaryStringKeyNotIdx(
+          Slice::FromStaticString(key), value_slice.Ref());
+      return;
+    }
+    encoder->EncodeAlwaysIndexed(
+        &previously_sent_index_, key, value_slice.Ref(),
+        hpack_constants::SizeForEntry(key.size(), value_slice.size()));
+    SaveCopyTo(value, previously_sent_value_);
+  }
+
+ private:
+  // Previously sent value
+  typename MetadataTrait::ValueType previously_sent_value_{};
+  // And its index in the table
+  uint32_t previously_sent_index_ = 0;
+};
+
+template <typename MetadataTrait, typename MetadataTrait::ValueType known_value>
+class Compressor<
+    MetadataTrait,
+    KnownValueCompressor<typename MetadataTrait::ValueType, known_value>> {
+ public:
+  void EncodeWith(MetadataTrait, const typename MetadataTrait::ValueType& value,
+                  Encoder* encoder) {
+    if (value != known_value) {
+      gpr_log(GPR_ERROR, "%s",
+              absl::StrCat("Not encoding bad ", MetadataTrait::key(), " header")
+                  .c_str());
+      return;
+    }
+    Slice encoded(MetadataTrait::Encode(known_value));
+    const auto encoded_length = encoded.length();
+    encoder->EncodeAlwaysIndexed(&previously_sent_index_, MetadataTrait::key(),
+                                 std::move(encoded),
+                                 MetadataTrait::key().size() + encoded_length +
+                                     hpack_constants::kEntryOverhead);
+  }
+
+ private:
+  uint32_t previously_sent_index_ = 0;
+};
+template <typename MetadataTrait, size_t N>
+class Compressor<MetadataTrait, SmallIntegralValuesCompressor<N>> {
+ public:
+  void EncodeWith(MetadataTrait, const typename MetadataTrait::ValueType& value,
+                  Encoder* encoder) {
+    uint32_t* index = nullptr;
+    auto& table = encoder->hpack_table();
+    if (static_cast<size_t>(value) < N) {
+      index = &previously_sent_[static_cast<uint32_t>(value)];
+      if (table.ConvertableToDynamicIndex(*index)) {
+        encoder->EmitIndexed(table.DynamicIndex(*index));
+        return;
+      }
+    }
+    auto key = Slice::FromStaticString(MetadataTrait::key());
+    auto encoded_value = MetadataTrait::Encode(value);
+    size_t transport_length =
+        key.length() + encoded_value.length() + hpack_constants::kEntryOverhead;
+    if (index != nullptr) {
+      *index = table.AllocateIndex(transport_length);
+      encoder->EmitLitHdrWithNonBinaryStringKeyIncIdx(std::move(key),
+                                                      std::move(encoded_value));
+    } else {
+      encoder->EmitLitHdrWithNonBinaryStringKeyNotIdx(std::move(key),
+                                                      std::move(encoded_value));
+    }
+  }
+
+ private:
+  uint32_t previously_sent_[N] = {};
+};
+
+class SliceIndex {
+ public:
+  void EmitTo(absl::string_view key, const Slice& value, Encoder* encoder);
+
+ private:
+  struct ValueIndex {
+    ValueIndex(Slice value, uint32_t index)
+        : value(std::move(value)), index(index) {}
+    Slice value;
+    uint32_t index;
+  };
+  std::vector<ValueIndex> values_;
+};
+
+template <typename MetadataTrait>
+class Compressor<MetadataTrait, SmallSetOfValuesCompressor> {
+ public:
+  void EncodeWith(MetadataTrait, const Slice& value, Encoder* encoder) {
+    index_.EmitTo(MetadataTrait::key(), value, encoder);
+  }
+
+ private:
+  SliceIndex index_;
+};
+
+struct PreviousTimeout {
+  Timeout timeout;
+  uint32_t index;
+};
+
+class TimeoutCompressorImpl {
+ public:
+  void EncodeWith(absl::string_view key, Timestamp deadline, Encoder* encoder);
+
+ private:
+  std::vector<PreviousTimeout> previous_timeouts_;
+};
+
+template <typename MetadataTrait>
+class Compressor<MetadataTrait, TimeoutCompressor>
+    : public TimeoutCompressorImpl {
+ public:
+  void EncodeWith(MetadataTrait, const typename MetadataTrait::ValueType& value,
+                  Encoder* encoder) {
+    TimeoutCompressorImpl::EncodeWith(MetadataTrait::key(), value, encoder);
+  }
+};
+
+template <>
+class Compressor<HttpStatusMetadata, HttpStatusCompressor> {
+ public:
+  void EncodeWith(HttpStatusMetadata, uint32_t status, Encoder* encoder);
+};
+
+template <>
+class Compressor<HttpMethodMetadata, HttpMethodCompressor> {
+ public:
+  void EncodeWith(HttpMethodMetadata, HttpMethodMetadata::ValueType method,
+                  Encoder* encoder);
+};
+
+template <>
+class Compressor<HttpSchemeMetadata, HttpSchemeCompressor> {
+ public:
+  void EncodeWith(HttpSchemeMetadata, HttpSchemeMetadata::ValueType value,
+                  Encoder* encoder);
+};
+
+}  // namespace hpack_encoder_detail
+
 class HPackCompressor {
   class SliceIndex;
 
@@ -75,87 +348,22 @@ class HPackCompressor {
   void EncodeHeaders(const EncodeHeaderOptions& options,
                      const HeaderSet& headers, grpc_slice_buffer* output) {
     SliceBuffer raw;
-    Encoder encoder(this, options.use_true_binary_metadata, raw);
+    hpack_encoder_detail::Encoder encoder(
+        this, options.use_true_binary_metadata, raw);
     headers.Encode(&encoder);
     Frame(options, raw, output);
   }
 
   template <typename HeaderSet>
   void EncodeRawHeaders(const HeaderSet& headers, SliceBuffer& output) {
-    Encoder encoder(this, true, output);
+    hpack_encoder_detail::Encoder encoder(this, true, output);
     headers.Encode(&encoder);
   }
 
  private:
-  class Encoder {
-   public:
-    Encoder(HPackCompressor* compressor, bool use_true_binary_metadata,
-            SliceBuffer& output);
-
-    void Encode(const Slice& key, const Slice& value);
-    void Encode(HttpPathMetadata, const Slice& value);
-    void Encode(HttpAuthorityMetadata, const Slice& value);
-    void Encode(HttpStatusMetadata, uint32_t status);
-    void Encode(GrpcTimeoutMetadata, Timestamp deadline);
-    void Encode(TeMetadata, TeMetadata::ValueType value);
-    void Encode(ContentTypeMetadata, ContentTypeMetadata::ValueType value);
-    void Encode(HttpSchemeMetadata, HttpSchemeMetadata::ValueType value);
-    void Encode(HttpMethodMetadata, HttpMethodMetadata::ValueType method);
-    void Encode(UserAgentMetadata, const Slice& slice);
-    void Encode(GrpcStatusMetadata, grpc_status_code status);
-    void Encode(GrpcEncodingMetadata, grpc_compression_algorithm value);
-    void Encode(GrpcAcceptEncodingMetadata, CompressionAlgorithmSet value);
-    void Encode(GrpcTagsBinMetadata, const Slice& slice);
-    void Encode(GrpcTraceBinMetadata, const Slice& slice);
-    void Encode(GrpcMessageMetadata, const Slice& slice) {
-      if (slice.empty()) return;
-      EmitLitHdrWithNonBinaryStringKeyNotIdx(
-          Slice::FromStaticString("grpc-message"), slice.Ref());
-    }
-    template <typename Which>
-    void Encode(Which, const typename Which::ValueType& value) {
-      const Slice& slice = MetadataValueAsSlice<Which>(value);
-      if (absl::EndsWith(Which::key(), "-bin")) {
-        EmitLitHdrWithBinaryStringKeyNotIdx(
-            Slice::FromStaticString(Which::key()), slice.Ref());
-      } else {
-        EmitLitHdrWithNonBinaryStringKeyNotIdx(
-            Slice::FromStaticString(Which::key()), slice.Ref());
-      }
-    }
-
-   private:
-    friend class SliceIndex;
-
-    void AdvertiseTableSizeChange();
-    void EmitIndexed(uint32_t index);
-    void EmitLitHdrWithNonBinaryStringKeyIncIdx(Slice key_slice,
-                                                Slice value_slice);
-    void EmitLitHdrWithBinaryStringKeyIncIdx(Slice key_slice,
-                                             Slice value_slice);
-    void EmitLitHdrWithBinaryStringKeyNotIdx(Slice key_slice,
-                                             Slice value_slice);
-    void EmitLitHdrWithBinaryStringKeyNotIdx(uint32_t key_index,
-                                             Slice value_slice);
-    void EmitLitHdrWithNonBinaryStringKeyNotIdx(Slice key_slice,
-                                                Slice value_slice);
-
-    void EncodeAlwaysIndexed(uint32_t* index, absl::string_view key,
-                             Slice value, size_t transport_length);
-    void EncodeIndexedKeyWithBinaryValue(uint32_t* index, absl::string_view key,
-                                         Slice value);
-
-    void EncodeRepeatingSliceValue(const absl::string_view& key,
-                                   const Slice& slice, uint32_t* index,
-                                   size_t max_compression_size);
-
-    const bool use_true_binary_metadata_;
-    HPackCompressor* const compressor_;
-    SliceBuffer& output_;
-  };
-
   static constexpr size_t kNumFilterValues = 64;
   static constexpr uint32_t kNumCachedGrpcStatusValues = 16;
+  friend class hpack_encoder_detail::Encoder;
 
   void Frame(const EncodeHeaderOptions& options, SliceBuffer& raw,
              grpc_slice_buffer* output);
@@ -168,49 +376,23 @@ class HPackCompressor {
   bool advertise_table_size_change_ = false;
   HPackEncoderTable table_;
 
-  class SliceIndex {
+  grpc_metadata_batch::StatefulCompressor<hpack_encoder_detail::Compressor>
-   public:
+      compression_state_;
-    void EmitTo(absl::string_view key, const Slice& value, Encoder* encoder);
+};
-
-   private:
-    struct ValueIndex {
-      ValueIndex(Slice value, uint32_t index)
-          : value(std::move(value)), index(index) {}
-      Slice value;
-      uint32_t index;
-    };
-    std::vector<ValueIndex> values_;
-  };
 
-  struct PreviousTimeout {
+namespace hpack_encoder_detail {
-    Timeout timeout;
-    uint32_t index;
-  };
 
-  // Index into table_ for the te:trailers metadata element
+template <typename MetadataTrait>
-  uint32_t te_index_ = 0;
+void Encoder::Encode(MetadataTrait,
-  // Index into table_ for the content-type metadata element
+                     const typename MetadataTrait::ValueType& value) {
-  uint32_t content_type_index_ = 0;
+  compressor_->compression_state_
-  // Index into table_ for the user-agent metadata element
+      .Compressor<MetadataTrait, typename MetadataTrait::CompressionTraits>::
-  uint32_t user_agent_index_ = 0;
+          EncodeWith(MetadataTrait(), value, this);
-  // Cached grpc-status values
+}
-  uint32_t cached_grpc_status_[kNumCachedGrpcStatusValues] = {};
+
-  // Cached grpc-encoding values
+inline HPackEncoderTable& Encoder::hpack_table() { return compressor_->table_; }
-  uint32_t cached_grpc_encoding_[GRPC_COMPRESS_ALGORITHMS_COUNT] = {};
+
-  // Cached grpc-accept-encoding value
+}  // namespace hpack_encoder_detail
-  uint32_t grpc_accept_encoding_index_ = 0;
-  // The grpc-accept-encoding string referred to by grpc_accept_encoding_index_
-  CompressionAlgorithmSet grpc_accept_encoding_;
-  // Index of something that was sent with grpc-tags-bin
-  uint32_t grpc_tags_bin_index_ = 0;
-  // Index of something that was sent with grpc-trace-bin
-  uint32_t grpc_trace_bin_index_ = 0;
-  // The user-agent string referred to by user_agent_index_
-  Slice user_agent_;
-  SliceIndex path_index_;
-  SliceIndex authority_index_;
-  std::vector<PreviousTimeout> previous_timeouts_;
-};
 
 }  // namespace grpc_core
 

src/core/lib/transport/metadata_batch.h

@@ -51,6 +51,50 @@
 
 namespace grpc_core {
 
+///////////////////////////////////////////////////////////////////////////////
+// Compression traits.
+//
+// Each metadata trait exposes exactly one compression trait.
+// This type directs how transports might choose to compress the metadata.
+// Adding a value here typically involves editing all transports to support the
+// trait, and so should not be done lightly.
+
+// No compression.
+struct NoCompressionCompressor {};
+
+// Expect a single value for this metadata key, but we don't know apriori its
+// value.
+// It's ok if it changes over time, but it should be mostly stable.
+// This is used for things like user-agent, which is expected to be the same
+// for all requests.
+struct StableValueCompressor {};
+
+// Expect a single value for this metadata key, and we know apriori its value.
+template <typename T, T value>
+struct KnownValueCompressor {};
+
+// Values are uncompressible, but expect the key to be in most requests and try
+// and compress that.
+struct FrequentKeyWithNoValueCompressionCompressor {};
+
+// Expect a small set of values for this metadata key.
+struct SmallSetOfValuesCompressor {};
+
+// Expect integral values up to N for this metadata key.
+template <size_t N>
+struct SmallIntegralValuesCompressor {};
+
+// Specialty compressor for grpc-timeout metadata.
+struct TimeoutCompressor {};
+
+// Specialty compressors for HTTP/2 psuedo headers.
+struct HttpSchemeCompressor {};
+struct HttpMethodCompressor {};
+struct HttpStatusCompressor {};
+
+///////////////////////////////////////////////////////////////////////////////
+// Metadata traits
+
 // Given a metadata key and a value, return the encoded size.
 // Defaults to calling the key's Encode() method and then calculating the size
 // of that, but can be overridden for specific keys if there's a better way of
@@ -72,6 +116,7 @@ struct GrpcTimeoutMetadata {
   static constexpr bool kRepeatable = false;
   using ValueType = Timestamp;
   using MementoType = Duration;
+  using CompressionTraits = TimeoutCompressor;
   static absl::string_view key() { return "grpc-timeout"; }
   static MementoType ParseMemento(Slice value, MetadataParseErrorFn on_error);
   static ValueType MementoToValue(MementoType timeout);
@@ -91,6 +136,7 @@ struct TeMetadata {
     kInvalid,
   };
   using MementoType = ValueType;
+  using CompressionTraits = KnownValueCompressor<ValueType, kTrailers>;
   static absl::string_view key() { return "te"; }
   static MementoType ParseMemento(Slice value, MetadataParseErrorFn on_error);
   static ValueType MementoToValue(MementoType te) { return te; }
@@ -118,6 +164,7 @@ struct ContentTypeMetadata {
     kInvalid,
   };
   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 ValueType MementoToValue(MementoType content_type) {
@@ -140,6 +187,7 @@ struct HttpSchemeMetadata {
     kInvalid,
   };
   using MementoType = ValueType;
+  using CompressionTraits = HttpSchemeCompressor;
   static absl::string_view key() { return ":scheme"; }
   static MementoType ParseMemento(Slice value, MetadataParseErrorFn on_error) {
     return Parse(value.as_string_view(), on_error);
@@ -168,6 +216,7 @@ struct HttpMethodMetadata {
     kInvalid,
   };
   using MementoType = ValueType;
+  using CompressionTraits = HttpMethodCompressor;
   static absl::string_view key() { return ":method"; }
   static MementoType ParseMemento(Slice value, MetadataParseErrorFn on_error);
   static ValueType MementoToValue(MementoType content_type) {
@@ -204,12 +253,15 @@ struct CompressionAlgorithmBasedMetadata {
 // grpc-encoding metadata trait.
 struct GrpcEncodingMetadata : public CompressionAlgorithmBasedMetadata {
   static constexpr bool kRepeatable = false;
+  using CompressionTraits =
+      SmallIntegralValuesCompressor<GRPC_COMPRESS_ALGORITHMS_COUNT>;
   static absl::string_view key() { return "grpc-encoding"; }
 };
 
 // grpc-internal-encoding-request metadata trait.
 struct GrpcInternalEncodingRequest : public CompressionAlgorithmBasedMetadata {
   static constexpr bool kRepeatable = false;
+  using CompressionTraits = NoCompressionCompressor;
   static absl::string_view key() { return "grpc-internal-encoding-request"; }
 };
 
@@ -219,6 +271,7 @@ struct GrpcAcceptEncodingMetadata {
   static absl::string_view key() { return "grpc-accept-encoding"; }
   using ValueType = CompressionAlgorithmSet;
   using MementoType = ValueType;
+  using CompressionTraits = StableValueCompressor;
   static MementoType ParseMemento(Slice value, MetadataParseErrorFn) {
     return CompressionAlgorithmSet::FromString(value.as_string_view());
   }
@@ -233,54 +286,63 @@ struct GrpcAcceptEncodingMetadata {
 // user-agent metadata trait.
 struct UserAgentMetadata : public SimpleSliceBasedMetadata {
   static constexpr bool kRepeatable = false;
+  using CompressionTraits = StableValueCompressor;
   static absl::string_view key() { return "user-agent"; }
 };
 
 // grpc-message metadata trait.
 struct GrpcMessageMetadata : public SimpleSliceBasedMetadata {
   static constexpr bool kRepeatable = false;
+  using CompressionTraits = NoCompressionCompressor;
   static absl::string_view key() { return "grpc-message"; }
 };
 
 // host metadata trait.
 struct HostMetadata : public SimpleSliceBasedMetadata {
   static constexpr bool kRepeatable = false;
+  using CompressionTraits = NoCompressionCompressor;
   static absl::string_view key() { return "host"; }
 };
 
 // endpoint-load-metrics-bin metadata trait.
 struct EndpointLoadMetricsBinMetadata : public SimpleSliceBasedMetadata {
   static constexpr bool kRepeatable = false;
+  using CompressionTraits = NoCompressionCompressor;
   static absl::string_view key() { return "endpoint-load-metrics-bin"; }
 };
 
 // grpc-server-stats-bin metadata trait.
 struct GrpcServerStatsBinMetadata : public SimpleSliceBasedMetadata {
   static constexpr bool kRepeatable = false;
+  using CompressionTraits = NoCompressionCompressor;
   static absl::string_view key() { return "grpc-server-stats-bin"; }
 };
 
 // grpc-trace-bin metadata trait.
 struct GrpcTraceBinMetadata : public SimpleSliceBasedMetadata {
   static constexpr bool kRepeatable = false;
+  using CompressionTraits = FrequentKeyWithNoValueCompressionCompressor;
   static absl::string_view key() { return "grpc-trace-bin"; }
 };
 
 // grpc-tags-bin metadata trait.
 struct GrpcTagsBinMetadata : public SimpleSliceBasedMetadata {
   static constexpr bool kRepeatable = false;
+  using CompressionTraits = FrequentKeyWithNoValueCompressionCompressor;
   static absl::string_view key() { return "grpc-tags-bin"; }
 };
 
 // :authority metadata trait.
 struct HttpAuthorityMetadata : public SimpleSliceBasedMetadata {
   static constexpr bool kRepeatable = false;
+  using CompressionTraits = SmallSetOfValuesCompressor;
   static absl::string_view key() { return ":authority"; }
 };
 
 // :path metadata trait.
 struct HttpPathMetadata : public SimpleSliceBasedMetadata {
   static constexpr bool kRepeatable = false;
+  using CompressionTraits = SmallSetOfValuesCompressor;
   static absl::string_view key() { return ":path"; }
 };
 
@@ -314,6 +376,7 @@ struct SimpleIntBasedMetadata : public SimpleIntBasedMetadataBase<Int> {
 struct GrpcStatusMetadata
     : public SimpleIntBasedMetadata<grpc_status_code, GRPC_STATUS_UNKNOWN> {
   static constexpr bool kRepeatable = false;
+  using CompressionTraits = SmallIntegralValuesCompressor<16>;
   static absl::string_view key() { return "grpc-status"; }
 };
 
@@ -321,6 +384,7 @@ struct GrpcStatusMetadata
 struct GrpcPreviousRpcAttemptsMetadata
     : public SimpleIntBasedMetadata<uint32_t, 0> {
   static constexpr bool kRepeatable = false;
+  using CompressionTraits = NoCompressionCompressor;
   static absl::string_view key() { return "grpc-previous-rpc-attempts"; }
 };
 
@@ -330,6 +394,7 @@ struct GrpcRetryPushbackMsMetadata {
   static absl::string_view key() { return "grpc-retry-pushback-ms"; }
   using ValueType = Duration;
   using MementoType = Duration;
+  using CompressionTraits = NoCompressionCompressor;
   static ValueType MementoToValue(MementoType x) { return x; }
   static Slice Encode(Duration x) { return Slice::FromInt64(x.millis()); }
   static int64_t DisplayValue(Duration x) { return x.millis(); }
@@ -341,6 +406,7 @@ struct GrpcRetryPushbackMsMetadata {
 // TODO(ctiller): consider moving to uint16_t
 struct HttpStatusMetadata : public SimpleIntBasedMetadata<uint32_t, 0> {
   static constexpr bool kRepeatable = false;
+  using CompressionTraits = HttpStatusCompressor;
   static absl::string_view key() { return ":status"; }
 };
 
@@ -353,6 +419,7 @@ struct GrpcLbClientStatsMetadata {
   static absl::string_view key() { return "grpclb_client_stats"; }
   using ValueType = GrpcLbClientStats*;
   using MementoType = ValueType;
+  using CompressionTraits = NoCompressionCompressor;
   static ValueType MementoToValue(MementoType value) { return value; }
   static Slice Encode(ValueType) { abort(); }
   static const char* DisplayValue(ValueType) { return "<internal-lb-stats>"; }
@@ -372,6 +439,7 @@ inline size_t EncodedSizeOfKey(GrpcLbClientStatsMetadata,
 // lb-token metadata
 struct LbTokenMetadata : public SimpleSliceBasedMetadata {
   static constexpr bool kRepeatable = false;
+  using CompressionTraits = NoCompressionCompressor;
   static absl::string_view key() { return "lb-token"; }
 };
 
@@ -384,6 +452,7 @@ struct LbCostBinMetadata {
     std::string name;
   };
   using MementoType = ValueType;
+  using CompressionTraits = NoCompressionCompressor;
   static ValueType MementoToValue(MementoType value) { return value; }
   static Slice Encode(const ValueType& x);
   static std::string DisplayValue(ValueType x);
@@ -978,6 +1047,35 @@ class UnknownMap {
   ChunkedVector<std::pair<Slice, Slice>, 10> unknown_;
 };
 
+// Given a factory template Factory, construct a type that derives from
+// Factory<MetadataTrait, MetadataTrait::CompressionTraits> for all
+// MetadataTraits. Useful for transports in defining the stateful parts of their
+// compression algorithm.
+template <template <typename, typename> class Factory,
+          typename... MetadataTraits>
+struct StatefulCompressor;
+
+template <template <typename, typename> class Factory, typename MetadataTrait,
+          bool kEncodable = IsEncodableTrait<MetadataTrait>::value>
+struct SpecificStatefulCompressor;
+
+template <template <typename, typename> class Factory, typename MetadataTrait>
+struct SpecificStatefulCompressor<Factory, MetadataTrait, true>
+    : public Factory<MetadataTrait, typename MetadataTrait::CompressionTraits> {
+};
+
+template <template <typename, typename> class Factory, typename MetadataTrait>
+struct SpecificStatefulCompressor<Factory, MetadataTrait, false> {};
+
+template <template <typename, typename> class Factory, typename MetadataTrait,
+          typename... MetadataTraits>
+struct StatefulCompressor<Factory, MetadataTrait, MetadataTraits...>
+    : public SpecificStatefulCompressor<Factory, MetadataTrait>,
+      public StatefulCompressor<Factory, MetadataTraits...> {};
+
+template <template <typename, typename> class Factory>
+struct StatefulCompressor<Factory> {};
+
 }  // namespace metadata_detail
 
 // Helper function for encoders
@@ -1096,6 +1194,15 @@ class MetadataMap {
   explicit MetadataMap(Arena* arena);
   ~MetadataMap();
 
+  // Given a compressor factory - template taking <MetadataTrait,
+  // CompressionTrait>, StatefulCompressor<Factory> provides a type
+  // derived from all Encodable traits in this MetadataMap.
+  // This can be used by transports to delegate compression to the appropriate
+  // compression algorithm.
+  template <template <typename, typename> class Factory>
+  using StatefulCompressor =
+      metadata_detail::StatefulCompressor<Factory, Traits...>;
+
   MetadataMap(const MetadataMap&) = delete;
   MetadataMap& operator=(const MetadataMap&) = delete;
   MetadataMap(MetadataMap&&) noexcept;

src/core/lib/transport/parsed_metadata.h

@@ -26,6 +26,7 @@
 
 #include "absl/functional/function_ref.h"
 #include "absl/meta/type_traits.h"
+#include "absl/strings/escaping.h"
 #include "absl/strings/match.h"
 #include "absl/strings/str_cat.h"
 #include "absl/strings/string_view.h"
@@ -395,12 +396,17 @@ ParsedMetadata<MetadataContainer>::KeyValueVTable(absl::string_view key) {
     return absl::StrCat(p->first.as_string_view(), ": ",
                         p->second.as_string_view());
   };
+  static const auto binary_debug_string = [](const Buffer& value) {
+    auto* p = static_cast<KV*>(value.pointer);
+    return absl::StrCat(p->first.as_string_view(), ": \"",
+                        absl::CEscape(p->second.as_string_view()), "\"");
+  };
   static const auto key_fn = [](const Buffer& value) {
     return static_cast<KV*>(value.pointer)->first.as_string_view();
   };
   static const VTable vtable[2] = {
       {false, destroy, set, with_new_value, debug_string, "", key_fn},
-      {true, destroy, set, with_new_value, debug_string, "", key_fn},
+      {true, destroy, set, with_new_value, binary_debug_string, "", key_fn},
   };
   return &vtable[absl::EndsWith(key, "-bin")];
 }