diff --git a/crates/community/src/cache.rs b/crates/community/src/cache.rs index 12ce5e18..16e849ad 100644 --- a/crates/community/src/cache.rs +++ b/crates/community/src/cache.rs @@ -227,105 +227,3 @@ fn state_document_of(event: &Event) -> Option { 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); - }); - } -} diff --git a/crates/community/src/community.rs b/crates/community/src/community.rs index e5cd9a24..ccb2abf9 100644 --- a/crates/community/src/community.rs +++ b/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, banner: Option, banner_ref: Option, - dirty: bool, - refresh_task: Option>>, - icon_task: Option>>, - banner_task: Option>>, rounds: HashMap, /// The last completed round per channel, for the panel's honest states. progress: HashMap, @@ -137,8 +133,6 @@ pub struct Community { last_round: HashMap, pages: PageRegistry, rekey_task: Option>>, - rekey_dirty: bool, - /// Spawned folds, round bookkeeping and publishes, cancelled on drop tasks: SmallVec<[Task>; 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) { 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) { - 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>, cx: &mut Context) { - 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) { 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| { diff --git a/crates/community/src/lib.rs b/crates/community/src/lib.rs index 71ca90fe..994e788a 100644 --- a/crates/community/src/lib.rs +++ b/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) { diff --git a/crates/community/src/rekey.rs b/crates/community/src/rekey.rs index 42891241..fb07302c 100644 --- a/crates/community/src/rekey.rs +++ b/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> { 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 = 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); } diff --git a/crates/community/src/sync.rs b/crates/community/src/sync.rs index edfe79e5..269a7639 100644 --- a/crates/community/src/sync.rs +++ b/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 {