refactor: Remove slot mutex and simplify per blob state

Cargo.toml

@@ -44,8 +44,6 @@ iroh-base = "0.90"
 reflink-copy = "0.1.24"
 irpc = { version = "0.5.0", features = ["rpc", "quinn_endpoint_setup", "message_spans", "stream", "derive"], default-features = false }
 iroh-metrics = { version = "0.35" }
-hashlink = "0.10.0"
-futures-buffered = "0.2.11"
 
 [dev-dependencies]
 clap = { version = "4.5.31", features = ["derive"] }
@@ -60,6 +58,7 @@ testresult = "0.4.1"
 tracing-subscriber = { version = "0.3.19", features = ["fmt"] }
 tracing-test = "0.2.5"
 walkdir = "2.5.0"
+atomic_refcell = "0.1.13"
 
 [features]
 hide-proto-docs = []

src/api/blobs.rs

@@ -127,7 +127,7 @@ impl Blobs {
         .await
     }
 
-    pub fn add_slice(&self, data: impl AsRef<[u8]>) -> AddProgress {
+    pub fn add_slice(&self, data: impl AsRef<[u8]>) -> AddProgress<'_> {
         let options = ImportBytesRequest {
             data: Bytes::copy_from_slice(data.as_ref()),
             format: crate::BlobFormat::Raw,
@@ -136,7 +136,7 @@ impl Blobs {
         self.add_bytes_impl(options)
     }
 
-    pub fn add_bytes(&self, data: impl Into<bytes::Bytes>) -> AddProgress {
+    pub fn add_bytes(&self, data: impl Into<bytes::Bytes>) -> AddProgress<'_> {
         let options = ImportBytesRequest {
             data: data.into(),
             format: crate::BlobFormat::Raw,
@@ -145,7 +145,7 @@ impl Blobs {
         self.add_bytes_impl(options)
     }
 
-    pub fn add_bytes_with_opts(&self, options: impl Into<AddBytesOptions>) -> AddProgress {
+    pub fn add_bytes_with_opts(&self, options: impl Into<AddBytesOptions>) -> AddProgress<'_> {
         let options = options.into();
         let request = ImportBytesRequest {
             data: options.data,
@@ -155,7 +155,7 @@ impl Blobs {
         self.add_bytes_impl(request)
     }
 
-    fn add_bytes_impl(&self, options: ImportBytesRequest) -> AddProgress {
+    fn add_bytes_impl(&self, options: ImportBytesRequest) -> AddProgress<'_> {
         trace!("{options:?}");
         let this = self.clone();
         let stream = Gen::new(|co| async move {
@@ -180,7 +180,7 @@ impl Blobs {
         AddProgress::new(self, stream)
     }
 
-    pub fn add_path_with_opts(&self, options: impl Into<AddPathOptions>) -> AddProgress {
+    pub fn add_path_with_opts(&self, options: impl Into<AddPathOptions>) -> AddProgress<'_> {
         let options = options.into();
         self.add_path_with_opts_impl(ImportPathRequest {
             path: options.path,
@@ -190,7 +190,7 @@ impl Blobs {
         })
     }
 
-    fn add_path_with_opts_impl(&self, options: ImportPathRequest) -> AddProgress {
+    fn add_path_with_opts_impl(&self, options: ImportPathRequest) -> AddProgress<'_> {
         trace!("{:?}", options);
         let client = self.client.clone();
         let stream = Gen::new(|co| async move {
@@ -215,7 +215,7 @@ impl Blobs {
         AddProgress::new(self, stream)
     }
 
-    pub fn add_path(&self, path: impl AsRef<Path>) -> AddProgress {
+    pub fn add_path(&self, path: impl AsRef<Path>) -> AddProgress<'_> {
         self.add_path_with_opts(AddPathOptions {
             path: path.as_ref().to_owned(),
             mode: ImportMode::Copy,
@@ -226,7 +226,7 @@ impl Blobs {
     pub async fn add_stream(
         &self,
         data: impl Stream<Item = io::Result<Bytes>> + Send + Sync + 'static,
-    ) -> AddProgress {
+    ) -> AddProgress<'_> {
         let inner = ImportByteStreamRequest {
             format: crate::BlobFormat::Raw,
             scope: Scope::default(),
@@ -521,7 +521,7 @@ pub struct Batch<'a> {
 }
 
 impl<'a> Batch<'a> {
-    pub fn add_bytes(&self, data: impl Into<Bytes>) -> BatchAddProgress {
+    pub fn add_bytes(&self, data: impl Into<Bytes>) -> BatchAddProgress<'_> {
         let options = ImportBytesRequest {
             data: data.into(),
             format: crate::BlobFormat::Raw,
@@ -530,7 +530,7 @@ impl<'a> Batch<'a> {
         BatchAddProgress(self.blobs.add_bytes_impl(options))
     }
 
-    pub fn add_bytes_with_opts(&self, options: impl Into<AddBytesOptions>) -> BatchAddProgress {
+    pub fn add_bytes_with_opts(&self, options: impl Into<AddBytesOptions>) -> BatchAddProgress<'_> {
         let options = options.into();
         BatchAddProgress(self.blobs.add_bytes_impl(ImportBytesRequest {
             data: options.data,
@@ -539,7 +539,7 @@ impl<'a> Batch<'a> {
         }))
     }
 
-    pub fn add_slice(&self, data: impl AsRef<[u8]>) -> BatchAddProgress {
+    pub fn add_slice(&self, data: impl AsRef<[u8]>) -> BatchAddProgress<'_> {
         let options = ImportBytesRequest {
             data: Bytes::copy_from_slice(data.as_ref()),
             format: crate::BlobFormat::Raw,
@@ -548,7 +548,7 @@ impl<'a> Batch<'a> {
         BatchAddProgress(self.blobs.add_bytes_impl(options))
     }
 
-    pub fn add_path_with_opts(&self, options: impl Into<AddPathOptions>) -> BatchAddProgress {
+    pub fn add_path_with_opts(&self, options: impl Into<AddPathOptions>) -> BatchAddProgress<'_> {
         let options = options.into();
         BatchAddProgress(self.blobs.add_path_with_opts_impl(ImportPathRequest {
             path: options.path,

src/api/remote.rs

@@ -1064,8 +1064,15 @@ mod tests {
     use testresult::TestResult;
 
     use crate::{
+        api::blobs::Blobs,
         protocol::{ChunkRangesSeq, GetRequest},
-        store::fs::{tests::INTERESTING_SIZES, FsStore},
+        store::{
+            fs::{
+                tests::{create_n0_bao, test_data, INTERESTING_SIZES},
+                FsStore,
+            },
+            mem::MemStore,
+        },
         tests::{add_test_hash_seq, add_test_hash_seq_incomplete},
         util::ChunkRangesExt,
     };
@@ -1117,6 +1124,38 @@ mod tests {
         Ok(())
     }
 
+    async fn test_observe_partial(blobs: &Blobs) -> TestResult<()> {
+        let sizes = INTERESTING_SIZES;
+        for size in sizes {
+            let data = test_data(size);
+            let ranges = ChunkRanges::chunk(0);
+            let (hash, bao) = create_n0_bao(&data, &ranges)?;
+            blobs.import_bao_bytes(hash, ranges.clone(), bao).await?;
+            let bitfield = blobs.observe(hash).await?;
+            if size > 1024 {
+                assert_eq!(bitfield.ranges, ranges);
+            } else {
+                assert_eq!(bitfield.ranges, ChunkRanges::all());
+            }
+        }
+        Ok(())
+    }
+
+    #[tokio::test]
+    async fn test_observe_partial_mem() -> TestResult<()> {
+        let store = MemStore::new();
+        test_observe_partial(store.blobs()).await?;
+        Ok(())
+    }
+
+    #[tokio::test]
+    async fn test_observe_partial_fs() -> TestResult<()> {
+        let td = tempfile::tempdir()?;
+        let store = FsStore::load(td.path()).await?;
+        test_observe_partial(store.blobs()).await?;
+        Ok(())
+    }
+
     #[tokio::test]
     async fn test_local_info_hash_seq() -> TestResult<()> {
         let sizes = INTERESTING_SIZES;

src/hash.rs

@@ -111,7 +111,7 @@ impl From<&[u8; 32]> for Hash {
 
 impl PartialOrd for Hash {
     fn partial_cmp(&self, other: &Self) -> Option<std::cmp::Ordering> {
-        Some(self.0.as_bytes().cmp(other.0.as_bytes()))
+        Some(self.cmp(other))
     }
 }
 

src/metrics.rs

@@ -4,6 +4,7 @@ use iroh_metrics::{Counter, MetricsGroup};
 
 /// Enum of metrics for the module
 #[allow(missing_docs)]
+#[allow(dead_code)]
 #[derive(Debug, Default, MetricsGroup)]
 #[metrics(name = "iroh-blobs")]
 pub struct Metrics {