Avoid unnecessary runtime clones in try_join helpers (#26426)

GitOrigin-RevId: d5ad8eebd2e5d63708b34a652226f92dc8d72d59

Author: sjudd

crates/application/src/export_worker.rs

@@ -313,7 +313,7 @@ impl<RT: Runtime> ExportWorker<RT> {
     }
 
     async fn upload_tables(
-        runtime: RT,
+        runtime: &RT,
         storage: Arc<dyn Storage>,
         format: ExportFormat,
         tables: BTreeMap<TabletId, (TableNumber, TableName, TableSummary)>,
@@ -398,7 +398,7 @@ impl<RT: Runtime> ExportWorker<RT> {
             },
             ExportFormat::CleanJsonl | ExportFormat::InternalJson => {
                 let mut table_uploads = Self::upload_tables(
-                    self.runtime.clone(),
+                    &self.runtime,
                     self.storage.clone(),
                     format,
                     tables.clone(),
@@ -440,8 +440,7 @@ impl<RT: Runtime> ExportWorker<RT> {
                         }
                     });
                 let table_object_keys: BTreeMap<_, _> =
-                    try_join_buffered(self.runtime.clone(), "table_uploads", complete_upload_futs)
-                        .await?;
+                    try_join_buffered(&self.runtime, "table_uploads", complete_upload_futs).await?;
                 tracing::info!(
                     "Export succeeded! {} snapshots written to storage. Format: {format:?}",
                     tablet_ids.len()

crates/common/src/runtime/mod.rs

@@ -106,14 +106,14 @@ pub async fn try_join_buffered<
     T: Send + 'static,
     C: Default + Send + 'static + Extend<T>,
 >(
-    rt: RT,
+    rt: &RT,
     name: &'static str,
     tasks: impl Iterator<Item = impl Future<Output = anyhow::Result<T>> + Send + 'static>
         + Send
         + 'static,
 ) -> anyhow::Result<C> {
     assert_send(
-        stream::iter(tasks.map(|task| assert_send(try_join(rt.clone(), name, assert_send(task)))))
+        stream::iter(tasks.map(|task| assert_send(try_join(rt, name, assert_send(task)))))
             .buffered(JOIN_BUFFER_SIZE)
             .try_collect(),
     )
@@ -134,22 +134,22 @@ pub async fn try_join_buffer_unordered<
     T: Send + 'static,
     C: Default + Send + 'static + Extend<T>,
 >(
-    rt: RT,
+    rt: &RT,
     name: &'static str,
     tasks: impl Iterator<Item = impl Future<Output = anyhow::Result<T>> + Send + 'static>
         + Send
         + 'static,
 ) -> anyhow::Result<C> {
     assert_send(
-        stream::iter(tasks.map(|task| try_join(rt.clone(), name, task)))
+        stream::iter(tasks.map(|task| try_join(rt, name, task)))
             .buffer_unordered(JOIN_BUFFER_SIZE)
             .try_collect(),
     )
     .await
 }
 
 pub async fn try_join<RT: Runtime, T: Send + 'static>(
-    rt: RT,
+    rt: &RT,
     name: &'static str,
     fut: impl Future<Output = anyhow::Result<T>> + Send + 'static,
 ) -> anyhow::Result<T> {

crates/database/src/index_workers/index_meta.rs

@@ -122,13 +122,13 @@ pub trait SearchIndex: Clone + Debug {
         Self: Sized;
 
     async fn download_previous_segments<RT: Runtime>(
-        rt: RT,
+        rt: &RT,
         storage: Arc<dyn Storage>,
         segment: Vec<Self::Segment>,
     ) -> anyhow::Result<Self::PreviousSegments>;
 
     async fn upload_previous_segments<RT: Runtime>(
-        rt: RT,
+        rt: &RT,
         storage: Arc<dyn Storage>,
         segments: Self::PreviousSegments,
     ) -> anyhow::Result<Vec<Self::Segment>>;

crates/database/src/index_workers/search_flusher.rs

@@ -552,7 +552,7 @@ impl<RT: Runtime, T: SearchIndexConfigParser + 'static> SearchFlusher<RT, T> {
         };
 
         let mut mutable_previous_segments = T::IndexType::download_previous_segments(
-            params.runtime.clone(),
+            &params.runtime,
             params.storage.clone(),
             previous_segments,
         )
@@ -575,7 +575,7 @@ impl<RT: Runtime, T: SearchIndexConfigParser + 'static> SearchFlusher<RT, T> {
         .await?;
 
         let updated_previous_segments = T::IndexType::upload_previous_segments(
-            params.runtime.clone(),
+            &params.runtime,
             params.storage,
             mutable_previous_segments,
         )

crates/database/src/index_workers/writer.rs

@@ -567,8 +567,7 @@ impl<RT: Runtime, T: SearchIndex> Inner<RT, T> {
             ),
         );
         let mut previous_segments =
-            T::download_previous_segments(runtime.clone(), storage.clone(), segments_to_update)
-                .await?;
+            T::download_previous_segments(&runtime, storage.clone(), segments_to_update).await?;
         let mut documents = database.load_documents_in_table(
             *index_name.table(),
             TimestampRange::new((Bound::Excluded(start_ts), Bound::Included(current_ts)))?,
@@ -582,6 +581,6 @@ impl<RT: Runtime, T: SearchIndex> Inner<RT, T> {
             }
         }
 
-        T::upload_previous_segments(runtime, storage, previous_segments).await
+        T::upload_previous_segments(&runtime, storage, previous_segments).await
     }
 }

crates/database/src/search_and_vector_bootstrap.rs

@@ -539,8 +539,7 @@ impl<RT: Runtime> SearchAndVectorIndexBootstrapWorker<RT> {
                 }
             });
         let indexes_with_fast_forward_ts =
