Fix retry code to hold refs to send_initial_metadata slices (grpc#27205)

* add test proving that we fail to take refs to send_initial_metadart payload

* fix grpc_slice_from_copied_string() to take refs and grpc_metadata_batch_copy() to copy the mdelems when necessary

* fix criteria used to determine if mdelem is reffable

* add support for inline slices

* fix sanity

src/core/lib/slice/slice.cc

@@ -223,11 +223,21 @@ grpc_core::UnmanagedMemorySlice::UnmanagedMemorySlice(const char* source)
                                                             strlen(source)) {}
 
 grpc_slice grpc_slice_from_copied_buffer(const char* source, size_t length) {
-  return grpc_core::UnmanagedMemorySlice(source, length);
+  grpc_slice slice;
+  if (length <= sizeof(slice.data.inlined.bytes)) {
+    slice.refcount = nullptr;
+    slice.data.inlined.length = length;
+  } else {
+    // Create a ref-counted slice.
+    slice = grpc_core::UnmanagedMemorySlice(
+        length, grpc_core::UnmanagedMemorySlice::ForceHeapAllocation());
+  }
+  memcpy(GRPC_SLICE_START_PTR(slice), source, length);
+  return slice;
 }
 
 grpc_slice grpc_slice_from_copied_string(const char* source) {
-  return grpc_core::UnmanagedMemorySlice(source, strlen(source));
+  return grpc_slice_from_copied_buffer(source, strlen(source));
 }
 
 grpc_slice grpc_slice_from_moved_buffer(grpc_core::UniquePtr<char> p,

src/core/lib/transport/metadata_batch.cc

@@ -401,9 +401,20 @@ void grpc_metadata_batch_copy(grpc_metadata_batch* src,
   size_t i = 0;
   for (grpc_linked_mdelem* elem = src->list.head; elem != nullptr;
        elem = elem->next) {
+    // If the mdelem is not external, take a ref.
+    // Otherwise, create a new copy, holding its own refs to the
+    // underlying slices.
+    grpc_mdelem md;
+    if (GRPC_MDELEM_STORAGE(elem->md) != GRPC_MDELEM_STORAGE_EXTERNAL) {
+      md = GRPC_MDELEM_REF(elem->md);
+    } else {
+      md = grpc_mdelem_from_slices(
+          grpc_slice_ref_internal(GRPC_MDKEY(elem->md)),
+          grpc_slice_ref_internal(GRPC_MDVALUE(elem->md)));
+    }
     // Error unused in non-debug builds.
-    grpc_error_handle GRPC_UNUSED error = grpc_metadata_batch_add_tail(
-        dst, &storage[i++], GRPC_MDELEM_REF(elem->md));
+    grpc_error_handle GRPC_UNUSED error =
+        grpc_metadata_batch_add_tail(dst, &storage[i++], md);
     // The only way that grpc_metadata_batch_add_tail() can fail is if
     // there's a duplicate entry for a callout.  However, that can't be
     // the case here, because we would not have been allowed to create

src/core/lib/transport/metadata_batch.h

@@ -196,6 +196,13 @@ void grpc_metadata_batch_assert_ok(grpc_metadata_batch* batch);
 
 /// Copies \a src to \a dst.  \a storage must point to an array of
 /// \a grpc_linked_mdelem structs of at least the same size as \a src.
+///
+/// For each mdelem in \a src, if the mdelem is of storage types
+/// GRPC_MDELEM_STORAGE_INTERNED or GRPC_MDELEM_STORAGE_ALLOCATED,
+/// refs the original mdelem for the copy.  Otherwise, makes a new
+/// mdelem that will hold its own refs to the key and value slices.
+///
+/// Currently used only in the retry code.
 void grpc_metadata_batch_copy(grpc_metadata_batch* src,
                               grpc_metadata_batch* dst,
                               grpc_linked_mdelem* storage);

test/core/end2end/generate_tests.bzl

@@ -320,6 +320,9 @@ END2END_TESTS = {
     "retry_recv_trailing_metadata_error": _test_options(
         needs_client_channel = True,
     ),
+    "retry_send_initial_metadata_refs": _test_options(
+        needs_client_channel = True,
+    ),
     "retry_send_op_fails": _test_options(needs_client_channel = True),
     "retry_server_pushback_delay": _test_options(needs_client_channel = True),
     "retry_server_pushback_disabled": _test_options(needs_client_channel = True),