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

updat

Browse Source
This commit is contained in:
reya
2026-09-22 16:38:08 +07:00
parent 6fac58d494
commit 1c5ef58049
11 changed files with 808 additions and 181 deletions
Download Patch File Download Diff File Expand all files Collapse all files
crates/community/src/community.rs
+89 -13
View File
@@ -5,8 +5,10 @@ use std::time::{Duration, Instant};
use anyhow::Result;
use concord::cord02::{ControlFold, ImageRef};
use concord::cord03::{self, ChatMessage, ReplyRef};
use concord::cord04::AuthorityCitation;
use concord::cord04::roles::{Permissions, citation_ok};
use concord::derive::channel_group_key;
use concord::cord06::RekeyScope;
use concord::derive::{channel_group_key, grant_locator};
use concord::state::{ChannelCursor, ChannelKeyRef, CommunityState, HeldKey, HeldRoot};
use concord::{ChannelId, CommunityId, Epoch};
use gpui::{App, AppContext, Context, EventEmitter, Task};
@@ -255,7 +257,7 @@ impl Community {
/// 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))
self.state.channel_cuts.get(channel).copied()
}
/// Whether an automatic catch-up for `channel` is worth asking for yet.
@@ -653,6 +655,78 @@ impl Community {
}));
}
/// Rotate one scope to its next epoch, cutting off `excluded`.
pub fn rotate(
&mut self,
scope: RekeyScope,
recipients: &[PublicKey],
excluded: &[PublicKey],
cx: &mut Context<Self>,
) -> Option<Task<Result<Epoch>>> {
let nostr = NostrRegistry::global(cx);
let me = nostr.read(cx).current_user()?;
let client = nostr.read(cx).client();
let signer = nostr.read(cx).signer();
if self.state.removed_at.is_some() || self.state.stranded || self.state.banned.contains(&me)
{
return None;
}
let rewrite = rekey::Rewrite {
scope,
recipients: recipients.to_vec(),
excluded: excluded.to_vec(),
citation: self.citation(&me),
};
if !rewrite.authorized(&self.control.roles, &self.state.owner, &me) {
return None;
}
let state = self.state.clone();
let roles = self.control.roles.clone();
Some(cx.spawn(async move |this, cx| {
let epoch = rekey::rotate(
&client,
&state,
&roles,
&signer,
me,
&rewrite,
Timestamp::now(),
)
.await?;
let adoptions = rekey::adopt(&client, &state, &roles, &signer, me).await?;
this.update(cx, |this, cx| {
if !adoptions.is_empty() {
this.merge_adoptions(adoptions, cx);
}
})?;
Ok(epoch)
}))
}
/// The rank this client cites when it acts.
fn citation(&self, me: &PublicKey) -> Option<AuthorityCitation> {
let entity = grant_locator(&self.state.id, &me.to_bytes());
self.state
.heads
.iter()
.find(|head| head.entity == entity)
.map(|head| AuthorityCitation {
entity: head.entity,
version: head.version,
hash: head.self_hash,
})
}
fn apply_rekey(&mut self, result: Result<Adoptions>, cx: &mut Context<Self>) {
self.rekey_task = None;
@@ -691,6 +765,8 @@ impl Community {
self.state.control_pks.insert(base.epoch.0, control_pk);
}
// The rotation is authoritative about the new epoch's signing root
self.state.control_root = base.control_root;
self.state.community_root = base.key;
self.state.root_epoch = base.epoch;
self.state.removed_at = None;
@@ -726,7 +802,7 @@ impl Community {
for (channel, epoch) in adoptions.cuts {
self.state.channels.retain(|held| held.id != channel);
self.state.channel_cuts.insert(channel, epoch.0);
self.state.channel_cuts.insert(channel, epoch);
self.state.cursors.remove(&channel);
}
@@ -934,7 +1010,7 @@ async fn sync_round(
Intent::Older { .. } => WrapPage::default(),
};
let bridge = match (intent, newest.oldest_ms, saved.newest_ms) {
let bridge = match (intent, newest.oldest, saved.newest) {
(Intent::CatchUp, Some(oldest), Some(saved_newest)) if oldest > saved_newest => {
let page = history::page(
client,
@@ -958,8 +1034,8 @@ async fn sync_round(
};
let resume = match intent {
Intent::CatchUp => saved.oldest_ms.or(newest.oldest_ms),
Intent::Older { .. } => saved.oldest_ms,
Intent::CatchUp => saved.oldest.or(newest.oldest),
Intent::Older { .. } => saved.oldest,
};
let budget = match intent {
@@ -994,16 +1070,16 @@ async fn sync_round(
if intent == Intent::CatchUp {
let complete = !newest.failed && bridge.exhausted;
let top = newest
.newest_ms
.unwrap_or(0)
.max(bridge.newest_ms.unwrap_or(0));
.newest
.unwrap_or_default()
.max(bridge.newest.unwrap_or_default());
if complete && top > 0 {
round.newest_ms = Some(top);
if complete && !top.is_zero() {
round.newest = Some(top);
}
}
round.oldest_ms = older.oldest_ms.or(newest.oldest_ms);
round.oldest = older.oldest.or(newest.oldest);
round.exhausted = older.exhausted;
Ok((progress, round))
@@ -1082,7 +1158,7 @@ mod tests {
// ...and a channel cut closes the one channel.
let mut cut = state(channel);
cut.channel_cuts.insert(channel, 0);
cut.channel_cuts.insert(channel, Epoch(0));
assert_eq!(community(cut).channel_secret(&channel), None);
}
}
crates/community/src/history.rs
+58 -56
View File
@@ -19,13 +19,13 @@ use crate::sync::connect_relays;
/// How long one relay is given to answer one page of history.
const PAGE_TIMEOUT: Duration = Duration::from_secs(10);
/// How far below a cursor a warm window reaches back.
pub const CURSOR_OVERLAP_MS: u64 = 60_000;
pub const CURSOR_OVERLAP: Duration = Duration::from_secs(60);
/// The region of history to read.
#[derive(Debug, Clone, Copy, Default, PartialEq, Eq)]
pub struct Window {
pub until_ms: Option<u64>,
pub since_ms: Option<u64>,
pub until: Option<Timestamp>,
pub since: Option<Timestamp>,
}
impl Window {
@@ -34,28 +34,28 @@ impl Window {
Self::default()
}
/// The wraps strictly older than `oldest_ms`.
pub fn older_than(oldest_ms: u64) -> Self {
/// The wraps strictly older than `oldest`.
pub fn older_than(oldest: Timestamp) -> Self {
Self {
until_ms: Some(oldest_ms.saturating_sub(1)),
since_ms: None,
until: Some(oldest - 1u64),
since: None,
}
}
/// The region between `since_ms` and `oldest_ms`, both inclusive.
pub fn between(since_ms: u64, oldest_ms: u64) -> Self {
/// The region between `since` and `oldest`, both inclusive.
pub fn between(since: Timestamp, oldest: Timestamp) -> Self {
Self {
until_ms: Some(oldest_ms.saturating_sub(1)),
since_ms: Some(since_ms),
until: Some(oldest - 1u64),
since: Some(since),
}
}
/// The window a channel is opened with.
pub fn opening(cursor: ChannelCursor) -> Self {
match cursor.newest_ms {
Some(newest_ms) => Self {
since_ms: Some(newest_ms.saturating_sub(CURSOR_OVERLAP_MS)),
until_ms: None,
match cursor.newest {
Some(newest) => Self {
since: Some(newest - CURSOR_OVERLAP),
until: None,
},
None => Self::default(),
}
@@ -139,8 +139,8 @@ pub struct WrapPage {
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 newest: Option<Timestamp>,
pub oldest: Option<Timestamp>,
pub exhausted: bool,
pub failed: bool,
pub errors: usize,
@@ -235,7 +235,7 @@ fn read_under(
) -> Result<(OpenedStream, ChatRumor)> {
if held
.retired_at
.is_some_and(|retired| wrap.created_at.as_secs() > retired)
.is_some_and(|retired| wrap.created_at > retired)
{
bail!("sealed after the key that reads it was retired");
}
@@ -250,12 +250,12 @@ fn wrap_filter(authors: &[PublicKey], window: Window, limit: usize) -> Filter {
.authors(authors.iter().copied())
.limit(limit);
if let Some(until_ms) = window.until_ms {
filter = filter.until(Timestamp::from_secs(until_ms / 1000));
if let Some(until) = window.until {
filter = filter.until(until);
}
if let Some(since_ms) = window.since_ms {
filter = filter.since(Timestamp::from_secs(since_ms / 1000));
if let Some(since) = window.since {
filter = filter.since(since);
}
filter
@@ -373,12 +373,12 @@ where
#[derive(Debug)]
struct Walk {
relays: Vec<Walker>,
since_ms: Option<u64>,
since: Option<Timestamp>,
/// The inclusive upper bound of the next page.
cursor: Option<u64>,
cursor: Option<Timestamp>,
seen: BTreeSet<EventId>,
newest_ms: Option<u64>,
oldest_ms: Option<u64>,
newest: Option<Timestamp>,
oldest: Option<Timestamp>,
raw: usize,
errors: usize,
/// Wraps the caller could not read under any held key.
@@ -402,11 +402,11 @@ impl Walk {
.cloned()
.map(|url| Walker { url, dead: false })
.collect(),
since_ms: window.since_ms,
cursor: window.until_ms,
since: window.since,
cursor: window.until,
seen: BTreeSet::new(),
newest_ms: None,
oldest_ms: None,
newest: None,
oldest: None,
raw: 0,
errors: 0,
unreadable: 0,
@@ -421,8 +421,8 @@ impl Walk {
/// The region the next page asks for.
fn region(&self) -> Window {
Window {
until_ms: self.cursor,
since_ms: self.since_ms,
until: self.cursor,
since: self.since,
}
}
@@ -444,13 +444,13 @@ impl Walk {
self.bottom = true;
}
let mut oldest: Option<u64> = None;
let mut oldest: Option<Timestamp> = None;
let mut events = Vec::with_capacity(page.len());
for event in page {
let at_ms = event.created_at.as_secs().saturating_mul(1000);
self.newest_ms = Some(self.newest_ms.map_or(at_ms, |newest| newest.max(at_ms)));
oldest = Some(oldest.map_or(at_ms, |oldest| oldest.min(at_ms)));
let at = event.created_at;
self.newest = Some(self.newest.map_or(at, |newest| newest.max(at)));
oldest = Some(oldest.map_or(at, |oldest| oldest.min(at)));
if self.seen.insert(event.id) {
self.raw += 1;
@@ -459,7 +459,7 @@ impl Walk {
}
match oldest {
Some(oldest) if oldest > 0 => self.cursor = Some(oldest - 1),
Some(oldest) if !oldest.is_zero() => self.cursor = Some(oldest - 1u64),
Some(_) => self.bottom = true,
None => {}
}
@@ -474,8 +474,8 @@ impl Walk {
opened,
raw: self.raw,
unreadable: self.unreadable,
newest_ms: self.newest_ms,
oldest_ms: self.oldest_ms,
newest: self.newest,
oldest: self.oldest,
exhausted: swept && self.raw > 0,
failed: self.errors > 0 || (self.bottom && self.raw == 0),
errors: self.errors,
@@ -500,9 +500,8 @@ mod tests {
let mut events: Vec<Event> = database
.iter()
.filter(|event| {
let at_ms = event.created_at.as_secs().saturating_mul(1000);
window.until_ms.is_none_or(|until| at_ms <= until)
&& window.since_ms.is_none_or(|since| at_ms >= since)
window.until.is_none_or(|until| event.created_at <= until)
&& window.since.is_none_or(|since| event.created_at >= since)
})
.cloned()
.collect();
@@ -613,7 +612,7 @@ mod tests {
let held = HeldKey {
epoch: Epoch(0),
key: SECRET,
retired_at: Some(1_000),
retired_at: Some(Timestamp::from_secs(1_000)),
};
let wrap_at = |channel: &ChannelId, at_ms: u64| {
@@ -656,7 +655,7 @@ mod tests {
assert!(page.failed);
assert!(!page.exhausted);
assert_eq!(page.raw, 0);
assert_eq!(page.oldest_ms, None);
assert_eq!(page.oldest, None);
}
#[test]
@@ -681,7 +680,10 @@ mod tests {
#[test]
fn a_page_boundary_is_exclusive() {
let database = BTreeSet::from([event_at(1_700_000_000_000), event_at(1_700_000_001_000)]);
let database = BTreeSet::from([
event_at(Timestamp::from_secs(1_700_000_000)),
event_at(Timestamp::from_secs(1_700_000_001)),
]);
let mut walk = Walk::new(&[relay_url("history")], Window::newest());
let first = walk.accept(serve_page(&database, walk.region(), 1), 1);
@@ -693,16 +695,16 @@ mod tests {
let oldest = second
.iter()
.map(|event| event.created_at.as_secs() * 1000)
.map(|event| event.created_at)
.min()
.expect("one wrap");
assert_eq!(oldest, 1_700_000_000_000);
assert_eq!(oldest, Timestamp::from_secs(1_700_000_000));
}
fn event_at(at_ms: u64) -> Event {
fn event_at(at: Timestamp) -> Event {
let keys = Keys::generate();
EventBuilder::new(Kind::TextNote, "page")
.custom_created_at(Timestamp::from_secs(at_ms / 1000))
.custom_created_at(at)
.finalize(&keys)
.expect("signs")
}
@@ -714,23 +716,23 @@ mod tests {
assert_eq!(Window::opening(ChannelCursor::default()), Window::default());
let warm = Window::opening(ChannelCursor {
newest_ms: Some(2_000_000),
oldest_ms: Some(1_000),
newest: Some(Timestamp::from_secs(2_000_000)),
oldest: Some(Timestamp::from_secs(1_000)),
exhausted: false,
});
assert_eq!(
warm,
Window {
since_ms: Some(2_000_000 - CURSOR_OVERLAP_MS),
until_ms: None,
since: Some(Timestamp::from_secs(2_000_000) - CURSOR_OVERLAP),
until: None,
}
);
assert_eq!(
Window::older_than(1_000),
Window::older_than(Timestamp::from_secs(1_000)),
Window {
since_ms: None,
until_ms: Some(999),
since: None,
until: Some(Timestamp::from_secs(999)),
}
);
}
crates/community/src/rekey.rs
+297 -17
View File
@@ -1,15 +1,21 @@
use std::collections::{BTreeMap, BTreeSet, HashMap};
use std::sync::{Arc, Mutex, MutexGuard};
use anyhow::Result;
use anyhow::{Result, bail};
use concord::cord01::{self, KIND_WRAP_EPHEMERAL, SealForm};
use concord::cord04::roles::{CommunityRoles, Permissions};
use concord::cord06::{self, Continuity, RekeyScope, Rotation, RotationKey};
use concord::derive::{GroupKey, channel_rekey_group_key};
use concord::cord04::{self, AuthorityCitation};
use concord::cord06::{
self, Continuity, Refounding, RekeyScope, Rotation, RotationKey, RotationPlan,
};
use concord::derive::{GroupKey, channel_rekey_group_key, epoch_key_commitment};
use concord::state::{CommunityState, HeldKey};
use concord::{ChannelId, CommunityId, Epoch, cord01};
use concord::{ChannelId, CommunityId, Epoch};
use nostr_sdk::prelude::*;
use state::UniversalSigner;
use crate::sync::{self, PlaneKind};
/// 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.
@@ -24,6 +30,15 @@ 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],
RekeyScope::Channel(_) => &[Permissions::MANAGE_CHANNELS, Permissions::BAN],
}
}
/// Every address a community's rotations can arrive at.
pub fn watches(state: &CommunityState) -> Result<Vec<Watch>> {
let mut watches = Vec::new();
@@ -135,6 +150,8 @@ pub struct BaseAdoption {
pub epoch: Epoch,
pub key: [u8; 32],
pub control_pk: Option<PublicKey>,
/// The new Control Plane signing root, delivered to staff only.
pub control_root: Option<[u8; 32]>,
pub stepped: Vec<HeldKey>,
}
@@ -209,7 +226,7 @@ pub async fn adopt(
let base = walk(
RekeyScope::Base,
&[Permissions::BAN],
permissions(RekeyScope::Base),
state.root_epoch,
state.community_root,
state,
@@ -226,9 +243,11 @@ pub async fn adopt(
epoch: step.epoch,
key: step.key,
control_pk: step.control_pk,
control_root: step.control_root,
stepped: step.stepped,
});
}
adoptions.removed_at = base.removed_at;
adoptions.stranded = base.stranded;
@@ -243,7 +262,7 @@ pub async fn adopt(
let step = walk(
RekeyScope::Channel(channel.id),
&[Permissions::MANAGE_CHANNELS, Permissions::BAN],
permissions(RekeyScope::Channel(channel.id)),
held_epoch,
held_key,
state,
@@ -272,6 +291,242 @@ pub async fn adopt(
Ok(adoptions)
}
/// A rotation this client is asked to perform.
#[derive(Debug, Clone)]
pub struct Rewrite {
pub scope: RekeyScope,
/// Every member the new key must reach.
pub recipients: Vec<PublicKey>,
/// The members it must not reach: whoever the rotation cuts off.
pub excluded: Vec<PublicKey>,
/// The rotator's claim to rank, when it holds the grant to cite.
pub citation: Option<AuthorityCitation>,
}
impl Rewrite {
/// Whether `rotator` may cut `excluded` off at all.
pub fn authorized(
&self,
roles: &CommunityRoles,
owner: &PublicKey,
rotator: &PublicKey,
) -> bool {
permissions(self.scope)
.iter()
.any(|bits| cord06::rekey_authorized(roles, owner, rotator, *bits, &self.excluded))
}
}
/// Publish this client's own rotation of one scope to its next epoch.
#[allow(clippy::too_many_arguments)]
pub async fn rotate(
client: &Client,
state: &CommunityState,
roles: &CommunityRoles,
signer: &UniversalSigner,
me: PublicKey,
rewrite: &Rewrite,
at: Timestamp,
) -> Result<Epoch> {
if rewrite.excluded.contains(&me) {
bail!("a rotation cannot cut off the member who publishes it");
}
if !rewrite.recipients.contains(&me) {
bail!("a rotation must deliver its own rotator");
}
if rewrite
.excluded
.iter()
.any(|target| rewrite.recipients.contains(target))
{
bail!("a rotation cannot both deliver to and cut off the same member");
}
let (held_epoch, held_key) = stepping_off(state, rewrite.scope)?;
let epoch = Epoch(held_epoch.0 + 1);
if epoch.0 > cord06::MAX_REKEY_EPOCH {
bail!("epoch {} is past the rekey ceiling", epoch.0);
}
let plan = cord06::plan_rotation(rewrite.scope, epoch)?;
let new_key = plan.new_key();
let control_pk = match &plan {
RotationPlan::Base(refounding) => Some(refounding.signer(&state.id)?.pk().to_bytes()),
RotationPlan::Channel { .. } => None,
};
let mut blobs = Vec::with_capacity(rewrite.recipients.len());
for recipient in &rewrite.recipients {
// A refounded Control Plane's root reaches staff only.
let control_root = match &plan {
RotationPlan::Base(refounding) => roles
.is_staff(recipient, &state.owner)
.then_some(refounding.new_control_root),
RotationPlan::Channel { .. } => None,
};
blobs.push(
cord06::build_blob(
signer,
recipient,
rewrite.scope,
epoch,
&new_key,
control_pk.as_ref(),
control_root.as_ref(),
)
.await?,
);
}
let group = cord06::rekey_group(rewrite.scope, &state.community_root, &state.id, epoch)?;
// A rotation that cuts somebody off is the severed kind.
let severed = !rewrite.excluded.is_empty();
let mut wraps = cord06::build_rekey_chunks(
signer,
&group,
rewrite.scope,
epoch,
held_epoch,
&epoch_key_commitment(held_epoch, &held_key),
&blobs,
rewrite.citation.as_ref(),
severed,
at.as_secs(),
)
.await?;
if let RotationPlan::Base(refounding) = &plan {
let carried = carry_heads(client, state, refounding, at).await?;
if carried.is_empty() && !state.heads.is_empty() {
log::warn!("community: a refounding carried no settled head forward");
}
wraps.extend(carried);
}
sync::publish_wraps(client, &wraps, &state.relays).await;
log::debug!(
"community: rotated epoch {} to {} for {} member(s)",
held_epoch.0,
epoch.0,
blobs.len()
);
Ok(epoch)
}
/// The epoch and key a rotation steps off.
fn stepping_off(state: &CommunityState, scope: RekeyScope) -> Result<(Epoch, [u8; 32])> {
match scope {
RekeyScope::Base => Ok((state.root_epoch, state.community_root)),
RekeyScope::Channel(channel) => state
.channels
.iter()
.find(|held| held.id == channel)
.and_then(|held| held.current())
.ok_or_else(|| anyhow::anyhow!("no current key is held for {}", channel.to_hex())),
}
}
/// Re-seal the settled control heads under a refounding's new groups.
///
/// A rotation only mints keys. Unless the heads ride across with it, the new
/// epoch's Control Plane starts empty, and a member whose material never carried
/// the root we are stepping off folds no roles, metadata or banlist at all.
async fn carry_heads(
client: &Client,
state: &CommunityState,
refounding: &Refounding,
at: Timestamp,
) -> Result<Vec<Event>> {
if state.heads.is_empty() {
return Ok(Vec::new());
}
let planes = sync::planes(state)?;
let authors: BTreeSet<PublicKey> = planes
.iter()
.filter(|plane| matches!(plane.kind, PlaneKind::Control(_)))
.map(|plane| plane.address)
.collect();
if authors.is_empty() {
return Ok(Vec::new());
}
let wraps = client
.database()
.query(
Filter::new()
.kinds([Kind::GiftWrap, Kind::Custom(KIND_WRAP_EPHEMERAL)])
.authors(authors),
)
.await?;
let mut seals = Vec::new();
let mut seen: BTreeSet<EventId> = BTreeSet::new();
for wrap in &wraps {
let Some(plane) = planes.iter().find(|plane| {
matches!(plane.kind, PlaneKind::Control(_)) && plane.address == wrap.pubkey
}) else {
continue;
};
// A retired root reads only what was sealed before its rotation.
if !plane.accepts(wrap) {
continue;
}
let Ok(opened) =
cord01::open_wrap_at(wrap, &plane.address, plane.group.conversation(), true)
else {
continue;
};
// Only a plaintext seal can be carried: `compact` re-signs nothing and
// re-wraps the edition the author already sealed.
if opened.seal_form != SealForm::Plaintext {
continue;
}
let Ok(edition) = cord04::parse_edition(&opened.rumor) else {
continue;
};
let settled = state.heads.iter().any(|head| {
head.entity == edition.entity
&& head.version == edition.version
&& head.self_hash == edition.self_hash
});
// The same edition can be reachable twice once a head has been carried
// forward before: a head is published once per refounding.
if settled && seen.insert(opened.seal.id) {
seals.push(opened.seal);
}
}
if seals.is_empty() {
return Ok(Vec::new());
}
Ok(cord06::compact(
&seals,
&refounding.read(&state.id)?,
&refounding.signer(&state.id)?,
at.as_secs(),
)?)
}
/// The key grouping a rotation's chunks, recomputed from a collected rotation.
fn rotation_key(rotation: &Rotation) -> RotationKey {
(
@@ -295,9 +550,19 @@ struct Adopted {
epoch: Epoch,
key: [u8; 32],
control_pk: Option<PublicKey>,
control_root: Option<[u8; 32]>,
stepped: Vec<HeldKey>,
}
/// What one rotation offered this client, and when it was published.
#[derive(Debug, Clone, Copy)]
struct Delivery {
key: [u8; 32],
control_pk: Option<PublicKey>,
control_root: Option<[u8; 32]>,
at_ms: u64,
}
/// Walk a scope's rotations forward, one epoch at a time, off the key held.
#[allow(clippy::too_many_arguments)]
async fn walk(
@@ -340,7 +605,7 @@ async fn walk(
break;
}
let mut delivery: Option<([u8; 32], Option<PublicKey>, u64)> = None;
let mut delivery: Option<Delivery> = None;
let mut addressed = false;
for rotation in candidates
@@ -384,34 +649,43 @@ async fn walk(
let control_pk = delivered
.control_pk
.and_then(|bytes| PublicKey::from_slice(&bytes).ok());
let raced = delivery.as_ref().map(|held| held.0);
let raced = delivery.map(|held| held.key);
// 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));
delivery = Some(Delivery {
key: delivered.new_key,
control_pk,
control_root: delivered.control_root,
at_ms,
});
} else if let Some(held) = delivery.as_mut() {
held.2 = held.2.min(at_ms);
held.at_ms = held.at_ms.min(at_ms);
}
}
}
if let Some((key, control_pk, at_ms)) = delivery {
if let Some(delivered) = delivery {
stepped.insert(
0,
HeldKey {
epoch: held_epoch,
key: held_key,
retired_at: Some(at_ms / 1000),
// The cutoff is a second-granular read boundary, so the
// rotation's millisecond publish time narrows to the second
// it landed in.
retired_at: Some(Timestamp::from_secs(delivered.at_ms / 1000)),
},
);
held_epoch = target;
held_key = key;
held_key = delivered.key;
step.adopted = Some(Adopted {
epoch: target,
key,
control_pk,
key: delivered.key,
control_pk: delivered.control_pk,
control_root: delivered.control_root,
stepped: stepped.clone(),
});
@@ -618,7 +892,10 @@ mod tests {
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_eq!(
base.stepped[0].retired_at,
Some(Timestamp::from_secs(AT_MS / 1000))
);
assert!(adoptions.removed_at.is_none());
assert!(!adoptions.stranded);
});
@@ -722,7 +999,10 @@ mod tests {
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_eq!(
adopted.stepped[0].retired_at,
Some(Timestamp::from_secs(AT_MS / 1000))
);
assert!(adoptions.cuts.is_empty());
});
}
crates/community/src/sync.rs
+20 -19
View File
@@ -17,7 +17,7 @@ use nostr_sdk::prelude::*;
use state::UniversalSigner;
use crate::cache::{self, Observed};
use crate::history::{CURSOR_OVERLAP_MS, Window};
use crate::history::{CURSOR_OVERLAP, Window};
/// How much of what a relay stores a cold subscription replays per relay.
const LIVE_REPLAY: usize = 500;
@@ -35,15 +35,15 @@ 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>,
/// When the rotation that retired this plane's key published.
pub retired_at: Option<Timestamp>,
}
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)
.is_none_or(|retired| wrap.created_at <= retired)
}
}
@@ -65,6 +65,7 @@ pub fn planes(state: &CommunityState) -> Result<Vec<Plane>> {
control_pk: None,
retired_at: None,
});
let group = control_group_key(&root.key, &state.id, epoch)?;
planes.push(Plane {
kind: PlaneKind::Control(epoch),
@@ -124,12 +125,12 @@ pub fn plane_filter(planes: &[Plane]) -> Filter {
pub fn live_filter(planes: &[Plane], window: Window) -> Filter {
let mut filter = plane_filter(planes);
if let Some(until_ms) = window.until_ms {
filter = filter.until(Timestamp::from_secs(until_ms / 1000));
if let Some(until) = window.until {
filter = filter.until(until);
}
if let Some(since_ms) = window.since_ms {
filter = filter.since(Timestamp::from_secs(since_ms / 1000));
if let Some(since) = window.since {
filter = filter.since(since);
}
filter.limit(LIVE_REPLAY)
@@ -140,13 +141,13 @@ pub fn live_window(state: &CommunityState) -> Window {
let floor = state
.channels
.iter()
.filter_map(|channel| state.cursors.get(&channel.id)?.newest_ms)
.filter_map(|channel| state.cursors.get(&channel.id)?.newest)
.min();
match floor {
Some(floor) => Window {
since_ms: Some(floor.saturating_sub(CURSOR_OVERLAP_MS)),
until_ms: None,
since: Some(floor - CURSOR_OVERLAP),
until: None,
},
None => Window::default(),
}
@@ -213,7 +214,7 @@ where
}
/// Best-effort publication of the genesis wraps to the community's relays.
async fn publish_wraps(client: &Client, wraps: &[Event], relays: &[RelayUrl]) {
pub(crate) async fn publish_wraps(client: &Client, wraps: &[Event], relays: &[RelayUrl]) {
connect_relays(client, relays).await;
for wrap in wraps {
@@ -428,7 +429,7 @@ fn refresh(mut held: CommunityState, fresh: CommunityState) -> CommunityState {
Some(held) => {
held.name = channel.name;
if cut.is_some_and(|cut| channel.epoch.0 <= cut) {
if cut.is_some_and(|cut| channel.epoch <= cut) {
continue;
}
@@ -767,16 +768,16 @@ mod tests {
state.cursors.insert(
channel,
concord::state::ChannelCursor {
newest_ms: Some(2_000_000),
oldest_ms: Some(1_000),
newest: Some(Timestamp::from_secs(2_000_000)),
oldest: Some(Timestamp::from_secs(1_000)),
exhausted: false,
},
);
assert_eq!(
live_window(&state),
Window {
since_ms: Some(2_000_000 - CURSOR_OVERLAP_MS),
until_ms: None,
since: Some(Timestamp::from_secs(2_000_000) - CURSOR_OVERLAP),
until: None,
}
);
}
@@ -1138,7 +1139,7 @@ mod tests {
priors: vec![HeldKey {
epoch: Epoch(0),
key: [0x07; 32],
retired_at: Some(1_000),
retired_at: Some(Timestamp::from_secs(1_000)),
}],
}];
@@ -1148,7 +1149,7 @@ mod tests {
.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_eq!(retired.retired_at, Some(Timestamp::from_secs(1_000)));
assert!(retired.accepts(&event_at(1_000_000)));
assert!(!retired.accepts(&event_at(1_001_000)));
}