Halve memory usage for remote cache writes. (pantsbuild#12083)

Previously we were inadvertantly double copying. We still buffer too
much in the case of large blobs but this does help.

src/rust/engine/fs/store/src/lib.rs

@@ -557,7 +557,7 @@ impl Store {
                 })
                 .await?;
               match maybe_bytes {
-                Some(bytes) => remote.store_bytes(&bytes).await,
+                Some(bytes) => remote.store_bytes(bytes).await,
                 None => Err(format!(
                   "Failed to upload digest {:?}: Not found in local store",
                   digest

src/rust/engine/fs/store/src/remote.rs

@@ -88,7 +88,7 @@ impl ByteStore {
     })
   }
 
-  pub async fn store_bytes(&self, bytes: &[u8]) -> Result<Digest, String> {
+  pub async fn store_bytes(&self, bytes: Bytes) -> Result<Digest, String> {
     let len = bytes.len();
     let digest = Digest::of_bytes(&bytes);
     let resource_name = format!(
@@ -110,11 +110,6 @@ impl ByteStore {
     let resource_name = resource_name.clone();
     let chunk_size_bytes = store.chunk_size_bytes;
 
-    // NOTE(tonic): The call into the Tonic library wants the slice to last for the 'static
-    // lifetime but the slice passed into this method generally points into the shared memory
-    // of the LMDB store which is on the other side of the FFI boundary.
-    let bytes = Bytes::copy_from_slice(bytes);
-
     let stream = futures::stream::unfold((0, false), move |(offset, has_sent_any)| {
       if offset >= bytes.len() && has_sent_any {
         futures::future::ready(None)

src/rust/engine/fs/store/src/remote_tests.rs

@@ -140,7 +140,7 @@ async fn write_file_one_chunk() {
 
   let store = new_byte_store(&cas);
   assert_eq!(
-    store.store_bytes(&testdata.bytes()).await,
+    store.store_bytes(testdata.bytes()).await,
     Ok(testdata.digest())
   );
 
@@ -168,7 +168,7 @@ async fn write_file_multiple_chunks() {
   let fingerprint = big_file_fingerprint();
 
   assert_eq!(
-    store.store_bytes(&all_the_henries).await,
+    store.store_bytes(all_the_henries.clone()).await,
     Ok(big_file_digest())
   );
 
@@ -198,7 +198,7 @@ async fn write_empty_file() {
 
   let store = new_byte_store(&cas);
   assert_eq!(
-    store.store_bytes(&empty_file.bytes()).await,
+    store.store_bytes(empty_file.bytes()).await,
     Ok(empty_file.digest())
   );
 
@@ -215,7 +215,7 @@ async fn write_file_errors() {
 
   let store = new_byte_store(&cas);
   let error = store
-    .store_bytes(&TestData::roland().bytes())
+    .store_bytes(TestData::roland().bytes())
     .await
     .expect_err("Want error");
   assert!(
@@ -238,7 +238,7 @@ async fn write_connection_error() {
   )
   .unwrap();
   let error = store
-    .store_bytes(&TestData::roland().bytes())
+    .store_bytes(TestData::roland().bytes())
     .await
     .expect_err("Want error");
   assert!(