Remove unnecessary uses of DashMap and Arc (#3413)

## Summary

All of the resolver code is run on the main thread, so a lot of the
`Send` bounds and uses of `DashMap` and `Arc` are unnecessary. We could
also switch to using single-threaded versions of `Mutex` and `Notify` in
some places, but there isn't really a crate that provides those I would
be comfortable with using.

The `Arc` in `OnceMap` can't easily be removed because of the uv-auth
code which uses the
[reqwest-middleware](https://docs.rs/reqwest-middleware/latest/reqwest_middleware/trait.Middleware.html)
crate, that seems to adds unnecessary `Send` bounds because of
`async-trait`. We could duplicate the code and create a `OnceMapLocal`
variant, but I don't feel that's worth it.

crates/uv-distribution/src/distribution_database.rs

@@ -1,5 +1,6 @@
 use std::io;
 use std::path::Path;
+use std::rc::Rc;
 use std::sync::Arc;
 
 use futures::{FutureExt, TryStreamExt};
@@ -41,20 +42,20 @@ use crate::{ArchiveMetadata, Error, LocalWheel, Reporter, SourceDistributionBuil
 ///
 /// This struct also has the task of acquiring locks around source dist builds in general and git
 /// operation especially.
-pub struct DistributionDatabase<'a, Context: BuildContext + Send + Sync> {
+pub struct DistributionDatabase<'a, Context: BuildContext> {
     client: &'a RegistryClient,
     build_context: &'a Context,
     builder: SourceDistributionBuilder<'a, Context>,
-    locks: Arc<Locks>,
+    locks: Rc<Locks>,
 }
 
-impl<'a, Context: BuildContext + Send + Sync> DistributionDatabase<'a, Context> {
+impl<'a, Context: BuildContext> DistributionDatabase<'a, Context> {
     pub fn new(client: &'a RegistryClient, build_context: &'a Context) -> Self {
         Self {
             client,
             build_context,
             builder: SourceDistributionBuilder::new(client, build_context),
-            locks: Arc::new(Locks::default()),
+            locks: Rc::new(Locks::default()),
         }
     }
 
@@ -307,7 +308,7 @@ impl<'a, Context: BuildContext + Send + Sync> DistributionDatabase<'a, Context>
         let built_wheel = self
             .builder
             .download_and_build(&BuildableSource::Dist(dist), tags, hashes)
-            .boxed()
+            .boxed_local()
             .await?;
 
         // If the wheel was unzipped previously, respect it. Source distributions are
@@ -360,7 +361,7 @@ impl<'a, Context: BuildContext + Send + Sync> DistributionDatabase<'a, Context>
             return Ok(ArchiveMetadata { metadata, hashes });
         }
 
-        match self.client.wheel_metadata(dist).boxed().await {
+        match self.client.wheel_metadata(dist).boxed_local().await {
             Ok(metadata) => Ok(ArchiveMetadata::from(metadata)),
             Err(err) if err.is_http_streaming_unsupported() => {
                 warn!("Streaming unsupported when fetching metadata for {dist}; downloading wheel directly ({err})");
@@ -404,7 +405,7 @@ impl<'a, Context: BuildContext + Send + Sync> DistributionDatabase<'a, Context>
         let metadata = self
             .builder
             .download_and_build_metadata(source, hashes)
-            .boxed()
+            .boxed_local()
             .await?;
         Ok(metadata)
     }

crates/uv-distribution/src/locks.rs

@@ -1,4 +1,4 @@
-use std::sync::Arc;
+use std::rc::Rc;
 
 use rustc_hash::FxHashMap;
 use tokio::sync::Mutex;
@@ -7,14 +7,14 @@ use distribution_types::{Identifier, ResourceId};
 
 /// A set of locks used to prevent concurrent access to the same resource.
 #[derive(Debug, Default)]
-pub(crate) struct Locks(Mutex<FxHashMap<ResourceId, Arc<Mutex<()>>>>);
+pub(crate) struct Locks(Mutex<FxHashMap<ResourceId, Rc<Mutex<()>>>>);
 
 impl Locks {
     /// Acquire a lock on the given resource.
-    pub(crate) async fn acquire(&self, dist: &impl Identifier) -> Arc<Mutex<()>> {
+    pub(crate) async fn acquire(&self, dist: &impl Identifier) -> Rc<Mutex<()>> {
         let mut map = self.0.lock().await;
         map.entry(dist.resource_id())
-            .or_insert_with(|| Arc::new(Mutex::new(())))
+            .or_insert_with(|| Rc::new(Mutex::new(())))
             .clone()
     }
 }

crates/uv-distribution/src/source/mod.rs

@@ -116,7 +116,7 @@ impl<'a, T: BuildContext> SourceDistributionBuilder<'a, T> {
                                 tags,
                                 hashes,
                             )
-                            .boxed()
+                            .boxed_local()
                             .await;
                     }
                 };
@@ -130,7 +130,7 @@ impl<'a, T: BuildContext> SourceDistributionBuilder<'a, T> {
                     tags,
                     hashes,
                 )
-                .boxed()
+                .boxed_local()
                 .await?
             }
             BuildableSource::Dist(SourceDist::DirectUrl(dist)) => {
@@ -153,18 +153,18 @@ impl<'a, T: BuildContext> SourceDistributionBuilder<'a, T> {
                     tags,
                     hashes,
                 )
-                .boxed()
+                .boxed_local()
                 .await?
             }
             BuildableSource::Dist(SourceDist::Git(dist)) => {
                 self.git(source, &GitSourceUrl::from(dist), tags, hashes)
-                    .boxed()
+                    .boxed_local()
                     .await?
             }
             BuildableSource::Dist(SourceDist::Path(dist)) => {
                 if dist.path.is_dir() {
                     self.source_tree(source, &PathSourceUrl::from(dist), tags, hashes)
-                        .boxed()
+                        .boxed_local()
                         .await?
                 } else {
                     let cache_shard = self
@@ -178,7 +178,7 @@ impl<'a, T: BuildContext> SourceDistributionBuilder<'a, T> {
                         tags,
                         hashes,
                     )
-                    .boxed()
+                    .boxed_local()
                     .await?
                 }
             }
@@ -205,16 +205,18 @@ impl<'a, T: BuildContext> SourceDistributionBuilder<'a, T> {
                     tags,
                     hashes,
                 )
-                .boxed()
+                .boxed_local()
                 .await?
             }
             BuildableSource::Url(SourceUrl::Git(resource)) => {
-                self.git(source, resource, tags, hashes).boxed().await?
+                self.git(source, resource, tags, hashes)
+                    .boxed_local()
+                    .await?
             }
             BuildableSource::Url(SourceUrl::Path(resource)) => {
                 if resource.path.is_dir() {
                     self.source_tree(source, resource, tags, hashes)
-                        .boxed()
+                        .boxed_local()
                         .await?
                 } else {
                     let cache_shard = self.build_context.cache().shard(
@@ -222,7 +224,7 @@ impl<'a, T: BuildContext> SourceDistributionBuilder<'a, T> {
                         WheelCache::Path(resource.url).root(),
                     );
                     self.archive(source, resource, &cache_shard, tags, hashes)
-                        .boxed()
+                        .boxed_local()
                         .await?
                 }
             }
@@ -268,7 +270,7 @@ impl<'a, T: BuildContext> SourceDistributionBuilder<'a, T> {
                                 &cache_shard,
                                 hashes,
                             )
-                            .boxed()
+                            .boxed_local()
                             .await;
                     }
                 };
@@ -281,7 +283,7 @@ impl<'a, T: BuildContext> SourceDistributionBuilder<'a, T> {
                     None,
                     hashes,
                 )
-                .boxed()
+                .boxed_local()
                 .await?
             }
             BuildableSource::Dist(SourceDist::DirectUrl(dist)) => {
@@ -303,18 +305,18 @@ impl<'a, T: BuildContext> SourceDistributionBuilder<'a, T> {
                     subdirectory.as_deref(),
                     hashes,
                 )
-                .boxed()
+                .boxed_local()
                 .await?
             }
             BuildableSource::Dist(SourceDist::Git(dist)) => {
                 self.git_metadata(source, &GitSourceUrl::from(dist), hashes)
-                    .boxed()
+                    .boxed_local()
                     .await?
             }
             BuildableSource::Dist(SourceDist::Path(dist)) => {
                 if dist.path.is_dir() {
                     self.source_tree_metadata(source, &PathSourceUrl::from(dist), hashes)
-                        .boxed()
+                        .boxed_local()
                         .await?
                 } else {
                     let cache_shard = self
@@ -322,7 +324,7 @@ impl<'a, T: BuildContext> SourceDistributionBuilder<'a, T> {
                         .cache()
                         .shard(CacheBucket::BuiltWheels, WheelCache::Path(&dist.url).root());
                     self.archive_metadata(source, &PathSourceUrl::from(dist), &cache_shard, hashes)
-                        .boxed()
+                        .boxed_local()
                         .await?
                 }
             }
@@ -348,16 +350,18 @@ impl<'a, T: BuildContext> SourceDistributionBuilder<'a, T> {
                     subdirectory.as_deref(),
                     hashes,
                 )
-                .boxed()
+                .boxed_local()
                 .await?
             }
             BuildableSource::Url(SourceUrl::Git(resource)) => {
-                self.git_metadata(source, resource, hashes).boxed().await?
+                self.git_metadata(source, resource, hashes)
+                    .boxed_local()
+                    .await?
             }
             BuildableSource::Url(SourceUrl::Path(resource)) => {
                 if resource.path.is_dir() {
                     self.source_tree_metadata(source, resource, hashes)
-                        .boxed()
+                        .boxed_local()
                         .await?
                 } else {
                     let cache_shard = self.build_context.cache().shard(
@@ -365,7 +369,7 @@ impl<'a, T: BuildContext> SourceDistributionBuilder<'a, T> {
                         WheelCache::Path(resource.url).root(),
                     );
                     self.archive_metadata(source, resource, &cache_shard, hashes)
-                        .boxed()
+                        .boxed_local()
                         .await?
                 }
             }
@@ -488,7 +492,7 @@ impl<'a, T: BuildContext> SourceDistributionBuilder<'a, T> {
         // If the backend supports `prepare_metadata_for_build_wheel`, use it.
         if let Some(metadata) = self
             .build_metadata(source, source_dist_entry.path(), subdirectory)
-            .boxed()
+            .boxed_local()
             .await?
         {
             // Store the metadata.
@@ -569,7 +573,7 @@ impl<'a, T: BuildContext> SourceDistributionBuilder<'a, T> {
 
                 Ok(revision.with_hashes(hashes))
             }
-            .boxed()
+            .boxed_local()
             .instrument(info_span!("download", source_dist = %source))
         };
         let req = self.request(url.clone())?;
@@ -706,7 +710,7 @@ impl<'a, T: BuildContext> SourceDistributionBuilder<'a, T> {
         // If the backend supports `prepare_metadata_for_build_wheel`, use it.
         if let Some(metadata) = self
             .build_metadata(source, source_entry.path(), None)
-            .boxed()
+            .boxed_local()
             .await?
         {
             // Store the metadata.
@@ -903,7 +907,7 @@ impl<'a, T: BuildContext> SourceDistributionBuilder<'a, T> {
         // If the backend supports `prepare_metadata_for_build_wheel`, use it.
         if let Some(metadata) = self
             .build_metadata(source, &resource.path, None)
-            .boxed()
+            .boxed_local()
             .await?
         {
             // Store the metadata.
@@ -1110,7 +1114,7 @@ impl<'a, T: BuildContext> SourceDistributionBuilder<'a, T> {
         // If the backend supports `prepare_metadata_for_build_wheel`, use it.
         if let Some(metadata) = self
             .build_metadata(source, fetch.path(), subdirectory.as_deref())
-            .boxed()
+            .boxed_local()
             .await?
         {
             // Store the metadata.