Fix non-drop of orbits from cache

Cargo.toml

@@ -14,7 +14,7 @@ async-recursion = "0.3"
 base64 = "0.13"
 bincode = "1.3"
 bs58 = "0.4"
-cached = "0.26"
+cached = "0.34"
 chrono = "0.4"
 didkit = "0.4"
 ethers-core = "0.6"

src/orbit.rs

@@ -243,13 +243,13 @@ pub async fn load_orbit(
     if !dir.exists() {
         return Ok(None);
     }
-    load_orbit_(dir, relay).await.map(Some)
+    load_orbit_inner(dir, relay).await.map(Some)
 }
 
 // Not using this function directly because cached cannot handle Result<Option<>> well.
 // 100 orbits => 600 FDs
 #[cached(size = 100, result = true, sync_writes = true)]
-async fn load_orbit_(dir: PathBuf, relay: (PeerId, Multiaddr)) -> Result<Orbit> {
+async fn load_orbit_inner(dir: PathBuf, relay: (PeerId, Multiaddr)) -> Result<Orbit> {
     let id: OrbitId = String::from_utf8(fs::read(dir.join("id")).await?)?.parse()?;
     let md = Manifest::resolve_dyn(&id, None)
         .await?

src/relay.rs

@@ -3,12 +3,14 @@ use ipfs::{p2p::transport::TransportBuilder, Ipfs, IpfsOptions, TestTypes, Unini
 use libp2p::core::{
     identity::Keypair, multiaddr::multiaddr, transport::MemoryTransport, Multiaddr, PeerId,
 };
-use rocket::tokio::{spawn, task::JoinHandle};
+use rocket::tokio::spawn;
+
+use crate::orbit::AbortOnDrop;
 
 pub struct RelayNode {
     pub port: u16,
     pub id: PeerId,
-    task: JoinHandle<()>,
+    _task: AbortOnDrop<()>,
     _ipfs: Ipfs<TestTypes>,
 }
 
@@ -49,7 +51,7 @@ impl RelayNode {
         let task = spawn(ipfs_task);
         Ok(Self {
             port,
-            task,
+            _task: AbortOnDrop::new(task),
             id,
             _ipfs,
         })
@@ -72,12 +74,6 @@ impl RelayNode {
     }
 }
 
-impl Drop for RelayNode {
-    fn drop(&mut self) {
-        self.task.abort();
-    }
-}
-
 #[cfg(test)]
 pub mod test {
     use super::*;

src/s3/mod.rs

@@ -9,24 +9,27 @@ pub mod behaviour;
 mod entries;
 mod store;
 
-use super::ipfs::{Block, Ipfs};
+use super::{
+    ipfs::{Block, Ipfs},
+    orbit::AbortOnDrop,
+};
 
 pub use entries::{Object, ObjectBuilder, ObjectReader};
 pub use store::Store;
 
-type TaskHandle = tokio::task::JoinHandle<()>;
+type TaskHandle = AbortOnDrop<()>;
 
 #[derive(Clone)]
 pub struct Service {
     pub store: Store,
-    task: Arc<TaskHandle>,
+    _task: Arc<TaskHandle>,
 }
 
 impl Service {
     pub(crate) fn new(store: Store, task: TaskHandle) -> Self {
         Self {
             store,
-            task: Arc::new(task),
+            _task: Arc::new(task),
         }
     }
 
@@ -40,18 +43,12 @@ impl Service {
                 Err(e) => Err(anyhow!(e)),
             });
         let peer_id = store.ipfs.identity().await?.0.to_peer_id();
-        let task = tokio::spawn(kv_task(events, store.clone(), peer_id));
+        let task = AbortOnDrop::new(tokio::spawn(kv_task(events, store.clone(), peer_id)));
         store.request_heads().await?;
         Ok(Service::new(store, task))
     }
 }
 
-impl Drop for Service {
-    fn drop(&mut self) {
-        self.task.abort();
-    }
-}
-
 impl std::ops::Deref for Service {
     type Target = Store;
     fn deref(&self) -> &Self::Target {