clean up

crates/community/src/cache.rs

@@ -227,105 +227,3 @@ fn state_document_of(event: &Event) -> Option<CommunityId> {
     let hex = identifier.strip_prefix(STATE_PREFIX)?;
     hex.parse().ok()
 }
-
-#[cfg(test)]
-mod tests {
-    use std::collections::BTreeSet;
-
-    use concord::Epoch;
-    use nostr_memory::MemoryDatabase;
-
-    use super::*;
-
-    fn client() -> Client {
-        ClientBuilder::default()
-            .database(MemoryDatabase::unbounded())
-            .build()
-    }
-
-    #[test]
-    fn load_states_reads_one_document_per_community_and_ignores_other_documents() {
-        smol::block_on(async {
-            let client = client();
-
-            let state = CommunityState {
-                id: CommunityId::from_bytes([0x42; 32]),
-                name: Some("Anime and Manga".to_owned()),
-                owner: Keys::generate().public_key(),
-                owner_salt: [0x01; 32],
-                community_root: [0x02; 32],
-                root_epoch: Epoch(0),
-                control_root: None,
-                control_pks: BTreeMap::new(),
-                channels: Vec::new(),
-                relays: Vec::new(),
-                heads: Vec::new(),
-                banned: BTreeSet::new(),
-                cursors: BTreeMap::new(),
-                held_roots: Vec::new(),
-                channel_cuts: BTreeMap::new(),
-                refounders: BTreeSet::new(),
-                removed_at: None,
-                stranded: false,
-                dissolved: false,
-                added_at_ms: 7,
-            };
-
-            save_state(&client, &state).await.expect("saves");
-
-            // A cached rumor is also an application-specific document, but not a
-            // state document, so the prefix keeps it out of the state scan.
-            let other = EventBuilder::new(Kind::ApplicationSpecificData, "{}")
-                .tags([Tag::identifier("deadbeef")])
-                .finalize(&*LOCAL_KEYS)
-                .expect("builds");
-            client.database().save_event(&other).await.expect("saves");
-
-            let loaded = load_states(&client).await.expect("loads");
-
-            assert_eq!(loaded, vec![state]);
-        });
-    }
-
-    #[test]
-    fn caching_the_same_rumor_twice_leaves_one_row() {
-        smol::block_on(async {
-            let client = client();
-            let channel = ChannelId::from_bytes([0x9c; 32]);
-            let keys = Keys::generate();
-            let group = concord::derive::channel_group_key(&[0x07; 32], &channel, Epoch(0))
-                .expect("a group key");
-
-            let rumor = concord::cord03::build_message(
-                keys.public_key(),
-                &channel,
-                Epoch(0),
-                "twice",
-                None,
-                1_700_000_000_000,
-                None,
-            );
-            let (wrap, _) = concord::cord03::seal_rumor(&rumor, &group, &keys, false)
-                .await
-                .expect("seals");
-            let (opened, _) =
-                concord::cord03::open(&wrap, &group, &channel, Epoch(0)).expect("opens");
-
-            assert!(
-                cache_rumor(&client, &channel, &opened)
-                    .await
-                    .expect("caches")
-            );
-            assert!(
-                cache_rumor(&client, &channel, &opened)
-                    .await
-                    .expect("caches")
-            );
-
-            let cached = query_rumors(&client, &channel, None, 10, None)
-                .await
-                .expect("reads");
-            assert_eq!(cached.len(), 1);
-        });
-    }
-}

crates/community/src/community.rs

@@ -19,7 +19,7 @@ use state::NostrRegistry;
 use crate::cache;
 use crate::history::{self, PageRegistry, Window, WrapPage};
 use crate::rekey::{self, Adoptions};
-use crate::sync::{self, Snapshot};
+use crate::sync::{self};
 
 /// Wraps one relay returns for one page request.
 const PAGE_WRAPS: usize = 50;
@@ -124,10 +124,6 @@ pub struct Community {
     icon_ref: Option<ImageRef>,
     banner: Option<PathBuf>,
     banner_ref: Option<ImageRef>,
-    dirty: bool,
-    refresh_task: Option<Task<Result<()>>>,
-    icon_task: Option<Task<Result<()>>>,
-    banner_task: Option<Task<Result<()>>>,
     rounds: HashMap<ChannelId, Round>,
     /// The last completed round per channel, for the panel's honest states.
     progress: HashMap<ChannelId, Progress>,
@@ -137,8 +133,6 @@ pub struct Community {
     last_round: HashMap<ChannelId, Instant>,
     pages: PageRegistry,
     rekey_task: Option<Task<Result<()>>>,
-    rekey_dirty: bool,
-    /// Spawned folds, round bookkeeping and publishes, cancelled on drop
     tasks: SmallVec<[Task<Result<()>>; 2]>,
 }
 
@@ -155,17 +149,12 @@ impl Community {
             icon_ref: None,
             banner: None,
             banner_ref: None,
-            dirty: false,
-            refresh_task: None,
-            icon_task: None,
-            banner_task: None,
             rounds: HashMap::new(),
             progress: HashMap::new(),
             unreadable: BTreeMap::new(),
             last_round: HashMap::new(),
             pages,
             rekey_task: None,
-            rekey_dirty: false,
             tasks: smallvec![],
         }
     }
