ignore: refactoring entity state

.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.toml

@@ -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" }

deny.toml

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

src/api.rs

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