refactor: Remove slot mutex and simplify per blob state (n0-computer#104)

## Description

Before, the state we kept per hash was a bit convoluted. The entry point
was a Slot, which contained a tokio mutex to cover the async loading of
an entry state from the metadata db. Then inside that, there was the
actual entry state, wrapped in an option to cover the case that we don't
have anything about the hash.

This was working, but it was an arc in an arc in an arc, and also led to
bugs in the case of trying to export a blob that doesn't exist.

Now all these states are flattened into a single enum, and we can easily
define what should happen when we e.g. do an export of an entry that
does not exist - return an appropriate io error.

## Breaking Changes

None

## Notes & open questions

Note: you could remove the Initial and Loading state by using a
tokio::sync::OnceCell. But that is a true sync primitive, and we want to
use an AtomicRefCell, and also using a OnceLock would come with its own
sync primitive (a semaphore) just for init, and then we have our own one
for the non-load state changes. I don't think it is that bad...

Note: a lot of changes are to add the wait_idle fn in the rpc protocol,
which isn't really related to the changes in the PR but just a way to
get rid of some flaky tests.

<!-- Any notes, remarks or open questions you have to make about the PR.
-->

## Change checklist

- [ ] Self-review.
- [ ] Documentation updates following the [style
guide](https://rust-lang.github.io/rfcs/1574-more-api-documentation-conventions.html#appendix-a-full-conventions-text),
if relevant.
- [ ] Tests if relevant.
- [ ] All breaking changes documented.

Author: rklaehn

.github/workflows/ci.yaml

@@ -143,7 +143,7 @@ jobs:
     - uses: taiki-e/install-action@cross
 
     - name: test
-      run: cross test --all --target ${{ matrix.target }} -- --test-threads=4
+      run: cross test --all --target ${{ matrix.target }} -- --test-threads=1
       env:
         RUST_LOG: ${{ runner.debug && 'TRACE' || 'DEBUG' }}
 

Cargo.lock

@@ -37,10 +37,10 @@ chrono = "0.4.39"
 nested_enum_utils = "0.2.1"
 ref-cast = "1.0.24"
 arrayvec = "0.7.6"
-iroh = "0.91"
+iroh = "0.91.1"
 self_cell = "1.1.0"
 genawaiter = { version = "0.99.1", features = ["futures03"] }
-iroh-base = "0.91"
+iroh-base = "0.91.1"
 reflink-copy = "0.1.24"
 irpc = { version = "0.7.0", features = ["rpc", "quinn_endpoint_setup", "spans", "stream", "derive"], default-features = false }
 iroh-metrics = { version = "0.35" }
@@ -59,9 +59,13 @@ tracing-subscriber = { version = "0.3.19", features = ["fmt"] }
 tracing-test = "0.2.5"
 walkdir = "2.5.0"
 atomic_refcell = "0.1.13"
-iroh = { version = "0.91", features = ["discovery-local-network"]}
+iroh = { version = "0.91.1", features = ["discovery-local-network"]}
 
 [features]
 hide-proto-docs = []
 metrics = []
 default = ["hide-proto-docs"]
+
+[patch.crates-io]
+iroh = { git = "https://github.com/n0-computer/iroh", branch = "main" }
+iroh-base = { git = "https://github.com/n0-computer/iroh", branch = "main" }

Cargo.toml

@@ -39,3 +39,8 @@ name = "ring"
 [[licenses.clarify.license-files]]
 hash = 3171872035
 path = "LICENSE"
+
+[sources]
+allow-git = [
+    "https://github.com/n0-computer/iroh",
+]

deny.toml

@@ -30,6 +30,7 @@ pub mod downloader;
 pub mod proto;
 pub mod remote;
 pub mod tags;
+use crate::api::proto::WaitIdleRequest;
 pub use crate::{store::util::Tag, util::temp_tag::TempTag};
 
 pub(crate) type ApiClient = irpc::Client<proto::Request>;
@@ -298,6 +299,23 @@ impl Store {
         Ok(())
     }
 
+    /// Waits for the store to become completely idle.
+    ///
+    /// This is mostly useful for tests, where you want to check that e.g. the
+    /// store has written all data to disk.
+    ///
+    /// Note that a store is not guaranteed to become idle, if it is being
+    /// interacted with concurrently. So this might wait forever.
+    ///
+    /// Also note that once you get the callback, the store is not guaranteed to
+    /// still be idle. All this tells you that there was a point in time where
+    /// the store was idle between the call and the response.
+    pub async fn wait_idle(&self) -> irpc::Result<()> {
+        let msg = WaitIdleRequest;
+        self.client.rpc(msg).await?;
+        Ok(())
+    }
+
     pub(crate) fn from_sender(client: ApiClient) -> Self {
         Self { client }
     }

src/api.rs

@@ -130,11 +130,16 @@ pub enum Request {
     #[rpc(tx = oneshot::Sender<super::Result<()>>)]
     SyncDb(SyncDbRequest),
     #[rpc(tx = oneshot::Sender<()>)]
+    WaitIdle(WaitIdleRequest),
+    #[rpc(tx = oneshot::Sender<()>)]
     Shutdown(ShutdownRequest),
     #[rpc(tx = oneshot::Sender<super::Result<()>>)]
     ClearProtected(ClearProtectedRequest),
 }
 
+#[derive(Debug, Serialize, Deserialize)]
+pub struct WaitIdleRequest;
+
 #[derive(Debug, Serialize, Deserialize)]
 pub struct SyncDbRequest;
 

src/api/proto.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;