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 15:30:12 +07:00
parent 04fb70e657
commit 28f4c1596d
9 changed files with 1431 additions and 80 deletions
Download Patch File Download Diff File Expand all files Collapse all files
crates/community/src/cache.rs
+4
View File
@@ -262,6 +262,10 @@ mod tests {
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: 7,
};
crates/community/src/community.rs
+172 -10
View File
@@ -6,7 +6,7 @@ use concord::cord02::{ControlFold, ImageRef};
use concord::cord03::{self, ChatMessage, ReplyRef};
use concord::cord04::roles::{Permissions, citation_ok};
use concord::derive::channel_group_key;
use concord::state::{ChannelCursor, ChannelKeyRef, CommunityState};
use concord::state::{ChannelCursor, ChannelKeyRef, CommunityState, HeldKey, HeldRoot};
use concord::{ChannelId, CommunityId, Epoch};
use gpui::{App, AppContext, Context, EventEmitter, Task};
use nostr_sdk::prelude::*;
@@ -14,6 +14,7 @@ use state::NostrRegistry;
use crate::cache;
use crate::history::{self, PageRegistry, Window, WrapPage};
use crate::rekey::{self, Adoptions};
use crate::sync::{self, Snapshot};
/// Wraps one relay returns for one page request.
@@ -68,6 +69,7 @@ struct Round {
pub struct SubscriptionKey {
control_pks: BTreeMap<u64, PublicKey>,
channels: Vec<(ChannelId, Epoch, bool)>,
roots: Vec<(u64, [u8; 32])>,
relays: Vec<RelayUrl>,
}
@@ -80,6 +82,11 @@ impl SubscriptionKey {
.iter()
.map(|channel| (channel.id, channel.epoch, channel.private))
.collect(),
roots: state
.roots()
.into_iter()
.map(|root| (root.epoch.0, root.key))
.collect(),
relays: state.relays.clone(),
}
}
@@ -113,6 +120,8 @@ pub struct Community {
banner_task: Option<Task<Result<()>>>,
rounds: HashMap<ChannelId, Round>,
pages: PageRegistry,
rekey_task: Option<Task<Result<()>>>,
rekey_dirty: bool,
}
impl EventEmitter<CommunityEvent> for Community {}
@@ -134,6 +143,8 @@ impl Community {
banner_task: None,
rounds: HashMap::new(),
pages,
rekey_task: None,
rekey_dirty: false,
}
}
@@ -190,6 +201,17 @@ impl Community {
&self.members
}
/// The base epoch a rotation excluded us at: readable history, no writes.
pub fn removed_at(&self) -> Option<Epoch> {
self.state.removed_at
}
/// A complete rotation ahead of our epoch predates our join and
/// carries no blob for us, so the invite landed us on a superseded epoch.
pub fn stranded(&self) -> bool {
self.state.stranded
}
pub fn channels(&self) -> &[ChannelKeyRef] {
&self.state.channels
}
@@ -198,7 +220,7 @@ impl Community {
SubscriptionKey::of(&self.state)
}
/// A public channel derives its plane from the community root; a private one uses its granted key.
/// A public channel derives its write plane from the community root.
fn channel_secret(&self, channel: &ChannelId) -> Option<(Epoch, [u8; 32])> {
let held = self
.state
@@ -206,16 +228,13 @@ impl Community {
.iter()
.find(|held| held.id == *channel)?;
if held.private {
return held.key.map(|key| (held.epoch, key));
}
Some((held.epoch, self.state.community_root))
held.current()
.or_else(|| (!held.private).then_some((held.epoch, self.state.community_root)))
}
/// Every secret the client holds for a channel, newest epoch first.
fn held_keys(&self, channel: &ChannelId) -> Vec<(Epoch, [u8; 32])> {
self.channel_secret(channel).into_iter().collect()
fn held_keys(&self, channel: &ChannelId) -> Vec<HeldKey> {
self.state.held_keys(channel)
}
pub fn sync_channel(
@@ -493,6 +512,149 @@ impl Community {
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;
}
let nostr = NostrRegistry::global(cx);
let Some(me) = nostr.read(cx).current_user() else {
return;
};
let client = nostr.read(cx).client();
let signer = nostr.read(cx).signer();
let state = self.state.clone();
let roles = self.control.roles.clone();
let task =
cx.background_spawn(
async move { rekey::adopt(&client, &state, &roles, &signer, me).await },
);
self.rekey_task = Some(cx.spawn(async move |this, cx| {
let result = task.await;
this.update(cx, |this, cx| this.apply_rekey(result, cx))?;
Ok(())
}));
}
fn apply_rekey(&mut self, result: Result<Adoptions>, cx: &mut Context<Self>) {
self.rekey_task = None;
match result {
Ok(adoptions) if !adoptions.is_empty() => self.merge_adoptions(adoptions, cx),
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.
fn merge_adoptions(&mut self, adoptions: Adoptions, cx: &mut Context<Self>) {
let mut touched: Vec<ChannelId> = Vec::new();
if let Some(base) = adoptions.base {
let mut held = Vec::with_capacity(base.stepped.len() + self.state.held_roots.len());
for key in base.stepped {
held.push(HeldRoot {
epoch: key.epoch,
key: key.key,
control_pk: self.state.control_pks.get(&key.epoch.0).copied(),
retired_at: key.retired_at,
});
}
held.extend(self.state.held_roots.iter().copied());
self.state.held_roots = held;
if let Some(control_pk) = base.control_pk {
self.state.control_pks.insert(base.epoch.0, control_pk);
}
self.state.community_root = base.key;
self.state.root_epoch = base.epoch;
self.state.removed_at = None;
self.state.stranded = false;
// Every channel's plane moved with the root.
touched.extend(self.state.channels.iter().map(|channel| channel.id));
}
for channel in adoptions.channels {
let Some(held) = self
.state
.channels
.iter_mut()
.find(|held| held.id == channel.channel)
else {
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);
}
}
held.key = Some(channel.key);
held.epoch = channel.epoch;
held.private = true;
touched.push(channel.channel);
}
for (channel, epoch) in adoptions.cuts {
self.state.channels.retain(|held| held.id != channel);
self.state.channel_cuts.insert(channel, epoch.0);
self.state.cursors.remove(&channel);
}
if let Some(epoch) = adoptions.removed_at {
self.state.removed_at = Some(epoch);
}
if adoptions.stranded {
self.state.stranded = true;
}
// A rotation re-opens the region its planes now cover: an exhausted
// verdict earned under the old keys cannot be trusted under the new.
for channel in &touched {
if let Some(cursor) = self.state.cursors.get_mut(channel) {
cursor.exhausted = false;
}
}
self.persist(cx);
cx.notify();
cx.emit(CommunityEvent::Updated(self.state.id));
let Some(channel) = self
.active
.or_else(|| self.state.channels.first().map(|channel| channel.id))
else {
return;
};
let catch_up = self.sync_channel(&channel, Intent::CatchUp, cx);
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();
}
/// 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() {
@@ -609,7 +771,7 @@ async fn sync_round(
client: &Client,
pages: &PageRegistry,
channel: &ChannelId,
held: &[(Epoch, [u8; 32])],
held: &[HeldKey],
relays: &[RelayUrl],
saved: ChannelCursor,
intent: Intent,
crates/community/src/history.rs
+43 -12
View File
@@ -6,9 +6,10 @@ use std::time::{Duration, Instant};
use anyhow::Result;
use concord::cord01::KIND_WRAP_EPHEMERAL;
use concord::cord03::{self, ChatRumor, plane_keys};
use concord::state::ChannelCursor;
use concord::{ChannelId, Epoch};
use concord::cord03::{self, ChatRumor};
use concord::derive::channel_group_key;
use concord::state::{ChannelCursor, HeldKey};
use concord::{ChannelId, GroupKey};
use futures::future::{Either, select};
use nostr_sdk::prelude::*;
@@ -149,13 +150,16 @@ pub async fn page(
client: &Client,
pages: &PageRegistry,
channel: &ChannelId,
held: &[(Epoch, [u8; 32])],
held: &[HeldKey],
relays: &[RelayUrl],
window: Window,
max_pages: usize,
limit: usize,
) -> Result<WrapPage> {
let planes = plane_keys(held, channel)?;
let planes: Vec<(HeldKey, GroupKey)> = held
.iter()
.map(|key| Ok((*key, channel_group_key(&key.key, channel, key.epoch)?)))
.collect::<Result<Vec<_>>>()?;
let authors: Vec<PublicKey> = planes.iter().map(|(_, group)| group.pk()).collect();
if authors.is_empty() || relays.is_empty() || limit == 0 {
@@ -201,12 +205,20 @@ pub async fn page(
let answered = client.database().query(filter).await?;
for wrap in walk.accept(answered, limit) {
let Some((epoch, group)) = planes.iter().find(|(_, group)| group.pk() == wrap.pubkey)
let Some((held, group)) = planes.iter().find(|(_, group)| group.pk() == wrap.pubkey)
else {
continue;
};
let Ok((stream, rumor)) = cord03::open(&wrap, group, channel, *epoch) else {
// 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 {
continue;
};
@@ -332,7 +344,6 @@ fn history_subscription() -> SubscriptionId {
SubscriptionId::new(format!("history-{}", NEXT.fetch_add(1, Ordering::Relaxed)))
}
/// Await `future`, giving up after `limit`.
async fn within<F>(limit: Duration, future: F) -> Option<F::Output>
where
F: Future,
@@ -460,6 +471,7 @@ impl Walk {
mod tests {
use std::cmp::Reverse;
use concord::Epoch;
use concord::cord03::{build_message, seal_rumor};
use concord::derive::channel_group_key;
@@ -492,8 +504,27 @@ mod tests {
fn a_walk_pages_back_across_a_rekey() {
let channel = ChannelId::from_bytes([0x9cu8; 32]);
let author = Keys::generate();
let held = [(Epoch(0), SECRET), (Epoch(1), NEXT_SECRET)];
let planes = plane_keys(&held, &channel).expect("derives");
let held = [
HeldKey {
epoch: Epoch(0),
key: SECRET,
retired_at: None,
},
HeldKey {
epoch: Epoch(1),
key: NEXT_SECRET,
retired_at: None,
},
];
let planes: Vec<(HeldKey, GroupKey)> = held
.iter()
.map(|key| {
(
*key,
channel_group_key(&key.key, &channel, key.epoch).expect("derives"),
)
})
.collect();
// Three messages a second apart: a page boundary falls between each.
let base = 1_700_000_000_000;
@@ -530,13 +561,13 @@ mod tests {
let page = serve_page(&relay, walk.region(), 2);
for wrap in walk.accept(page, 2) {
let Some((epoch, group)) =
let Some((held, group)) =
planes.iter().find(|(_, group)| group.pk() == wrap.pubkey)
else {
continue;
};
let (_, rumor) = cord03::open(&wrap, group, &channel, *epoch).expect("opens");
let (_, rumor) = cord03::open(&wrap, group, &channel, held.epoch).expect("opens");
found.push(rumor);
}
}
crates/community/src/lib.rs
+101 -9
View File
@@ -13,10 +13,12 @@ use smallvec::{SmallVec, smallvec};
use state::NostrRegistry;
use crate::history::{PageRegistry, Settled, auth_required};
use crate::rekey::WatchRegistry;
pub mod cache;
mod community;
pub mod history;
mod rekey;
mod sync;
pub use community::*;
@@ -37,6 +39,7 @@ impl Global for GlobalCommunityRegistry {}
enum Signal {
Event(CommunityId),
List,
Rekey(CommunityId),
}
/// Which standing subscription an id belongs to.
@@ -59,6 +62,7 @@ fn route_of(id: &SubscriptionId) -> Option<Route> {
struct Batch {
list: bool,
communities: BTreeSet<CommunityId>,
rekeys: BTreeSet<CommunityId>,
}
/// Whether the pump should keep listening.
@@ -69,17 +73,25 @@ enum Flow {
}
/// Fold one notification into the window, or settle a page it belongs to.
fn route(notification: ClientNotification, pages: &PageRegistry, batch: &mut Batch) -> Flow {
fn route(
notification: ClientNotification,
pages: &PageRegistry,
watches: &WatchRegistry,
batch: &mut Batch,
) -> Flow {
match notification {
ClientNotification::Event {
subscription_id, ..
} => match route_of(&subscription_id) {
Some(Route::List) => batch.list = true,
// Only the database is told here; the fold reads it when the window closes.
Some(Route::Community(id)) => {
batch.communities.insert(id);
}
None => {}
None => {
if let Some(id) = watches.community_of(&subscription_id) {
batch.rekeys.insert(id);
}
}
},
ClientNotification::Message { relay_url, message } => match *message {
RelayMessage::EndOfStoredEvents(id) => {
@@ -109,6 +121,10 @@ async fn flush(tx: &flume::Sender<Signal>, batch: &mut Batch) -> Result<()> {
tx.send_async(Signal::Event(id)).await?;
}
for id in std::mem::take(&mut batch.rekeys) {
tx.send_async(Signal::Rekey(id)).await?;
}
if std::mem::take(&mut batch.list) {
tx.send_async(Signal::List).await?;
}
@@ -129,6 +145,8 @@ pub struct CommunityRegistry {
signal_rx: flume::Receiver<Signal>,
/// The page subscriptions in flight, shared with the notification pump.
pages: PageRegistry,
/// The rekey watch subscriptions, resolved to their community.
watches: WatchRegistry,
tasks: SmallVec<[Task<Result<()>>; 2]>,
/// Notification listener task (cancelled on signer change)
notification_listener: Option<Task<Result<()>>>,
@@ -181,6 +199,7 @@ impl CommunityRegistry {
signal_tx: tx,
signal_rx: rx,
pages: PageRegistry::default(),
watches: WatchRegistry::default(),
tasks: smallvec![],
notification_listener: None,
signal_consumer: None,
@@ -256,6 +275,7 @@ impl CommunityRegistry {
self.tasks.clear();
self.observers.clear();
self.pages.clear();
self.watches.clear();
let nostr = NostrRegistry::global(cx);
let client = nostr.read(cx).client();
@@ -264,9 +284,11 @@ impl CommunityRegistry {
for id in ids {
let client = client.clone();
let subscription = sync::subscription_id(&id);
let rekey = rekey::subscription_id(&id);
self.tasks.push(cx.background_spawn(async move {
client.unsubscribe(&subscription).await?;
client.unsubscribe(&rekey).await?;
Ok(())
}));
}
@@ -274,6 +296,7 @@ impl CommunityRegistry {
self.communities.clear();
self.index.clear();
self.synced.clear();
cx.notify();
}
@@ -392,6 +415,15 @@ impl CommunityRegistry {
community.update(cx, |community, cx| community.refresh(cx));
}
/// Adopt whatever a community's rekey watch has delivered.
fn rekey(&mut self, id: CommunityId, cx: &mut Context<Self>) {
let Some(community) = self.index.get(&id).cloned() else {
return;
};
community.update(cx, |community, cx| community.rekey(cx));
}
/// Re-subscribe every community whose held planes moved.
fn sync_subscriptions(&mut self, cx: &mut Context<Self>) {
let nostr = NostrRegistry::global(cx);
@@ -424,6 +456,18 @@ impl CommunityRegistry {
let filter = sync::live_filter(&planes, sync::live_window(&state));
let relays = key.relays().to_vec();
// The rekey watch is a second standing REQ over the same relays.
let watch = match rekey::watches(&state) {
Ok(watches) => rekey::watch_filter(&watches),
Err(error) => {
cx.emit(CommunityEvent::Error(error.to_string()));
continue;
}
};
let rekey_subscription = rekey::subscription_id(&id);
self.watches.register(rekey_subscription.clone(), id);
self.synced.insert(id, key);
self.tasks.push(cx.spawn(async move |this, cx| {
@@ -433,6 +477,12 @@ impl CommunityRegistry {
})?;
}
if let Err(error) = subscribe(&client, &rekey_subscription, &relays, watch).await {
this.update(cx, |_this, cx| {
cx.emit(CommunityEvent::Error(error.to_string()));
})?;
}
Ok(())
}));
}
@@ -448,6 +498,7 @@ impl CommunityRegistry {
let tx = self.signal_tx.clone();
let rx = self.signal_rx.clone();
let pages = self.pages.clone();
let watches = self.watches.clone();
let executor = cx.background_executor().clone();
self.notification_listener = Some(cx.background_spawn(async move {
@@ -457,7 +508,7 @@ impl CommunityRegistry {
'outer: loop {
match notifications.next().await {
Some(notification) => {
if route(notification, &pages, &mut batch) == Flow::Stop {
if route(notification, &pages, &watches, &mut batch) == Flow::Stop {
flush(&tx, &mut batch).await?;
break 'outer;
}
@@ -481,7 +532,7 @@ impl CommunityRegistry {
match select(next, timer).await {
Either::Left((Some(notification), _)) => {
if route(notification, &pages, &mut batch) == Flow::Stop {
if route(notification, &pages, &watches, &mut batch) == Flow::Stop {
flush(&tx, &mut batch).await?;
break 'outer;
}
@@ -501,6 +552,7 @@ impl CommunityRegistry {
while let Ok(signal) = rx.recv_async().await {
match signal {
Signal::Event(id) => this.update(cx, |this, cx| this.refresh(id, cx))?,
Signal::Rekey(id) => this.update(cx, |this, cx| this.rekey(id, cx))?,
Signal::List => this.update(cx, |this, cx| this.load(cx))?,
}
}
@@ -605,25 +657,53 @@ mod tests {
#[test]
fn a_burst_collapses_to_one_signal_per_community() {
let pages = PageRegistry::default();
let watches = WatchRegistry::default();
let community = CommunityId::from_bytes([0x42; 32]);
let mut batch = Batch::default();
let plane = event(sync::subscription_id(&community));
for _ in 0..50 {
assert_eq!(route(plane.clone(), &pages, &mut batch), Flow::Continue);
assert_eq!(
route(plane.clone(), &pages, &watches, &mut batch),
Flow::Continue
);
}
route(event(sync::list_subscription_id()), &pages, &mut batch);
route(
event(sync::list_subscription_id()),
&pages,
&watches,
&mut batch,
);
// A page's event is folded from the database later, not routed here.
route(
event(SubscriptionId::new("concord-history-7")),
&pages,
&watches,
&mut batch,
);
assert!(batch.list);
assert_eq!(batch.communities, BTreeSet::from([community]));
assert!(batch.rekeys.is_empty());
}
/// A rekey watch's wraps put themselves in the database; the pump's only job
/// is to wake the adoption pass, and it resolves the community by id.
#[test]
fn a_rekey_watch_event_wakes_its_community() {
let pages = PageRegistry::default();
let watches = WatchRegistry::default();
let community = CommunityId::from_bytes([0x42; 32]);
let id = rekey::subscription_id(&community);
watches.register(id.clone(), community);
let mut batch = Batch::default();
route(event(id), &pages, &watches, &mut batch);
assert_eq!(batch.rekeys, BTreeSet::from([community]));
assert!(batch.communities.is_empty());
}
#[test]
@@ -637,7 +717,12 @@ mod tests {
pages.register(other, other_tx);
let mut batch = Batch::default();
let flow = route(message(RelayMessage::eose(mine)), &pages, &mut batch);
let flow = route(
message(RelayMessage::eose(mine)),
&pages,
&WatchRegistry::default(),
&mut batch,
);
assert_eq!(flow, Flow::Continue);
assert!(matches!(
@@ -658,6 +743,7 @@ mod tests {
route(
message(RelayMessage::closed(id, "blocked: not allowed")),
&pages,
&WatchRegistry::default(),
&mut batch,
);
@@ -683,6 +769,7 @@ mod tests {
"auth-required: please authenticate",
)),
&pages,
&WatchRegistry::default(),
&mut batch,
);
@@ -695,7 +782,12 @@ mod tests {
let mut batch = Batch::default();
assert_eq!(
route(ClientNotification::Shutdown, &pages, &mut batch),
route(
ClientNotification::Shutdown,
&pages,
&WatchRegistry::default(),
&mut batch
),
Flow::Stop
);
}
crates/community/src/rekey.rs
+729
View File
@@ -0,0 +1,729 @@
use std::collections::{BTreeMap, BTreeSet, HashMap};
use std::sync::{Arc, Mutex, MutexGuard};
use anyhow::Result;
use concord::cord04::roles::{CommunityRoles, Permissions};
use concord::cord06::{self, Continuity, RekeyScope, Rotation, RotationKey};
use concord::derive::{GroupKey, channel_rekey_group_key};
use concord::state::{CommunityState, HeldKey};
use concord::{ChannelId, CommunityId, Epoch, cord01};
use nostr_sdk::prelude::*;
use state::UniversalSigner;
/// Epochs ahead of a held epoch a rotation is looked for.
pub const REKEY_LOOKAHEAD: u64 = 8;
/// How much of what a relay stores a rekey watch replays.
const REKEY_REPLAY: usize = 200;
/// One address the rekey watch asks a relay for, and what a wrap at it means.
#[derive(Debug, Clone)]
pub struct Watch {
pub address: PublicKey,
pub group: GroupKey,
pub scope: RekeyScope,
pub epoch: Epoch,
}
/// Every address a community's rotations can arrive at.
pub fn watches(state: &CommunityState) -> Result<Vec<Watch>> {
let mut watches = Vec::new();
let roots = state.roots();
let next = Epoch(state.root_epoch.0 + 1);
let group = cord06::rekey_group(RekeyScope::Base, &state.community_root, &state.id, next)?;
watches.push(Watch {
address: group.pk(),
group,
scope: RekeyScope::Base,
epoch: next,
});
for channel in &state.channels {
if !channel.private {
continue;
}
for root in &roots {
for ahead in 1..=REKEY_LOOKAHEAD {
let epoch = Epoch(channel.epoch.0 + ahead);
let group = channel_rekey_group_key(&root.key, &channel.id, epoch)?;
watches.push(Watch {
address: group.pk(),
group,
scope: RekeyScope::Channel(channel.id),
epoch,
});
}
}
}
Ok(watches)
}
pub fn watch_filter(watches: &[Watch]) -> Filter {
Filter::new()
.kind(Kind::GiftWrap)
.authors(watches.iter().map(|watch| watch.address))
.limit(REKEY_REPLAY)
}
/// The subscription carrying a community's rekey watch.
pub fn subscription_id(id: &CommunityId) -> SubscriptionId {
SubscriptionId::new(format!("rekey-{}", &id.to_hex()[..32]))
}
/// Which community a rekey watch subscription belongs to.
#[derive(Default)]
pub struct WatchRegistry {
watches: Arc<Mutex<HashMap<SubscriptionId, CommunityId>>>,
}
impl Clone for WatchRegistry {
fn clone(&self) -> Self {
Self {
watches: Arc::clone(&self.watches),
}
}
}
impl WatchRegistry {
pub fn register(&self, id: SubscriptionId, community: CommunityId) {
self.lock().insert(id, community);
}
pub fn community_of(&self, id: &SubscriptionId) -> Option<CommunityId> {
self.lock().get(id).copied()
}
/// Forget every watch, because the account that installed them is gone.
pub fn clear(&self) {
self.lock().clear();
}
fn lock(&self) -> MutexGuard<'_, HashMap<SubscriptionId, CommunityId>> {
match self.watches.lock() {
Ok(watches) => watches,
Err(poisoned) => poisoned.into_inner(),
}
}
}
/// What one adoption pass learned, ready to be folded into the held state.
#[derive(Debug, Default)]
pub struct Adoptions {
pub base: Option<BaseAdoption>,
pub channels: Vec<ChannelAdoption>,
/// Channels a complete rotation removed us from, with the epoch that did it.
pub cuts: Vec<(ChannelId, Epoch)>,
/// The base epoch a complete rotation excluded us at.
pub removed_at: Option<Epoch>,
pub stranded: bool,
}
impl Adoptions {
pub fn is_empty(&self) -> bool {
self.base.is_none()
&& self.channels.is_empty()
&& self.cuts.is_empty()
&& self.removed_at.is_none()
&& !self.stranded
}
}
#[derive(Debug, Clone)]
pub struct BaseAdoption {
pub epoch: Epoch,
pub key: [u8; 32],
pub control_pk: Option<PublicKey>,
pub stepped: Vec<HeldKey>,
}
#[derive(Debug, Clone)]
pub struct ChannelAdoption {
pub channel: ChannelId,
pub epoch: Epoch,
pub key: [u8; 32],
/// The keys this rotation stepped off, newest first.
pub stepped: Vec<HeldKey>,
}
/// Read every rekey wrap the local database holds and adopt what is admissible.
pub async fn adopt(
client: &Client,
state: &CommunityState,
roles: &CommunityRoles,
signer: &UniversalSigner,
me: PublicKey,
) -> Result<Adoptions> {
let watches = watches(state)?;
let mut authors: BTreeSet<PublicKey> = BTreeSet::new();
for watch in &watches {
authors.insert(watch.address);
}
if authors.is_empty() {
return Ok(Adoptions::default());
}
let wraps = client
.database()
.query(Filter::new().kind(Kind::GiftWrap).authors(authors))
.await?;
let mut chunks = Vec::new();
let mut published: BTreeMap<RotationKey, u64> = BTreeMap::new();
for wrap in &wraps {
let Some(watch) = watches.iter().find(|watch| watch.address == wrap.pubkey) else {
continue;
};
let Ok(opened) = cord01::open_wrap(wrap, &watch.group) else {
continue;
};
let Ok(chunk) = cord06::parse_rekey_chunk(&opened) else {
continue;
};
if chunk.scope != watch.scope || chunk.new_epoch != watch.epoch {
continue;
}
let at_ms = wrap.created_at.as_secs().saturating_mul(1000);
published
.entry(chunk.correlation())
.and_modify(|held| *held = (*held).min(at_ms))
.or_insert(at_ms);
chunks.push(chunk);
}
if chunks.is_empty() {
return Ok(Adoptions::default());
}
let rotations = cord06::collect_rotations(&chunks);
let mut adoptions = Adoptions::default();
let base = walk(
RekeyScope::Base,
&[Permissions::BAN],
state.root_epoch,
state.community_root,
state,
roles,
signer,
me,
&rotations,
&published,
)
.await?;
if let Some(step) = base.adopted {
adoptions.base = Some(BaseAdoption {
epoch: step.epoch,
key: step.key,
control_pk: step.control_pk,
stepped: step.stepped,
});
}
adoptions.removed_at = base.removed_at;
adoptions.stranded = base.stranded;
for channel in &state.channels {
if !channel.private {
continue;
}
let Some((held_epoch, held_key)) = channel.current() else {
continue;
};
let step = walk(
RekeyScope::Channel(channel.id),
&[Permissions::MANAGE_CHANNELS, Permissions::BAN],
held_epoch,
held_key,
state,
roles,
signer,
me,
&rotations,
&published,
)
.await?;
if let Some(adopted) = step.adopted {
adoptions.channels.push(ChannelAdoption {
channel: channel.id,
epoch: adopted.epoch,
key: adopted.key,
stepped: adopted.stepped,
});
}
if let Some(epoch) = step.removed_at {
adoptions.cuts.push((channel.id, epoch));
}
}
Ok(adoptions)
}
/// The key grouping a rotation's chunks, recomputed from a collected rotation.
fn rotation_key(rotation: &Rotation) -> RotationKey {
(
rotation.rotator.to_bytes(),
rotation.scope.id32(),
rotation.new_epoch.0,
rotation.prev_commit,
)
}
/// What one scope's walk found.
#[derive(Debug, Default)]
struct Step {
adopted: Option<Adopted>,
removed_at: Option<Epoch>,
stranded: bool,
}
#[derive(Debug, Clone)]
struct Adopted {
epoch: Epoch,
key: [u8; 32],
control_pk: Option<PublicKey>,
stepped: Vec<HeldKey>,
}
/// Walk a scope's rotations forward, one epoch at a time, off the key held.
#[allow(clippy::too_many_arguments)]
async fn walk(
scope: RekeyScope,
permissions: &[u64],
mut held_epoch: Epoch,
mut held_key: [u8; 32],
state: &CommunityState,
roles: &CommunityRoles,
signer: &UniversalSigner,
me: PublicKey,
rotations: &[Rotation],
published: &BTreeMap<RotationKey, u64>,
) -> Result<Step> {
let mut step = Step::default();
let mut stepped: Vec<HeldKey> = Vec::new();
let ceiling = held_epoch.0 + REKEY_LOOKAHEAD;
loop {
let target = Epoch(held_epoch.0 + 1);
if target.0 > ceiling {
break;
}
let candidates: Vec<&Rotation> = rotations
.iter()
.filter(|rotation| {
rotation.scope == scope
&& rotation.new_epoch == target
&& rotation.is_complete()
&& !state.banned.contains(&rotation.rotator)
&& permissions
.iter()
.any(|bits| roles.is_authorized(&rotation.rotator, &state.owner, *bits))
})
.collect();
if candidates.is_empty() {
break;
}
let mut delivery: Option<([u8; 32], Option<PublicKey>, u64)> = None;
let mut addressed = false;
for rotation in candidates
.iter()
.filter(|rotation| rotation.continuity(held_epoch, &held_key) == Continuity::Extends)
{
let at_ms = published
.get(&rotation_key(rotation))
.copied()
.unwrap_or_default();
let blobs = cord06::find_my_blobs(
&rotation.blobs,
&rotation.rotator,
&me,
scope,
rotation.new_epoch,
)
.collect::<Vec<_>>();
if blobs.is_empty() {
continue;
}
addressed = true;
for blob in blobs {
let Ok(delivered) = cord06::open_blob(
signer,
&rotation.rotator,
scope,
rotation.new_epoch,
blob,
&state.id,
)
.await
else {
continue;
};
let control_pk = delivered
.control_pk
.and_then(|bytes| PublicKey::from_slice(&bytes).ok());
let raced = delivery.as_ref().map(|held| held.0);
// Racing rotations converge on the lowest new key.
if raced.is_none_or(|raced| delivered.new_key < raced) {
delivery = Some((delivered.new_key, control_pk, at_ms));
} else if let Some(held) = delivery.as_mut() {
held.2 = held.2.min(at_ms);
}
}
}
if let Some((key, control_pk, at_ms)) = delivery {
stepped.insert(
0,
HeldKey {
epoch: held_epoch,
key: held_key,
retired_at: Some(at_ms / 1000),
},
);
held_epoch = target;
held_key = key;
step.adopted = Some(Adopted {
epoch: target,
key,
control_pk,
stepped: stepped.clone(),
});
continue;
}
if addressed {
break;
}
let judged: Vec<&&Rotation> = candidates
.iter()
.filter(|rotation| rotation.continuity(held_epoch, &held_key) != Continuity::Fork)
.collect();
let published_at = |rotation: &&Rotation| {
published
.get(&rotation_key(rotation))
.copied()
.unwrap_or_default()
};
if judged.iter().any(|rotation| {
published_at(rotation) >= state.added_at_ms
&& permissions.iter().any(|bits| {
roles.can_act_on_member(&rotation.rotator, &state.owner, &me, *bits)
})
}) {
step.removed_at = Some(target);
} else if scope == RekeyScope::Base
&& judged
.iter()
.any(|rotation| published_at(rotation) < state.added_at_ms)
{
step.stranded = true;
}
break;
}
Ok(step)
}
#[cfg(test)]
mod tests {
use std::collections::BTreeMap;
use concord::cord04::roles::CommunityRoles;
use concord::cord06::{RekeyBlob, build_blob, build_rekey_chunks};
use concord::derive::{
base_rekey_group_key, channel_rekey_group_key, control_signer_group_key,
epoch_key_commitment,
};
use concord::state::ChannelKeyRef;
use nostr_memory::MemoryDatabase;
use super::*;
const AT_MS: u64 = 1_700_000_000_000;
const ROOT: [u8; 32] = [0x55; 32];
const NEW_ROOT: [u8; 32] = [0x66; 32];
const CHANNEL_KEY: [u8; 32] = [0x07; 32];
const NEW_CHANNEL_KEY: [u8; 32] = [0x08; 32];
fn client() -> Client {
ClientBuilder::default()
.database(MemoryDatabase::unbounded())
.build()
}
fn state(owner: PublicKey, id: CommunityId, channel: ChannelId) -> CommunityState {
CommunityState {
id,
name: Some("Room".to_owned()),
owner,
owner_salt: [0x01; 32],
community_root: ROOT,
root_epoch: Epoch(0),
control_root: None,
control_pks: BTreeMap::from([(0, owner)]),
channels: vec![ChannelKeyRef {
id: channel,
name: "staff".to_owned(),
private: true,
epoch: Epoch(0),
key: Some(CHANNEL_KEY),
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: AT_MS,
}
}
async fn store(client: &Client, wraps: &[Event]) {
for wrap in wraps {
client.database().save_event(wrap).await.expect("saves");
}
}
fn base_chunks(
owner: &Keys,
id: &CommunityId,
prior_commit: &[u8; 32],
blobs: &[RekeyBlob],
) -> Vec<Event> {
let group = base_rekey_group_key(&ROOT, id, Epoch(1)).expect("derives");
smol::block_on(build_rekey_chunks(
owner,
&group,
RekeyScope::Base,
Epoch(1),
Epoch(0),
prior_commit,
blobs,
None,
false,
AT_MS / 1000,
))
.expect("builds")
}
fn channel_chunks(
owner: &Keys,
channel: &ChannelId,
prior_commit: &[u8; 32],
blobs: &[RekeyBlob],
) -> Vec<Event> {
let scope = RekeyScope::Channel(*channel);
let group = channel_rekey_group_key(&ROOT, channel, Epoch(1)).expect("derives");
smol::block_on(build_rekey_chunks(
owner,
&group,
scope,
Epoch(1),
Epoch(0),
prior_commit,
blobs,
None,
false,
AT_MS / 1000,
))
.expect("builds")
}
fn blob_for(
rotator: &Keys,
recipient: &Keys,
scope: RekeyScope,
key: [u8; 32],
control_pk: Option<&[u8; 32]>,
) -> RekeyBlob {
smol::block_on(build_blob(
rotator,
&recipient.public_key(),
scope,
Epoch(1),
&key,
control_pk,
None,
))
.expect("builds")
}
#[test]
fn a_complete_base_rotation_is_adopted_and_retires_the_prior_root() {
smol::block_on(async {
let client = client();
let owner = Keys::generate();
let me = Keys::generate();
let id = CommunityId::from_bytes([0x42; 32]);
let channel = ChannelId::from_bytes([0x9c; 32]);
let state = state(owner.public_key(), id, channel);
let control_root = [0xAB; 32];
let control_pk = control_signer_group_key(&control_root, &id, Epoch(1))
.expect("derives")
.pk()
.to_bytes();
let blob = blob_for(&owner, &me, RekeyScope::Base, NEW_ROOT, Some(&control_pk));
let wraps = base_chunks(&owner, &id, &epoch_key_commitment(Epoch(0), &ROOT), &[blob]);
store(&client, &wraps).await;
let signer = UniversalSigner::new(me.clone());
let roles = CommunityRoles::default();
let adoptions = adopt(&client, &state, &roles, &signer, me.public_key())
.await
.expect("reads");
let base = adoptions.base.expect("adopted");
assert_eq!(base.epoch, Epoch(1));
assert_eq!(base.key, NEW_ROOT);
assert_eq!(base.control_pk, PublicKey::from_slice(&control_pk).ok());
assert_eq!(base.stepped.len(), 1);
assert_eq!(base.stepped[0].epoch, Epoch(0));
assert_eq!(base.stepped[0].key, ROOT);
assert_eq!(base.stepped[0].retired_at, Some(AT_MS / 1000));
assert!(adoptions.removed_at.is_none());
assert!(!adoptions.stranded);
});
}
/// A rotation off a key we do not hold is a fork: adoptable by nobody here.
#[test]
fn a_rotation_that_does_not_extend_the_held_key_is_never_adopted() {
smol::block_on(async {
let client = client();
let owner = Keys::generate();
let me = Keys::generate();
let id = CommunityId::from_bytes([0x42; 32]);
let channel = ChannelId::from_bytes([0x9c; 32]);
let state = state(owner.public_key(), id, channel);
let blob = blob_for(&owner, &me, RekeyScope::Base, NEW_ROOT, None);
let wraps = base_chunks(
&owner,
&id,
&epoch_key_commitment(Epoch(0), &[0x99; 32]),
&[blob],
);
store(&client, &wraps).await;
let signer = UniversalSigner::new(me.clone());
let roles = CommunityRoles::default();
let adoptions = adopt(&client, &state, &roles, &signer, me.public_key())
.await
.expect("reads");
assert!(adoptions.is_empty());
});
}
/// Every chunk held, none carrying my blob, from a rotator who outranks me
/// and published after I joined: I was excluded, not stranded.
#[test]
fn a_blobless_rotation_from_an_outranking_rotator_removes_the_member() {
smol::block_on(async {
let client = client();
let owner = Keys::generate();
let me = Keys::generate();
let other = Keys::generate();
let id = CommunityId::from_bytes([0x42; 32]);
let channel = ChannelId::from_bytes([0x9c; 32]);
let state = state(owner.public_key(), id, channel);
let blob = blob_for(&owner, &other, RekeyScope::Base, NEW_ROOT, None);
let wraps = base_chunks(&owner, &id, &epoch_key_commitment(Epoch(0), &ROOT), &[blob]);
store(&client, &wraps).await;
let signer = UniversalSigner::new(me.clone());
let roles = CommunityRoles::default();
let adoptions = adopt(&client, &state, &roles, &signer, me.public_key())
.await
.expect("reads");
assert!(adoptions.base.is_none());
assert_eq!(adoptions.removed_at, Some(Epoch(1)));
assert!(!adoptions.stranded);
});
}
#[test]
fn a_channel_rotation_replaces_the_key_and_keeps_the_prior() {
smol::block_on(async {
let client = client();
let owner = Keys::generate();
let me = Keys::generate();
let id = CommunityId::from_bytes([0x42; 32]);
let channel = ChannelId::from_bytes([0x9c; 32]);
let state = state(owner.public_key(), id, channel);
let blob = blob_for(
&owner,
&me,
RekeyScope::Channel(channel),
NEW_CHANNEL_KEY,
None,
);
let wraps = channel_chunks(
&owner,
&channel,
&epoch_key_commitment(Epoch(0), &CHANNEL_KEY),
&[blob],
);
store(&client, &wraps).await;
let signer = UniversalSigner::new(me.clone());
let roles = CommunityRoles::default();
let adoptions = adopt(&client, &state, &roles, &signer, me.public_key())
.await
.expect("reads");
assert_eq!(adoptions.channels.len(), 1);
let adopted = &adoptions.channels[0];
assert_eq!(adopted.channel, channel);
assert_eq!(adopted.epoch, Epoch(1));
assert_eq!(adopted.key, NEW_CHANNEL_KEY);
assert_eq!(adopted.stepped.len(), 1);
assert_eq!(adopted.stepped[0].epoch, Epoch(0));
assert_eq!(adopted.stepped[0].key, CHANNEL_KEY);
assert_eq!(adopted.stepped[0].retired_at, Some(AT_MS / 1000));
assert!(adoptions.cuts.is_empty());
});
}
}
crates/community/src/sync.rs
+137 -21
View File
@@ -10,7 +10,7 @@ use concord::cord04::roles::{Permissions, citation_ok};
use concord::derive::{
channel_group_key, control_group_key, control_signer_group_key, guestbook_group_key,
};
use concord::state::{CommunityState, list_entry};
use concord::state::{CommunityState, HeldKey, HeldRoot, list_entry};
use concord::{ChannelId, CommunityId, Epoch, GroupKey};
use gpui::AsyncApp;
use nostr_sdk::prelude::*;
@@ -35,18 +35,42 @@ pub struct Plane {
/// The wrap's author: the control signer for Control, the group's own key otherwise.
pub address: PublicKey,
pub group: GroupKey,
/// Epoch seconds the key behind this plane was retired.
pub retired_at: Option<u64>,
}
impl Plane {
/// Whether a wrap sealed under this plane is still inside its key's life.
pub fn accepts(&self, wrap: &Event) -> bool {
self.retired_at
.is_none_or(|retired| wrap.created_at.as_secs() <= retired)
}
}
pub fn planes(state: &CommunityState) -> Result<Vec<Plane>> {
let mut planes = Vec::new();
let roots = state.roots();
for (epoch, address) in &state.control_pks {
let epoch = Epoch(*epoch);
let group = control_group_key(&state.community_root, &state.id, epoch)?;
// An epoch's Control Plane reads under the root that was current then,
// so a rotation that kept a floor for the prior root republishes here.
let root = roots
.iter()
.find(|root| root.epoch == epoch)
.copied()
.unwrap_or(HeldRoot {
epoch,
key: state.community_root,
control_pk: None,
retired_at: None,
});
let group = control_group_key(&root.key, &state.id, epoch)?;
planes.push(Plane {
kind: PlaneKind::Control(epoch),
address: *address,
group,
retired_at: root.retired_at,
});
}
@@ -56,31 +80,38 @@ pub fn planes(state: &CommunityState) -> Result<Vec<Plane>> {
kind: PlaneKind::Control(state.root_epoch),
address: group.pk(),
group,
retired_at: None,
});
}
let group = guestbook_group_key(&state.community_root, &state.id, state.root_epoch)?;
planes.push(Plane {
kind: PlaneKind::Guestbook,
address: group.pk(),
group,
});
for channel in &state.channels {
let secret = match (channel.private, channel.key) {
(true, None) => continue,
(true, Some(key)) => key,
(false, _) => state.community_root,
};
let group = channel_group_key(&secret, &channel.id, channel.epoch)?;
for root in &roots {
let group = guestbook_group_key(&root.key, &state.id, root.epoch)?;
planes.push(Plane {
kind: PlaneKind::Channel(channel.id, channel.epoch),
kind: PlaneKind::Guestbook,
address: group.pk(),
group,
retired_at: root.retired_at,
});
}
for channel in &state.channels {
for held in state.held_keys(&channel.id) {
let group = channel_group_key(&held.key, &channel.id, held.epoch)?;
planes.push(Plane {
kind: PlaneKind::Channel(channel.id, held.epoch),
address: group.pk(),
group,
retired_at: held.retired_at,
});
}
}
// A rotation can re-derive an address the current root already produced.
//
// A duplicate author would only repeat a filter, so keep the set unique.
let mut seen = BTreeSet::new();
planes.retain(|plane| seen.insert(plane.address));
Ok(planes)
}
@@ -389,17 +420,44 @@ fn refresh(mut held: CommunityState, fresh: CommunityState) -> CommunityState {
held.relays = fresh.relays;
for channel in fresh.channels {
let cut = held.channel_cuts.get(&channel.id).copied();
match held.channels.iter_mut().find(|held| held.id == channel.id) {
Some(held) => {
held.name = channel.name;
held.epoch = channel.epoch;
if cut.is_some_and(|cut| channel.epoch.0 <= cut) {
continue;
}
if channel.private {
held.private = true;
held.key = channel.key;
if let Some(key) = channel.key
&& (held.key != Some(key) || held.epoch != channel.epoch)
{
// The key being superseded still reads everything
// written under it, so it is retained, never overwritten.
if let Some((epoch, previous)) = held.current()
&& !held.priors.iter().any(|prior| prior.epoch == epoch)
{
held.priors.push(HeldKey {
epoch,
key: previous,
retired_at: None,
});
}
held.key = Some(key);
held.epoch = channel.epoch;
}
}
}
None => {
if cut.is_none() {
held.channels.push(channel);
}
}
None => held.channels.push(channel),
}
}
@@ -486,6 +544,11 @@ 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(_) => {
if let Ok(edition) = cord02::open_edition(wrap, &plane.group, &plane.address, true)
@@ -609,6 +672,7 @@ mod tests {
private: false,
epoch: Epoch(0),
key: None,
priors: Vec::new(),
},
concord::state::ChannelKeyRef {
id: staff,
@@ -616,6 +680,7 @@ mod tests {
private: true,
epoch: Epoch(0),
key: Some([0x04; 32]),
priors: Vec::new(),
},
concord::state::ChannelKeyRef {
id: ChannelId::from_bytes([0x9e; 32]),
@@ -623,12 +688,17 @@ mod tests {
private: true,
epoch: Epoch(0),
key: None,
priors: Vec::new(),
},
],
relays: vec![RelayUrl::parse("wss://relay.example").expect("a url")],
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,
};
@@ -719,11 +789,16 @@ mod tests {
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: 1_700_000_000_000,
}
@@ -1021,4 +1096,45 @@ mod tests {
assert!(subscription_id(&id).as_str().len() <= 64);
}
fn event_at(at_ms: u64) -> Event {
EventBuilder::new(Kind::TextNote, "wrap")
.custom_created_at(Timestamp::from_secs(at_ms / 1000))
.finalize(&Keys::generate())
.expect("signs")
}
/// A key a rotation stepped off still reads its own history, but nothing
/// sealed after the rotation published it.
#[test]
fn a_superseded_key_reads_only_up_to_the_rotation_that_retired_it() {
let channel = ChannelId::from_bytes([0x9c; 32]);
let mut state = held(
CommunityId::from_bytes([0x42; 32]),
Keys::generate().public_key(),
);
state.channels = vec![concord::state::ChannelKeyRef {
id: channel,
name: "staff".to_owned(),
private: true,
epoch: Epoch(1),
key: Some([0x08; 32]),
priors: vec![HeldKey {
epoch: Epoch(0),
key: [0x07; 32],
retired_at: Some(1_000),
}],
}];
let planes = planes(&state).expect("planes");
let retired = planes
.iter()
.find(|plane| plane.kind == PlaneKind::Channel(channel, Epoch(0)))
.expect("the retired epoch keeps a plane");
assert_eq!(retired.retired_at, Some(1_000));
assert!(retired.accepts(&event_at(1_000_000)));
assert!(!retired.accepts(&event_at(1_001_000)));
}
}