-            try_join_buffer_unordered(self.runtime.clone(), "get_index_futs", get_index_futs)
-                .await?;
+            try_join_buffer_unordered(&self.runtime, "get_index_futs", get_index_futs).await?;
         let indexes_to_bootstrap = IndexesToBootstrap::create(
             self.persistence.upper_bound(),
             indexes_with_fast_forward_ts,

crates/database/src/text_index_worker/text_meta.rs

@@ -169,7 +169,7 @@ impl SearchIndex for TextSearchIndex {
     }
 
     async fn download_previous_segments<RT: Runtime>(
-        rt: RT,
+        rt: &RT,
         storage: Arc<dyn Storage>,
         segments: Vec<Self::Segment>,
     ) -> anyhow::Result<Self::PreviousSegments> {
@@ -189,7 +189,7 @@ impl SearchIndex for TextSearchIndex {
     }
 
     async fn upload_previous_segments<RT: Runtime>(
-        rt: RT,
+        rt: &RT,
         storage: Arc<dyn Storage>,
         segments: Self::PreviousSegments,
     ) -> anyhow::Result<Vec<Self::Segment>> {

crates/database/src/vector_index_worker/vector_meta.rs

@@ -181,7 +181,7 @@ impl SearchIndex for VectorSearchIndex {
     }
 
     async fn download_previous_segments<RT: Runtime>(
-        rt: RT,
+        rt: &RT,
         storage: Arc<dyn Storage>,
         segments: Vec<Self::Segment>,
     ) -> anyhow::Result<Self::PreviousSegments> {
@@ -197,7 +197,7 @@ impl SearchIndex for VectorSearchIndex {
     }
 
     async fn upload_previous_segments<RT: Runtime>(
-        rt: RT,
+        rt: &RT,
         storage: Arc<dyn Storage>,
         segments: Self::PreviousSegments,
     ) -> anyhow::Result<Vec<Self::Segment>> {

crates/search/src/incremental_index.rs

@@ -472,8 +472,7 @@ async fn get_all_segment_term_metadata<RT: Runtime>(
         },
     );
     let segments_term_metadata: Vec<_> =
-        try_join_buffer_unordered(rt.clone(), "text_term_metadata", segment_term_metadata_futs)
-            .await?;
+        try_join_buffer_unordered(rt, "text_term_metadata", segment_term_metadata_futs).await?;
     Ok(segments_term_metadata)
 }
 
@@ -544,7 +543,7 @@ pub async fn fetch_compact_and_upload_text_segment<RT: Runtime>(
 ) -> anyhow::Result<FragmentedTextSegment> {
     let _storage = storage.clone();
     let opened_segments = try_join_buffer_unordered(
-        rt.clone(),
+        rt,
         "text_segment_merge",
         segments.into_iter().map(move |segment| {
             let pool = blocking_thread_pool.clone();