fix: fixes to subscriptions and add test

Author: Frando

iroh-willow/src/store/memory.rs

@@ -103,7 +103,12 @@ impl traits::SecretStorage for Rc<RefCell<SecretStore>> {
 
 #[derive(Debug, Default)]
 pub struct EntryStore {
-    entries: HashMap<NamespaceId, Vec<AuthorisedEntry>>,
+    stores: HashMap<NamespaceId, NamespaceStore>,
+}
+
+#[derive(Debug, Default)]
+pub struct NamespaceStore {
+    entries: Vec<AuthorisedEntry>,
     events: EventQueue<StoreEvent>,
 }
 
@@ -199,8 +204,9 @@ impl traits::EntryReader for Rc<RefCell<EntryStore>> {
         range: &Range3d,
     ) -> impl Iterator<Item = Result<AuthorisedEntry>> + 'a {
         let slf = self.borrow();
-        slf.entries
+        slf.stores
             .get(&namespace)
+            .map(|s| &s.entries)
             .into_iter()
             .flatten()
             .filter(|entry| range.includes_entry(entry.entry()))
@@ -216,10 +222,11 @@ impl traits::EntryReader for Rc<RefCell<EntryStore>> {
         path: &Path,
     ) -> Result<Option<AuthorisedEntry>> {
         let inner = self.borrow();
-        let Some(entries) = inner.entries.get(&namespace) else {
+        let Some(entries) = inner.stores.get(&namespace) else {
             return Ok(None);
         };
         Ok(entries
+            .entries
             .iter()
             .find(|e| {
                 let e = e.entry();
@@ -231,10 +238,11 @@ impl traits::EntryReader for Rc<RefCell<EntryStore>> {
 
 impl EntryStore {
     fn ingest_entry(&mut self, entry: &AuthorisedEntry, origin: EntryOrigin) -> Result<bool> {
-        let entries = self
-            .entries
+        let store = self
+            .stores
             .entry(*entry.entry().namespace_id())
             .or_default();
+        let entries = &mut store.entries;
         let new = entry.entry();
         let mut to_prune = vec![];
         for (i, existing) in entries.iter().enumerate() {
@@ -258,7 +266,7 @@ impl EntryStore {
         }
         for i in to_prune {
             let pruned = entries.remove(i);
-            self.events.insert(move |id| {
+            store.events.insert(move |id| {
                 StoreEvent::Pruned(
                     id,
                     traits::PruneEvent {
@@ -269,7 +277,8 @@ impl EntryStore {
             });
         }
         entries.push(entry.clone());
-        self.events
+        store
+            .events
             .insert(|id| StoreEvent::Ingested(id, entry.clone(), origin));
         Ok(true)
     }
@@ -284,11 +293,23 @@ impl traits::EntryStorage for Rc<RefCell<EntryStore>> {
     }
 
     fn snapshot(&self) -> Result<Self::Snapshot> {
-        let entries = self.borrow().entries.clone();
-        Ok(Rc::new(RefCell::new(EntryStore {
-            entries,
-            events: EventQueue::default(),
-        })))
+        // This is quite ugly. But this is a quick memory impl only.
+        // But we should really maybe strive to not expose snapshots.
+        let stores = self
+            .borrow()
+            .stores
+            .iter()
+            .map(|(key, value)| {
+                (
+                    *key,
+                    NamespaceStore {
+                        entries: value.entries.clone(),
+                        events: Default::default(),
+                    },
+                )
+            })
+            .collect();
+        Ok(Rc::new(RefCell::new(EntryStore { stores })))
     }
 
     fn ingest_entry(&self, entry: &AuthorisedEntry, origin: EntryOrigin) -> Result<bool> {
@@ -302,11 +323,18 @@ impl traits::EntryStorage for Rc<RefCell<EntryStore>> {
         area: Area,
         params: SubscribeParams,
     ) -> impl Stream<Item = traits::StoreEvent> + Unpin + 'static {
+        let progress_id = self
+            .borrow_mut()
+            .stores
+            .entry(namespace)
+            .or_default()
+            .events
+            .next_progress_id();
         EventStream {
             area,
             params,
             namespace,
-            progress_id: self.borrow().events.next_progress_id(),
+            progress_id,
             store: Rc::downgrade(&self),
         }
     }
@@ -345,16 +373,18 @@ struct EventStream {
 impl Stream for EventStream {
     type Item = StoreEvent;
     fn poll_next(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<Self::Item>> {
-        let Some(store) = self.store.upgrade() else {
+        let Some(inner) = self.store.upgrade() else {
             return Poll::Ready(None);
         };
-        let mut store = store.borrow_mut();
+        let mut inner_mut = inner.borrow_mut();
+        let store = inner_mut.stores.entry(self.namespace).or_default();
         let res = ready!(store.events.poll_next(
             self.progress_id,
             |e| e.matches(self.namespace, &self.area, &self.params),
             cx,
         ));
-        drop(store);
+        drop(inner_mut);
+        drop(inner);
         Poll::Ready(match res {
             None => None,
             Some((next_id, event)) => {

iroh/src/client/spaces.rs

@@ -36,7 +36,7 @@ use iroh_willow::{
     },
     session::{
         intents::{serde_encoding::Event, Completion, IntentUpdate},
-        SessionInit,
+        SessionInit, SessionMode,
     },
     store::traits::{StoreEvent, SubscribeParams},
 };
@@ -99,7 +99,11 @@ impl Client {
     }
 
     /// Import a ticket and start to synchronize.
-    pub async fn import_and_sync(&self, ticket: SpaceTicket) -> Result<(Space, SyncHandleSet)> {
+    pub async fn import_and_sync(
+        &self,
+        ticket: SpaceTicket,
+        mode: SessionMode,
+    ) -> Result<(Space, SyncHandleSet)> {
         if ticket.caps.is_empty() {
             anyhow::bail!("Invalid ticket: Does not include any capabilities");
         }
@@ -111,7 +115,7 @@ impl Client {
 
         self.import_caps(ticket.caps).await?;
         let interests = Interests::builder().add_full_cap(CapSelector::any(namespace));
-        let init = SessionInit::reconcile_once(interests);
+        let init = SessionInit::new(interests, mode);
         let mut intents = SyncHandleSet::default();
         for addr in ticket.nodes {
             let node_id = addr.node_id;

iroh/tests/spaces.rs

@@ -10,7 +10,8 @@ use iroh_willow::{
         keys::NamespaceKind,
         meadowcap::AccessMode,
     },
-    session::intents::Completion,
+    session::{intents::Completion, SessionMode},
+    store::traits::{EntryOrigin, StoreEvent},
 };
 use tracing::info;
 
@@ -75,7 +76,10 @@ async fn spaces_smoke() -> Result<()> {
         .await?;
 
     println!("ticket {ticket:?}");
-    let (betty_space, betty_sync_intent) = betty.spaces().import_and_sync(ticket).await?;
+    let (betty_space, betty_sync_intent) = betty
+        .spaces()
+        .import_and_sync(ticket, SessionMode::ReconcileOnce)
+        .await?;
 
     let mut completion = betty_sync_intent.complete_all().await;
     assert_eq!(completion.len(), 1);
@@ -132,3 +136,78 @@ async fn spaces_smoke() -> Result<()> {
 
     Ok(())
 }
+
+#[tokio::test]
+async fn spaces_subscription() -> Result<()> {
+    iroh_test::logging::setup_multithreaded();
+    let (alfie_addr, alfie) = spawn_node().await;
+    let (betty_addr, betty) = spawn_node().await;
+    info!("alfie is {}", alfie_addr.node_id.fmt_short());
+    info!("betty is {}", betty_addr.node_id.fmt_short());
+
+    let betty_user = betty.spaces().create_user().await?;
+    let alfie_user = alfie.spaces().create_user().await?;
+    let alfie_space = alfie
+        .spaces()
+        .create(NamespaceKind::Owned, alfie_user)
+        .await?;
+
+    let _namespace = alfie_space.namespace_id();
+
+    let mut alfie_sub = alfie_space
+        .subscribe_area(Area::new_full(), Default::default())
+        .await?;
+
+    let ticket = alfie_space
+        .share(betty_user, AccessMode::Write, RestrictArea::None)
+        .await?;
+
+    let (betty_space, betty_sync_intent) = betty
+        .spaces()
+        .import_and_sync(ticket, SessionMode::ReconcileOnce)
+        .await?;
+
+    let _sync_task = tokio::task::spawn(async move {
+        // TODO: We should add a "detach" method to a sync intent!
+        // (leaves the sync running but stop consuming events)
+        let _ = betty_sync_intent.complete_all().await;
+    });
+
+    let mut betty_sub = betty_space
+        .resume_subscription(0, Area::new_full(), Default::default())
+        .await?;
+
+    alfie_space
+        .insert_bytes(
+            EntryForm::new(alfie_user, Path::from_bytes(&[b"foo"])?),
+            "hi",
+        )
+        .await?;
+
+    betty_space
+        .insert_bytes(
+            EntryForm::new(betty_user, Path::from_bytes(&[b"foo"])?),
+            "hi",
+        )
+        .await?;
+
+    alfie_space
+        .insert_bytes(
+            EntryForm::new(alfie_user, Path::from_bytes(&[b"foo"])?),
+            "hi!!",
+        )
+        .await?;
+
+    let ev = alfie_sub.next().await.unwrap();
+    println!("ALFIE 2");
+    assert!(matches!(ev, StoreEvent::Ingested(0, _, EntryOrigin::Local)));
+
+    let ev = betty_sub.next().await.unwrap();
+    println!("BETTY 2");
+    assert!(matches!(
+        ev,
+        StoreEvent::Ingested(0, _, EntryOrigin::Remote(_))
+    ));
+
+    Ok(())
+}