@@ -620,7 +609,6 @@ impl Community {
             let (opened, _) = cord03::open(&wrap, &group, &channel, epoch)?;
             cache::cache_rumor(&client, &channel, &opened).await?;
 
-            sync::connect_relays(&client, &relays).await;
             sync::publish_wrap(&client, &wrap, &relays).await;
 
             Ok(opened.rumor_id)
@@ -631,16 +619,12 @@ impl Community {
     ///
     /// The caller re-folds afterwards; this only seeds the new planes.
     pub(crate) fn adopt(&mut self, state: CommunityState) {
-        // A fold already in flight would write its pre-adoption state back.
-        self.refresh_task = None;
-        self.dirty = false;
         self.state = state;
     }
 
     /// Adopt whatever the rekey watch has delivered, then re-page what it moved.
     pub fn rekey(&mut self, cx: &mut Context<Self>) {
         if self.rekey_task.is_some() {
-            self.rekey_dirty = true;
             return;
         }
 
@@ -651,6 +635,7 @@ impl Community {
 
         let client = nostr.read(cx).client();
         let signer = nostr.read(cx).signer();
+
         let state = self.state.clone();
         let roles = self.control.roles.clone();
 
@@ -746,11 +731,6 @@ impl Community {
             Ok(_) => {}
             Err(error) => cx.emit(CommunityEvent::Error(error.to_string())),
         }
-
-        if self.rekey_dirty {
-            self.rekey_dirty = false;
-            self.rekey(cx);
-        }
     }
 
     /// Fold an adoption into the held state, persist it, and re-page what moved.
@@ -818,8 +798,6 @@ impl Community {
                 continue;
             };
 
-            // `stepped` carries the key held before the walk plus every epoch it
-            // passed through, each with the cutoff its superseding rotation set.
             for key in channel.stepped {
                 if !held.priors.iter().any(|prior| prior.epoch == key.epoch) {
                     held.priors.push(key);
@@ -889,67 +867,53 @@ impl Community {
 
     /// Rebuilds the community from the wraps in the local database.
     pub fn refresh(&mut self, cx: &mut Context<Self>) {
-        if self.refresh_task.is_some() {
-            self.dirty = true;
-            return;
-        }
-
         let nostr = NostrRegistry::global(cx);
         let client = nostr.read(cx).client();
 
         let state = self.state.clone();
         let folded = cx.background_spawn(async move { sync::fold(&client, &state).await });
 
-        self.refresh_task = Some(cx.spawn(async move |this, cx| {
+        self.tasks.push(cx.spawn(async move |this, cx| {
             let result = folded.await;
-            this.update(cx, |this, cx| this.apply(result, cx))?;
+
+            this.update(cx, |this, cx| {
+                match result {
+                    Ok(Some(snapshot)) => {
+                        let mut state = snapshot.state;
+                        state.cursors = std::mem::take(&mut this.state.cursors);
+
+                        this.state = state;
+                        this.control = snapshot.control;
+                        this.members = snapshot.members;
+                        this.load_images(cx);
+
+                        let reported = !snapshot.unreadable.is_empty();
+                        this.unreadable = snapshot.unreadable;
+
+                        if reported {
+                            cx.emit(CommunityEvent::Unreadable(this.state.id));
+                        }
+
+                        cx.emit(CommunityEvent::Updated(this.state.id));
+                        cx.notify();
+                    }
+                    Err(error) => {
+                        cx.emit(CommunityEvent::Error(error.to_string()));
+                    }
+                    _ => {}
+                };
+            })?;
+
             Ok(())
         }));
     }
 
-    fn apply(&mut self, result: Result<Option<Snapshot>>, cx: &mut Context<Self>) {
-        self.refresh_task = None;
-
-        match result {
-            Ok(Some(snapshot)) => {
-                let mut state = snapshot.state;
-                state.cursors = std::mem::take(&mut self.state.cursors);
-                self.state = state;
-                self.control = snapshot.control;
-                self.members = snapshot.members;
-                self.load_images(cx);
-
-                // A fold reads every plane, so its count is the truth for every
-                // channel rather than a sample of the region one round read.
-                // Replacing it is what lets the count fall again once a key is
-                // adopted; a partial round only ever raises it.
-                let reported = !snapshot.unreadable.is_empty();
-                self.unreadable = snapshot.unreadable;
-
-                if reported {
-                    cx.emit(CommunityEvent::Unreadable(self.state.id));
-                }
-
-                cx.emit(CommunityEvent::Updated(self.state.id));
-                cx.notify();
-            }
-            Ok(None) => {}
-            Err(error) => cx.emit(CommunityEvent::Error(error.to_string())),
-        }
-
-        if self.dirty {
-            self.dirty = false;
-            self.refresh(cx);
-        }
-    }
-
     /// Resolve the folded icon and banner into local files.
     fn load_images(&mut self, cx: &mut Context<Self>) {
         let (icon, banner) = match self.control.community.as_ref() {
             Some(metadata) => (metadata.icon.clone(), metadata.banner.clone()),
             None => (None, None),
         };
-
         self.load_icon(icon, cx);
         self.load_banner(banner, cx);
     }
@@ -966,7 +930,7 @@ impl Community {
             return;
         };
 
-        self.icon_task = Some(cx.spawn(async move |this, cx| {
+        self.tasks.push(cx.spawn(async move |this, cx| {
             match sync::resolve_image(&icon, cx).await {
                 Ok(path) => {
                     this.update(cx, |this, cx| {
@@ -992,7 +956,7 @@ impl Community {
             return;
         };
 
-        self.banner_task = Some(cx.spawn(async move |this, cx| {
+        self.tasks.push(cx.spawn(async move |this, cx| {
             match sync::resolve_image(&banner, cx).await {
                 Ok(path) => {
                     this.update(cx, |this, cx| {

crates/community/src/lib.rs

@@ -24,10 +24,7 @@ mod sync;
 pub use community::*;
 pub use sync::*;
 
-/// How long a burst of relay notifications is collected before it is folded.
 const PUMP_WINDOW: Duration = Duration::from_millis(200);
-/// How long a community's relays may deliver nothing before
-/// its standing subscription is torn down and re-issued.
 const LIVE_ROTATE: Duration = Duration::from_secs(90);
 
 pub fn init(cx: &mut App) {

crates/community/src/rekey.rs

@@ -29,8 +29,6 @@ pub struct Watch {
     pub epoch: Epoch,
 }
 
-/// The permission a rotation of `scope` is judged under, by whoever reads it and
-/// by this client when it publishes one.
 fn permissions(scope: RekeyScope) -> &'static [u64] {
     match scope {
         RekeyScope::Base => &[Permissions::BAN],
@@ -39,11 +37,6 @@ fn permissions(scope: RekeyScope) -> &'static [u64] {
 }
 
 /// Every address a community's rotations can arrive at.
-///
-/// The base scope watches the epoch after *every* held root, not only the
-/// current one: a rotation published while this client was away is read from
-/// the root it stepped off, and the npub that minted an epoch is only knowable
-/// from the rotation that minted it.
 pub fn watches(state: &CommunityState) -> Result<Vec<Watch>> {
     let mut watches = Vec::new();
     let roots = state.roots();
@@ -185,8 +178,8 @@ where
     S: AsyncNip44 + ?Sized,
 {
     let watches = watches(state)?;
-
     let mut authors: BTreeSet<PublicKey> = BTreeSet::new();
+
     for watch in &watches {
         authors.insert(watch.address);
     }
@@ -666,12 +659,6 @@ where
             .iter()
             .filter(|rotation| rotation.continuity(held_epoch, &held_key) == Continuity::Extends)
         {
-            // Who minted an epoch is proven by the rotation itself, not by a blob:
-            // a rotation is only a candidate after continuity against a key we
-            // hold, so its rotator minted this epoch whether or not it addressed
-            // us. A member who joined on a stale bundle never held the epochs
-            // between, and the snapshot that seeds them is only honored on this
-            // npub's authority (CORD-02 §5).
             if scope == RekeyScope::Base {
                 step.refounders.insert(rotation.rotator);
             }

crates/community/src/sync.rs

@@ -214,8 +214,6 @@ where
 
 /// Best-effort publication of the genesis wraps to the community's relays.
 pub(crate) async fn publish_wraps(client: &Client, wraps: &[Event], relays: &[RelayUrl]) {
-    connect_relays(client, relays).await;
-
     for wrap in wraps {
         publish_wrap(client, wrap, relays).await;
     }
@@ -232,6 +230,8 @@ pub(crate) async fn connect_relays(client: &Client, relays: &[RelayUrl]) {
 
 /// Best-effort publication of a single wrap to the community's relays.
 pub(crate) async fn publish_wrap(client: &Client, wrap: &Event, relays: &[RelayUrl]) {
+    connect_relays(client, relays).await;
+
     let sent = if relays.is_empty() {
         client.send_event(wrap).broadcast().await
     } else {