reya/coop
1
0
Fork 0
Code Issues Pull Requests 1 Releases 3 Activity

update

Browse Source
This commit is contained in:
reya
2026-09-22 16:03:03 +07:00
parent 28f4c1596d
commit 6fac58d494
7 changed files with 774 additions and 117 deletions
Download Patch File Download Diff File Expand all files Collapse all files
crates/community/src/community.rs
+217 -21
View File
@@ -1,5 +1,6 @@
use std::collections::{BTreeMap, BTreeSet, HashMap};
use std::path::PathBuf;
use std::time::{Duration, Instant};
use anyhow::Result;
use concord::cord02::{ControlFold, ImageRef};
@@ -10,6 +11,7 @@ use concord::state::{ChannelCursor, ChannelKeyRef, CommunityState, HeldKey, Held
use concord::{ChannelId, CommunityId, Epoch};
use gpui::{App, AppContext, Context, EventEmitter, Task};
use nostr_sdk::prelude::*;
use smallvec::{SmallVec, smallvec};
use state::NostrRegistry;
use crate::cache;
@@ -25,9 +27,12 @@ const CATCH_UP_PAGES: usize = 20;
pub const LOAD_OLDER_PAGES: usize = 6;
/// Rows one timeline read returns before the caller asks for more.
pub const TIMELINE_PAGE: usize = 100;
/// Side events read per row, so a reaction flood cannot displace the rows it
/// decorates.
/// Side events read per row, so a reaction flood cannot displace the rows it decorates.
const SIDE_EVENT_FACTOR: usize = 4;
/// The shortest gap between two automatic catch-up rounds for one channel.
pub const MIN_ROUND_INTERVAL: Duration = Duration::from_secs(30);
/// How long a channel may go unsynced before the scheduler repairs it.
const STALE_AFTER: Duration = Duration::from_secs(300);
/// Which direction a sync round reads.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
@@ -41,6 +46,7 @@ pub enum Intent {
pub struct Progress {
pub fetched: usize,
pub opened: usize,
pub unreadable: usize,
pub exhausted: bool,
pub failed: bool,
pub errors: usize,
@@ -102,6 +108,10 @@ pub enum CommunityEvent {
Open(CommunityId),
Close(CommunityId),
Channel(CommunityId, ChannelId),
/// History exists here that no held key can open.
Unreadable(CommunityId),
/// The last round could not reach the community's relays.
Failed(CommunityId),
Error(String),
}
@@ -119,9 +129,17 @@ pub struct Community {
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>,
/// Wraps held per channel that no key we hold can open.
unreadable: BTreeMap<ChannelId, usize>,
/// When each channel's last round started, which paces the automatic ones.
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]>,
}
impl EventEmitter<CommunityEvent> for Community {}
@@ -142,9 +160,13 @@ impl Community {
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![],
}
}
@@ -212,6 +234,44 @@ impl Community {
self.state.stranded
}
/// Wraps held here that no key we hold can open.
pub fn unreadable(&self, channel: &ChannelId) -> usize {
self.unreadable.get(channel).copied().unwrap_or(0)
}
/// The last round's outcome for `channel`, when one has run.
pub fn progress(&self, channel: &ChannelId) -> Option<Progress> {
self.progress.get(channel).copied()
}
/// The epoch a private channel's key is missing for, when we hold none.
pub fn missing_key(&self, channel: &ChannelId) -> Option<Epoch> {
self.state
.channels
.iter()
.find(|held| held.id == *channel && held.private && held.key.is_none())
.map(|held| held.epoch)
}
/// The epoch a channel rotation removed us at, when it removed us.
pub fn channel_removed_at(&self, channel: &ChannelId) -> Option<Epoch> {
self.state.channel_cuts.get(channel).map(|cut| Epoch(*cut))
}
/// Whether an automatic catch-up for `channel` is worth asking for yet.
pub fn due(&self, channel: &ChannelId) -> bool {
self.last_round
.get(channel)
.is_none_or(|at| at.elapsed() >= MIN_ROUND_INTERVAL)
}
/// Whether `channel` has been synced before and has since gone stale.
fn stale(&self, channel: &ChannelId) -> bool {
self.last_round
.get(channel)
.is_some_and(|at| at.elapsed() >= STALE_AFTER)
}
pub fn channels(&self) -> &[ChannelKeyRef] {
&self.state.channels
}
@@ -222,12 +282,20 @@ impl Community {
/// A public channel derives its write plane from the community root.
fn channel_secret(&self, channel: &ChannelId) -> Option<(Epoch, [u8; 32])> {
if self.state.removed_at.is_some() || self.state.stranded {
return None;
}
let held = self
.state
.channels
.iter()
.find(|held| held.id == *channel)?;
if self.state.channel_cut(channel, held.epoch) {
return None;
}
held.current()
.or_else(|| (!held.private).then_some((held.epoch, self.state.community_root)))
}
@@ -296,18 +364,23 @@ impl Community {
round.queued = None;
}
self.last_round.insert(channel, Instant::now());
let round = cx.background_spawn(async move {
sync_round(&client, &pages, &channel, &held, &relays, saved, intent).await
});
cx.spawn(async move |this, cx| {
let finisher = cx.spawn(async move |this, cx| {
let result = round.await;
if let Err(error) = this.update(cx, |this, cx| this.finish_round(channel, result, cx)) {
log::warn!("community: a channel round outlived its community: {error}");
}
})
.detach();
Ok(())
});
self.tasks.push(finisher);
}
fn finish_round(
@@ -342,20 +415,56 @@ impl Community {
self.start_round(channel, intent, cx);
}
if let Ok(progress) = &outcome
&& progress.failed
&& progress.errors > 0
&& progress.fetched == 0
{
cx.emit(CommunityEvent::Error(format!(
"could not reach {} of the community's relays",
progress.errors
)));
if let Ok(progress) = &outcome {
self.progress.insert(channel, *progress);
self.record_unreadable(channel, progress.unreadable);
if progress.failed && progress.errors > 0 {
cx.emit(CommunityEvent::Failed(self.state.id));
}
if progress.unreadable > 0 {
cx.emit(CommunityEvent::Unreadable(self.state.id));
}
}
cx.emit(CommunityEvent::Updated(self.state.id));
}
/// Remember the wraps no held key could open.
fn record_unreadable(&mut self, channel: ChannelId, count: usize) {
if count == 0 {
return;
}
let seen = self.unreadable.entry(channel).or_default();
*seen = (*seen).max(count);
}
pub(crate) fn tick(&mut self, cx: &mut Context<Self>) {
let Some(channel) = self.active_channel() else {
return;
};
if !self.stale(&channel) {
return;
}
self.refresh(cx);
let catch_up = self.sync_channel(&channel, Intent::CatchUp, cx);
let task = cx.spawn(async move |_this, _cx| {
if let Err(error) = catch_up.await {
log::warn!("community: the scheduler's catch-up failed: {error}");
}
Ok(())
});
self.tasks.push(task);
}
/// Fold a round's cursor findings in, monotonically, and persist them.
fn merge_cursor(&mut self, channel: ChannelId, cursor: ChannelCursor, cx: &mut Context<Self>) {
let held = self
@@ -375,16 +484,19 @@ impl Community {
}
/// Write the local state document out.
fn persist(&self, cx: &Context<Self>) {
fn persist(&mut self, cx: &Context<Self>) {
let client = NostrRegistry::global(cx).read(cx).client();
let state = self.state.clone();
cx.background_spawn(async move {
let task = cx.background_spawn(async move {
if let Err(error) = cache::save_state(&client, &state).await {
log::warn!("community: failed to persist the local state: {error}");
}
})
.detach();
Ok(())
});
self.tasks.push(task);
}
/// The channel's timeline, folded from the local cache, oldest first.
@@ -647,12 +759,15 @@ impl Community {
let catch_up = self.sync_channel(&channel, Intent::CatchUp, cx);
cx.spawn(async move |_this, _cx| {
let task = cx.spawn(async move |_this, _cx| {
if let Err(error) = catch_up.await {
log::warn!("community: the catch-up after a rekey failed: {error}");
}
})
.detach();
Ok(())
});
self.tasks.push(task);
}
/// Rebuilds the community from the wraps in the local database.
@@ -686,6 +801,17 @@ impl Community {
self.control = snapshot.control;
self.members = snapshot.members;
self.load_images(cx);
let reported = !snapshot.unreadable.is_empty();
for (channel, count) in snapshot.unreadable {
self.record_unreadable(channel, count);
}
if reported {
cx.emit(CommunityEvent::Unreadable(self.state.id));
}
cx.emit(CommunityEvent::Updated(self.state.id));
cx.notify();
}
@@ -886,7 +1012,77 @@ async fn sync_round(
fn absorb(progress: &mut Progress, page: &WrapPage) {
progress.fetched += page.raw;
progress.opened += page.opened.len();
progress.unreadable += page.unreadable;
progress.exhausted |= page.exhausted;
progress.failed |= page.failed;
progress.errors += page.errors;
}
#[cfg(test)]
mod tests {
use super::*;
/// One public channel under a held root, and nothing else.
fn state(channel: ChannelId) -> CommunityState {
CommunityState {
id: CommunityId::from_bytes([0x42; 32]),
name: None,
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![ChannelKeyRef {
id: channel,
name: "general".to_owned(),
private: false,
epoch: Epoch(0),
key: None,
priors: Vec::new(),
}],
relays: Vec::new(),
heads: Vec::new(),
banned: BTreeSet::new(),
cursors: BTreeMap::new(),
held_roots: Vec::new(),
channel_cuts: BTreeMap::new(),
removed_at: None,
stranded: false,
dissolved: false,
added_at_ms: 0,
}
}
fn community(state: CommunityState) -> Community {
Community::new(state, PageRegistry::default())
}
/// A rotation that excluded us leaves no key to write under: a wrap sealed
/// with the retired root would reach nobody who rotated with it.
#[test]
fn a_removed_member_holds_no_write_key() {
let channel = ChannelId::from_bytes([0x9c; 32]);
// A public channel writes from the community root while we hold it.
assert_eq!(
community(state(channel)).channel_secret(&channel),
Some((Epoch(0), [0x02; 32]))
);
// A base removal closes the community...
let mut removed = state(channel);
removed.removed_at = Some(Epoch(1));
assert_eq!(community(removed).channel_secret(&channel), None);
// ...a strand closes it too, because the rotation moved past our epoch...
let mut stranded = state(channel);
stranded.stranded = true;
assert_eq!(community(stranded).channel_secret(&channel), None);
// ...and a channel cut closes the one channel.
let mut cut = state(channel);
cut.channel_cuts.insert(channel, 0);
assert_eq!(community(cut).channel_secret(&channel), None);
}
}
crates/community/src/history.rs
+69 -11
View File
@@ -4,8 +4,8 @@ use std::sync::atomic::{AtomicU64, Ordering};
use std::sync::{Arc, Mutex, MutexGuard};
use std::time::{Duration, Instant};
use anyhow::Result;
use concord::cord01::KIND_WRAP_EPHEMERAL;
use anyhow::{Result, bail};
use concord::cord01::{KIND_WRAP_EPHEMERAL, OpenedStream};
use concord::cord03::{self, ChatRumor};
use concord::derive::channel_group_key;
use concord::state::{ChannelCursor, HeldKey};
@@ -137,6 +137,8 @@ impl PageRegistry {
pub struct WrapPage {
pub opened: Vec<ChatRumor>,
pub raw: usize,
/// Wraps that reached us under a held plane but that no held key could open.
pub unreadable: usize,
pub newest_ms: Option<u64>,
pub oldest_ms: Option<u64>,
pub exhausted: bool,
@@ -210,15 +212,8 @@ pub async fn page(
continue;
};
// A retired key reads only what was sealed before its rotation published.
if held
.retired_at
.is_some_and(|retired| wrap.created_at.as_secs() > retired)
{
continue;
}
let Ok((stream, rumor)) = cord03::open(&wrap, group, channel, held.epoch) else {
let Ok((stream, rumor)) = read_under(&wrap, held, group, channel) else {
walk.unreadable += 1;
continue;
};
@@ -231,6 +226,23 @@ pub async fn page(
Ok(walk.finish(opened))
}
/// Opens one wrap under a held key.
fn read_under(
wrap: &Event,
held: &HeldKey,
group: &GroupKey,
channel: &ChannelId,
) -> Result<(OpenedStream, ChatRumor)> {
if held
.retired_at
.is_some_and(|retired| wrap.created_at.as_secs() > retired)
{
bail!("sealed after the key that reads it was retired");
}
Ok(cord03::open(wrap, group, channel, held.epoch)?)
}
/// The one filter a page is asked for.
fn wrap_filter(authors: &[PublicKey], window: Window, limit: usize) -> Filter {
let mut filter = Filter::new()
@@ -369,6 +381,8 @@ struct Walk {
oldest_ms: Option<u64>,
raw: usize,
errors: usize,
/// Wraps the caller could not read under any held key.
unreadable: usize,
/// A short page ended the walk.
bottom: bool,
}
@@ -395,6 +409,7 @@ impl Walk {
oldest_ms: None,
raw: 0,
errors: 0,
unreadable: 0,
bottom: false,
}
}
@@ -458,6 +473,7 @@ impl Walk {
WrapPage {
opened,
raw: self.raw,
unreadable: self.unreadable,
newest_ms: self.newest_ms,
oldest_ms: self.oldest_ms,
exhausted: swept && self.raw > 0,
@@ -586,6 +602,48 @@ mod tests {
assert_eq!(page.raw, 3);
}
/// A wrap that reaches us and still will not open is history we cannot read,
/// not history that does not exist.
#[test]
fn a_wrap_no_held_key_can_open_reads_as_unreadable() {
let channel = ChannelId::from_bytes([0x9cu8; 32]);
let other = ChannelId::from_bytes([0x9du8; 32]);
let author = Keys::generate();
let group = channel_group_key(&SECRET, &channel, Epoch(0)).expect("derives");
let held = HeldKey {
epoch: Epoch(0),
key: SECRET,
retired_at: Some(1_000),
};
let wrap_at = |channel: &ChannelId, at_ms: u64| {
let rumor = build_message(
author.public_key(),
channel,
Epoch(0),
"sealed",
None,
at_ms,
None,
);
smol::block_on(seal_rumor(&rumor, &group, &author, false))
.expect("seals")
.0
};
// Sealed before the rotation superseded this key, so it still reads.
let before = wrap_at(&channel, 999_000);
assert!(read_under(&before, &held, &group, &channel).is_ok());
// Sealed after the cutoff the rotation set on that key.
let after = wrap_at(&channel, 1_001_000);
assert!(read_under(&after, &held, &group, &channel).is_err());
// Sealed to this plane but bound to another channel.
let misbound = wrap_at(&other, 999_000);
assert!(read_under(&misbound, &held, &group, &channel).is_err());
}
#[test]
fn an_empty_answer_never_seals_the_channel() {
let database: BTreeSet<Event> = BTreeSet::new();
crates/community/src/lib.rs
+29 -1
View File
@@ -5,7 +5,7 @@ use anyhow::Result;
pub use concord::cord02::CommunityMetadata;
pub use concord::cord03::{ChatMessage, ReplyRef};
use concord::state::CommunityState;
pub use concord::{ChannelId, CommunityId};
pub use concord::{ChannelId, CommunityId, Epoch};
use futures::future::{Either, select};
use gpui::{App, AppContext, Context, Entity, EventEmitter, Global, Subscription, Task, Window};
use nostr_sdk::prelude::*;
@@ -152,6 +152,8 @@ pub struct CommunityRegistry {
notification_listener: Option<Task<Result<()>>>,
/// Signal consumer task (cancelled on signer change)
signal_consumer: Option<Task<Result<()>>>,
/// The round scheduler (cancelled on signer change)
scheduler: Option<Task<Result<()>>>,
_subscriptions: SmallVec<[Subscription; 2]>,
}
@@ -203,6 +205,7 @@ impl CommunityRegistry {
tasks: smallvec![],
notification_listener: None,
signal_consumer: None,
scheduler: None,
_subscriptions: subscriptions,
}
}
@@ -272,6 +275,7 @@ impl CommunityRegistry {
pub fn reset(&mut self, cx: &mut Context<Self>) {
self.notification_listener = None;
self.signal_consumer = None;
self.scheduler = None;
self.tasks.clear();
self.observers.clear();
self.pages.clear();
@@ -424,6 +428,13 @@ impl CommunityRegistry {
community.update(cx, |community, cx| community.rekey(cx));
}
/// One scheduler pass: every community repairs itself if it has gone stale.
fn tick(&mut self, cx: &mut Context<Self>) {
for community in self.communities.clone() {
community.update(cx, |community, cx| community.tick(cx));
}
}
/// Re-subscribe every community whose held planes moved.
fn sync_subscriptions(&mut self, cx: &mut Context<Self>) {
let nostr = NostrRegistry::global(cx);
@@ -491,6 +502,7 @@ impl CommunityRegistry {
fn handle_notifications(&mut self, cx: &mut Context<Self>) {
self.notification_listener = None;
self.signal_consumer = None;
self.scheduler = None;
let nostr = NostrRegistry::global(cx);
let client = nostr.read(cx).client();
@@ -558,6 +570,22 @@ impl CommunityRegistry {
}
Ok(())
}));
self.scheduler = Some(cx.spawn(async move |this, cx| {
loop {
cx.background_executor().timer(MIN_ROUND_INTERVAL).await;
if let Some(registry) = this.upgrade() {
registry.update(cx, |this, cx| {
this.tick(cx);
});
} else {
break;
}
}
Ok(())
}));
}
}
crates/community/src/sync.rs
+64 -10
View File
@@ -172,6 +172,8 @@ pub struct Snapshot {
pub state: CommunityState,
pub control: ControlFold,
pub members: BTreeSet<PublicKey>,
/// Wraps the store holds per channel that no held key can open.
pub unreadable: BTreeMap<ChannelId, usize>,
}
pub async fn create<S>(
@@ -529,6 +531,7 @@ pub async fn fold(client: &Client, state: &CommunityState) -> Result<Option<Snap
let mut editions = Vec::new();
let mut observed: BTreeMap<PublicKey, u64> = BTreeMap::new();
let mut guestbook_rumors = Vec::new();
let mut unreadable: BTreeMap<ChannelId, usize> = BTreeMap::new();
let mut cached: BTreeMap<ChannelId, BTreeMap<EventId, Observed>> = BTreeMap::new();
for plane in &planes {
@@ -544,19 +547,23 @@ pub async fn fold(client: &Client, state: &CommunityState) -> Result<Option<Snap
continue;
};
// A retired key reads only what was sealed before its rotation published.
if !plane.accepts(wrap) {
continue;
}
match plane.kind {
PlaneKind::Control(_) => {
// A retired root reads only what was sealed before its rotation.
if !plane.accepts(wrap) {
continue;
}
if let Ok(edition) = cord02::open_edition(wrap, &plane.group, &plane.address, true)
{
editions.push(edition);
}
}
PlaneKind::Guestbook => {
if !plane.accepts(wrap) {
continue;
}
if let Ok((_, rumor)) = cord02::guestbook::open(wrap, &plane.group) {
observe(&mut observed, rumor.author, rumor.at_ms);
guestbook_rumors.push(rumor);
@@ -568,11 +575,18 @@ pub async fn fold(client: &Client, state: &CommunityState) -> Result<Option<Snap
continue;
}
if let Ok((opened, rumor)) =
concord::cord03::open(wrap, &plane.group, &channel, epoch)
{
cache::cache_rumor(client, &channel, &opened).await?;
observe(&mut observed, rumor.author, rumor.at_ms);
let opened = if plane.accepts(wrap) {
concord::cord03::open(wrap, &plane.group, &channel, epoch).ok()
} else {
None
};
match opened {
Some((opened, rumor)) => {
cache::cache_rumor(client, &channel, &opened).await?;
observe(&mut observed, rumor.author, rumor.at_ms);
}
None => *unreadable.entry(channel).or_default() += 1,
}
}
}
@@ -627,6 +641,7 @@ pub async fn fold(client: &Client, state: &CommunityState) -> Result<Option<Snap
state,
control,
members,
unreadable,
}))
}
@@ -1137,4 +1152,43 @@ mod tests {
assert!(retired.accepts(&event_at(1_000_000)));
assert!(!retired.accepts(&event_at(1_001_000)));
}
/// A wrap addressed to a held channel plane that will not open is counted, so
/// the panel can tell a quiet room from one whose history it cannot read.
#[test]
fn a_fold_counts_the_channel_wraps_it_cannot_open() {
smol::block_on(async {
let client = client();
let keys = Keys::generate();
let signer = UniversalSigner::new(keys.clone());
let created = create(&client, &signer, &metadata("coop"))
.await
.expect("creates");
let (channel, plane) = planes(&created)
.expect("planes")
.into_iter()
.find_map(|plane| match plane.kind {
PlaneKind::Channel(channel, _) => Some((channel, plane)),
_ => None,
})
.expect("a channel plane");
// Sealed to the channel's own address, but not a seal at all, so no
// held key opens it.
let junk = EventBuilder::new(Kind::GiftWrap, "not a seal")
.custom_created_at(Timestamp::from_secs(1_700_000_000))
.finalize(plane.group.keys())
.expect("signs");
client.database().save_event(&junk).await.expect("saves");
let snapshot = fold(&client, &created)
.await
.expect("folds")
.expect("a control plane");
assert_eq!(snapshot.unreadable.get(&channel).copied(), Some(1));
});
}
}