From d2e96eb6f56c46da29bdac72cf6eb9e55c48ee0f Mon Sep 17 00:00:00 2001 From: dawn Date: Thu, 15 Oct 2026 21:48:34 +0300 Subject: [PATCH] knot2: keep private repos out of the firehose and public listings Signed-off-by: dawn --- knot2/crates/knot-acl/src/repos.rs | 6 +- knot2/crates/knot-index/src/lib.rs | 3 +- knot2/crates/knot-index/src/projections.rs | 27 +- knot2/crates/knot-index/tests/projections.rs | 8 +- knot2/crates/knot-types/src/ids.rs | 4 + knot2/crates/knot-xrpc/Cargo.toml | 1 + knot2/crates/knot-xrpc/src/atproto.rs | 17 +- knot2/crates/knot-xrpc/src/blobs.rs | 16 + knot2/crates/knot-xrpc/src/collaborators.rs | 6 +- knot2/crates/knot-xrpc/src/firehose.rs | 355 ++++++++++++++- knot2/crates/knot-xrpc/src/grants.rs | 5 +- knot2/crates/knot-xrpc/src/lib.rs | 2 +- knot2/crates/knot-xrpc/src/materialize.rs | 2 +- knot2/crates/knot-xrpc/src/migrate.rs | 5 +- knot2/crates/knot-xrpc/src/refrecords.rs | 12 +- knot2/crates/knot-xrpc/src/repos.rs | 64 +-- knot2/crates/knot-xrpc/src/reservations.rs | 13 +- knot2/crates/knot-xrpc/src/social.rs | 39 +- knot2/crates/knot-xrpc/src/tests.rs | 437 ++++++++++++++++--- knot2/crates/knot-xrpc/tests/common/mod.rs | 10 +- knot2/crates/knot-xrpc/tests/ref_records.rs | 11 +- 21 files changed, 877 insertions(+), 166 deletions(-) diff --git a/knot2/crates/knot-acl/src/repos.rs b/knot2/crates/knot-acl/src/repos.rs index 613dddb6f..6eab1193f 100644 --- a/knot2/crates/knot-acl/src/repos.rs +++ b/knot2/crates/knot-acl/src/repos.rs @@ -1,7 +1,7 @@ use std::path::{Path, PathBuf}; use knot_git::{GitError, Layout, Repo}; -use knot_index::{Index, Resolved, SocialIndex, SocialView}; +use knot_index::{Index, PublicRepo, Resolved, SocialIndex, SocialView}; use knot_types::{ AccountDid, ClonePath, ContributionPermission, KnotId, ObjectFormat, OwnerDid, RepoDid, Visibility, @@ -24,6 +24,10 @@ impl Repos { self.layout.open(grant.repo()) } + pub fn open_public(&self, repo: &PublicRepo) -> Result { + self.layout.open(repo.repo()) + } + pub fn path(&self, grant: &ReadGrant) -> Result { self.layout.guarded_path(grant.repo()) } diff --git a/knot2/crates/knot-index/src/lib.rs b/knot2/crates/knot-index/src/lib.rs index ce9ffc925..1b3c8e983 100644 --- a/knot2/crates/knot-index/src/lib.rs +++ b/knot2/crates/knot-index/src/lib.rs @@ -10,6 +10,7 @@ pub use coverage::{Coverage, Resolved}; pub use error::IndexError; pub use generation::{GrantGeneration, IndexGeneration}; pub use knot_types::OfferedKey; +pub use projections::PublicRepo; pub use social::{SocialEntry, SocialEntryState, SocialGuard, SocialResidencyBytes}; pub use social_index::{SocialIndex, SocialView}; @@ -672,7 +673,7 @@ impl Index { self.registry.visibility_of(&self.interner, repo) } - pub fn public_repos(&self) -> Vec { + pub fn public_repos(&self) -> Resolved> { self.registry.public_repos(&self.interner) } diff --git a/knot2/crates/knot-index/src/projections.rs b/knot2/crates/knot-index/src/projections.rs index 6798009ed..856170a31 100644 --- a/knot2/crates/knot-index/src/projections.rs +++ b/knot2/crates/knot-index/src/projections.rs @@ -21,6 +21,21 @@ use crate::{ GrantGeneration, KeyBudget, KeyLease, KeyRecord, KeyReprieve, KeyReprieved, SweepInterval, }; +#[derive(Debug, Clone, PartialEq, Eq)] +pub struct PublicRepo(RepoDid); + +impl PublicRepo { + pub fn repo(&self) -> &RepoDid { + &self.0 + } + + #[cfg(feature = "test-support")] + #[doc(hidden)] + pub fn assumed(repo: RepoDid) -> Self { + Self(repo) + } +} + #[derive(Debug, Clone, Copy)] pub(crate) struct Provenance { added_by: AccountKey, @@ -619,8 +634,16 @@ impl RegistryProjection { self.repos_where(interner, |_| true) } - pub(crate) fn public_repos(&self, interner: &Interner) -> Vec { - self.repos_where(interner, |slot| slot.visibility == Visibility::Public) + pub(crate) fn public_repos(&self, interner: &Interner) -> Resolved> { + if self.coverage.get() == Coverage::Warming { + return Resolved::Warming; + } + Resolved::Ready( + self.repos_where(interner, |slot| slot.visibility == Visibility::Public) + .into_iter() + .map(PublicRepo) + .collect(), + ) } fn repos_where(&self, interner: &Interner, keep: impl Fn(&RecordSlot) -> bool) -> Vec { diff --git a/knot2/crates/knot-index/tests/projections.rs b/knot2/crates/knot-index/tests/projections.rs index 22c8c49ee..8d6c36a3f 100644 --- a/knot2/crates/knot-index/tests/projections.rs +++ b/knot2/crates/knot-index/tests/projections.rs @@ -1267,7 +1267,13 @@ fn a_private_repo_is_hosted_but_not_public() { ); assert_eq!(index.visibility_of(&missing), Resolved::Ready(None)); assert_eq!(index.hosted_repos(), vec![open.clone(), hidden.clone()]); - assert_eq!(index.public_repos(), vec![open]); + assert_eq!( + index.public_repos().map(|repos| repos + .iter() + .map(|public| public.repo().clone()) + .collect::>()), + Resolved::Ready(vec![open]) + ); } #[test] diff --git a/knot2/crates/knot-types/src/ids.rs b/knot2/crates/knot-types/src/ids.rs index 318aef2bd..b124ac41e 100644 --- a/knot2/crates/knot-types/src/ids.rs +++ b/knot2/crates/knot-types/src/ids.rs @@ -852,6 +852,10 @@ impl UnixSeconds { pub const fn saturating_sub_secs(self, secs: i64) -> Self { Self(self.0.saturating_sub(secs)) } + + pub fn micros(self) -> UnixMicros { + UnixMicros((self.0.max(0) as u64).saturating_mul(1_000_000)) + } } #[derive(Debug, Clone, Copy, PartialEq, Eq, PartialOrd, Ord, Hash, Serialize, Deserialize)] diff --git a/knot2/crates/knot-xrpc/Cargo.toml b/knot2/crates/knot-xrpc/Cargo.toml index 128032fd2..5af1cf884 100644 --- a/knot2/crates/knot-xrpc/Cargo.toml +++ b/knot2/crates/knot-xrpc/Cargo.toml @@ -58,6 +58,7 @@ subtle = { workspace = true } zeroize = { workspace = true } url = { workspace = true } [dev-dependencies] +knot-index = { workspace = true, features = ["test-support"] } knot-fixtures = { workspace = true } knot-config = { workspace = true } k256 = { workspace = true } diff --git a/knot2/crates/knot-xrpc/src/atproto.rs b/knot2/crates/knot-xrpc/src/atproto.rs index 03ac4a536..0295b551d 100644 --- a/knot2/crates/knot-xrpc/src/atproto.rs +++ b/knot2/crates/knot-xrpc/src/atproto.rs @@ -537,16 +537,15 @@ pub(crate) async fn sync_list_repos( State(state): State>>, axum::extract::Query(params): axum::extract::Query, ) -> Result { - if matches!( - state.index.coverage().registry, - knot_index::Coverage::Warming - ) { - return Err(XrpcError::warming( - "registry projection is still warming, please retry shortly", - )); - } + let knot_index::Resolved::Ready(public) = state.index.public_repos() else { + return Err(crate::reads::warming()); + }; let dids: Vec = std::iter::once(RepoSubject::Did(state.knot_repo.clone())) - .chain(state.index.hosted_repos().into_iter().map(RepoSubject::Did)) + .chain( + public + .iter() + .map(|public| RepoSubject::Did(public.repo().clone())), + ) .collect(); let cursor = next_offset(params.cursor.get(), params.limit.get(), dids.len()); let page: Vec = dids diff --git a/knot2/crates/knot-xrpc/src/blobs.rs b/knot2/crates/knot-xrpc/src/blobs.rs index 642916b63..fa94e2066 100644 --- a/knot2/crates/knot-xrpc/src/blobs.rs +++ b/knot2/crates/knot-xrpc/src/blobs.rs @@ -515,6 +515,22 @@ impl BlobViews { } } } + + pub(crate) fn evict(&self, repo: &RepoDid) { + self.cached + .lock() + .expect("blob view cache lock poisoned") + .remove(repo); + } + + #[cfg(test)] + pub(crate) fn holds(&self, repo: &RepoDid) -> bool { + self.cached + .lock() + .expect("blob view cache lock poisoned") + .contains_key(repo) + } + pub fn live_roots(&self, git: &Repo, repo: &RepoDid) -> Result, RootsDenied> { let index = self .get(git, repo) diff --git a/knot2/crates/knot-xrpc/src/collaborators.rs b/knot2/crates/knot-xrpc/src/collaborators.rs index de47c513c..889c4f12c 100644 --- a/knot2/crates/knot-xrpc/src/collaborators.rs +++ b/knot2/crates/knot-xrpc/src/collaborators.rs @@ -32,13 +32,11 @@ fn require_owner( viewer: &Viewer, repo: &RepoDid, ) -> Result<(knot_types::OwnerDid, ReadGrant), XrpcError> { + let grant = crate::reads::require_hosted(state, repo.clone(), viewer)?; let owner = registered_owner(state, repo)?; let acl = KnotAcl::new(&state.admins, state.admission, &state.index); if can_manage_collaborators(&acl, actor, repo).is_allowed() { - Ok(( - owner, - crate::reads::require_hosted(state, repo.clone(), viewer)?, - )) + Ok((owner, grant)) } else { Err(XrpcError::forbidden( "only repository owner may manage collaborators", diff --git a/knot2/crates/knot-xrpc/src/firehose.rs b/knot2/crates/knot-xrpc/src/firehose.rs index 217d6f5e6..754c7dc84 100644 --- a/knot2/crates/knot-xrpc/src/firehose.rs +++ b/knot2/crates/knot-xrpc/src/firehose.rs @@ -16,10 +16,12 @@ use knot_events::{ BatchEnd, EventCursor, EventLog, ReplayBounds, ReplayBytes, ReplayEvents, Replayed, }; use knot_git::{GitError, RefUpdate, Repo}; +use knot_index::{PublicRepo, Resolved}; use knot_record::chain::{self, CommitChunk, Emit, MAX_COMMIT_BLOCKS_BYTES, Op, blocks_bytes}; use knot_runtime::{Clock, HttpTransport, Signer, UnixMicros}; use knot_types::{ AccountDid, Datetime, KnotId, Oid, PushOption, PushOptions, RefName, RepoDid, UnixSeconds, + Visibility, }; use serde::{Deserialize, Serialize}; @@ -38,6 +40,213 @@ const TRY_AGAIN_LATER: u16 = 1013; const MAX_PUSH_EDITS: usize = 200; const MAX_PUSH_OPTIONS: usize = 10; +// the only place that picks a feed by visibility so private writes can't end up in the public log +pub fn feed_of( + state: &XrpcState, + did: &RepoDid, +) -> Arc { + // we can't unsend a frame so a repo we don't know yet stays off the public feed + let visibility = match state.index.visibility_of(did) { + _ if *did == state.knot_repo => Visibility::Public, + Resolved::Ready(Some(visibility)) => visibility, + _ => Visibility::Private, + }; + feed_for(&state.firehose, visibility) +} + +pub(crate) fn feed_for(log: &Arc>, visibility: Visibility) -> Arc { + match visibility { + Visibility::Public => Arc::new(PublicFeed(Arc::clone(log))), + Visibility::Private => Arc::new(PrivateFeed), + } +} + +pub trait Feed: Send + Sync { + fn emit<'a>(&'a self, did: &'a RepoDid, now: UnixSeconds) -> Box; + + // identity and account frames have to go out before the first commit + fn reserve_birth(&self) -> Birth; + + fn publish_sync(&self, did: &RepoDid, rev: &Tid, commit_bytes: &[u8], now: UnixSeconds); + + fn publish_push( + &self, + did: &RepoDid, + pusher: &AccountDid, + applied: &[RefUpdate], + options: &PushOptions, + signer: &dyn Signer, + now: UnixSeconds, + ); + + fn publish_account( + &self, + did: &RepoDid, + active: bool, + status: Option<&'static str>, + now: UnixSeconds, + ); + + // counts as a publish because the tombstone goes on the firehose and answers getRepoStatus + fn record_deletion( + &self, + meta: &Repo, + did: &RepoDid, + deleted_at: UnixMicros, + ) -> Result<(), XrpcError>; +} + +pub trait ChainEmit: Emit { + fn finish(self: Box) -> Result<(), UnstampedFrames>; +} + +pub struct PublicFeed(Arc>); + +impl Feed for PublicFeed { + fn emit<'a>(&'a self, did: &'a RepoDid, now: UnixSeconds) -> Box { + Box::new(FirehoseEmit::new(&self.0, did, now)) + } + + fn reserve_birth(&self) -> Birth { + Birth(Some((self.0.reserve(), self.0.reserve()))) + } + + fn publish_sync(&self, did: &RepoDid, rev: &Tid, commit_bytes: &[u8], now: UnixSeconds) { + publish_sync(&self.0, did, rev, commit_bytes, now); + } + + fn publish_push( + &self, + did: &RepoDid, + pusher: &AccountDid, + applied: &[RefUpdate], + options: &PushOptions, + signer: &dyn Signer, + now: UnixSeconds, + ) { + publish_push(&self.0, did, pusher, applied, options, signer, now); + } + + fn publish_account( + &self, + did: &RepoDid, + active: bool, + status: Option<&'static str>, + now: UnixSeconds, + ) { + publish_account(&self.0, did, active, status, now); + } + + fn record_deletion( + &self, + meta: &Repo, + did: &RepoDid, + deleted_at: UnixMicros, + ) -> Result<(), XrpcError> { + crate::plc::record_deletion(meta, did, deleted_at) + } +} + +pub struct PrivateFeed; + +impl Feed for PrivateFeed { + fn emit<'a>(&'a self, did: &'a RepoDid, now: UnixSeconds) -> Box { + Box::new(PrivateEmit::new(did, now)) + } + + fn reserve_birth(&self) -> Birth { + Birth(None) + } + + fn publish_sync(&self, _did: &RepoDid, _rev: &Tid, _commit_bytes: &[u8], _now: UnixSeconds) {} + + fn publish_push( + &self, + _did: &RepoDid, + _pusher: &AccountDid, + _applied: &[RefUpdate], + _options: &PushOptions, + _signer: &dyn Signer, + _now: UnixSeconds, + ) { + } + + fn publish_account( + &self, + _did: &RepoDid, + _active: bool, + _status: Option<&'static str>, + _now: UnixSeconds, + ) { + } + + fn record_deletion( + &self, + _meta: &Repo, + _did: &RepoDid, + _deleted_at: UnixMicros, + ) -> Result<(), XrpcError> { + Ok(()) + } +} + +pub struct Birth(Option<(knot_events::Reservation, knot_events::Reservation)>); + +impl Birth { + pub fn announce(self, did: &RepoDid, now: UnixSeconds) { + if let Some((identity, account)) = self.0 { + fulfill_identity(identity, did, now); + fulfill_account(account, did, true, None, now); + } + } +} + +// still frames every commit so the chain on disk looks public, nothing gets sent +pub struct PrivateEmit<'a> { + did: &'a RepoDid, + now: UnixSeconds, + last: Option, +} + +impl<'a> PrivateEmit<'a> { + pub(crate) fn new(did: &'a RepoDid, now: UnixSeconds) -> Self { + Self { + did, + now, + last: None, + } + } +} + +impl Emit for PrivateEmit<'_> { + fn rev(&mut self, prev: Option<&Tid>) -> Tid { + let floor = prev + .map(|prev| UnixMicros::new(prev.timestamp())) + .max(self.last); + let wall = self.now.micros(); + let micros = floor.map_or(wall, |floor| wall.max(floor.next())); + self.last = Some(micros); + Tid::from_time(micros.get(), 0) + } + + fn frame(&self, chunk: &CommitChunk) -> Option { + commit_frame( + self.did, + chunk, + UnixMicros::new(chunk.rev.timestamp()), + self.now, + ) + } + + fn committed(&mut self, _frame: Option) {} +} + +impl ChainEmit for PrivateEmit<'_> { + fn finish(self: Box) -> Result<(), UnstampedFrames> { + Ok(()) + } +} + pub struct FirehoseEmit<'a, C: Clock> { log: &'a EventLog, did: &'a RepoDid, @@ -47,7 +256,7 @@ pub struct FirehoseEmit<'a, C: Clock> { } impl<'a, C: Clock> FirehoseEmit<'a, C> { - pub fn new(log: &'a EventLog, did: &'a RepoDid, now: UnixSeconds) -> Self { + pub(crate) fn new(log: &'a EventLog, did: &'a RepoDid, now: UnixSeconds) -> Self { Self { log, did, @@ -95,6 +304,12 @@ impl From for XrpcError { } } +impl ChainEmit for FirehoseEmit<'_, C> { + fn finish(self: Box) -> Result<(), UnstampedFrames> { + (*self).finish() + } +} + impl Emit for FirehoseEmit<'_, C> { fn rev(&mut self, prev: Option<&Tid>) -> Tid { if let Some(prev) = prev { @@ -119,7 +334,7 @@ impl Emit for FirehoseEmit<'_, C> { self.frames.push(frame); } } -pub fn fulfill_identity(reservation: knot_events::Reservation, did: &RepoDid, now: UnixSeconds) { +fn fulfill_identity(reservation: knot_events::Reservation, did: &RepoDid, now: UnixSeconds) { let seq = reservation.micros(); fulfill( reservation, @@ -132,7 +347,7 @@ pub fn fulfill_identity(reservation: knot_events::Reservation, did: &RepoDid, no ); } -pub fn fulfill_account( +fn fulfill_account( reservation: knot_events::Reservation, did: &RepoDid, active: bool, @@ -153,7 +368,7 @@ pub fn fulfill_account( ); } -pub fn publish_account( +fn publish_account( log: &EventLog, did: &RepoDid, active: bool, @@ -176,7 +391,7 @@ pub(crate) fn account_deleted_frame(did: &RepoDid, deleted_at: UnixMicros) -> By ) } -pub fn publish_sync( +fn publish_sync( log: &EventLog, did: &RepoDid, rev: &Tid, @@ -206,7 +421,7 @@ pub fn publish_sync( ); } -pub fn publish_push( +fn publish_push( log: &EventLog, did: &RepoDid, pusher: &AccountDid, @@ -565,10 +780,14 @@ pub(crate) async fn subscribe_repos( close_with(socket, rejection).await; }); } - let (knot, repos, hosted, log) = ( + let hosted = match (asked_seq, state.index.public_repos()) { + (None, _) => Vec::new(), + (Some(_), Resolved::Ready(public)) => public, + (Some(_), Resolved::Warming) => return crate::reads::warming().into_response(), + }; + let (knot, repos, log) = ( state.knot_did.clone(), state.repos.clone(), - state.index.hosted_repos(), Arc::clone(&state.firehose), ); upgrade.on_upgrade(move |socket| async move { @@ -611,7 +830,12 @@ pub(crate) async fn subscribe_repos( }) } -fn replay_index(knot: KnotId, repos: Repos, hosted: &[RepoDid], cursor: UnixMicros) -> ReplayFeed { +fn replay_index( + knot: KnotId, + repos: Repos, + hosted: &[PublicRepo], + cursor: UnixMicros, +) -> ReplayFeed { let opener = Opener { knot, repos }; let mut feed = ReplayFeed::empty(); match opener.open(&ReplaySource::Meta) { @@ -636,8 +860,8 @@ fn replay_index(knot: KnotId, repos: Repos, hosted: &[RepoDid], cursor: UnixMicr tracing::warn!(%error, "the meta repo wouldn't open, so its commits stay unreplayed") } } - for did in hosted { - let source = ReplaySource::Repo(ReadGrant::internal(did.clone())); + for public in hosted { + let source = ReplaySource::Repo(public.clone()); match opener.open(&source) { Ok(repo) => { let slot = feed.sources.len(); @@ -645,7 +869,7 @@ fn replay_index(knot: KnotId, repos: Repos, hosted: &[RepoDid], cursor: UnixMicr index_repo(&mut feed, slot, &repo, cursor, MAX_REPLAY_INDEX_FRAMES); } Err(error) => { - tracing::warn!(repo = did.as_str(), %error, "a repo wouldn't open, so its commits stay unreplayed") + tracing::warn!(repo = public.repo().as_str(), %error, "a repo wouldn't open, so its commits stay unreplayed") } } } @@ -705,7 +929,7 @@ impl IndexEntry { #[derive(Clone)] enum ReplaySource { Meta, - Repo(ReadGrant), + Repo(PublicRepo), } #[derive(Clone)] @@ -718,7 +942,7 @@ impl Opener { fn open(&self, source: &ReplaySource) -> Result { match source { ReplaySource::Meta => self.repos.open_meta(&self.knot), - ReplaySource::Repo(grant) => self.repos.open(grant), + ReplaySource::Repo(public) => self.repos.open_public(public), } } } @@ -1467,7 +1691,7 @@ mod tests { let replay = replay_index( knot_types::KnotId::new("did:web:knot.test").unwrap(), Repos::new(layout), - std::slice::from_ref(did), + std::slice::from_ref(&PublicRepo::assumed(did.clone())), UnixMicros::new(0), ); let mut pumped = Box::pin(pump( @@ -1517,7 +1741,7 @@ mod tests { knot: knot_types::KnotId::new("did:web:knot.test").unwrap(), repos: Repos::new(layout), }), - sources: vec![ReplaySource::Repo(ReadGrant::internal(rigged))], + sources: vec![ReplaySource::Repo(PublicRepo::assumed(rigged))], index: VecDeque::from([IndexEntry::Chain(FrameAt { micros: UnixMicros::new(tip.rev.timestamp()), repo: 0, @@ -1649,4 +1873,103 @@ mod tests { "a re-sequenced frame must still verify, or the event cannot cross a relay" ); } + + #[test] + fn a_private_repo_writes_its_chain_but_publishes_no_frame() { + let (_dir, _layout, _did, repo, signer) = rig(); + let untouched = log().last_issued(); + let log = Arc::new(log()); + let did = did(); + let feed = PrivateFeed; + let mut emit = feed.emit(&did, UnixSeconds::new(1_000)); + let chunks = futures::executor::block_on(chain::materialize( + &repo, + &did, + &chain::in_tree(records(2)), + &signer, + &mut *emit, + UnixSeconds::new(1), + )) + .unwrap(); + emit.finish().unwrap(); + assert!(!chunks.is_empty()); + let tip = chain::tip(&repo) + .unwrap() + .expect("the chain a later sync reads is still written"); + assert!( + chain::frame_blob(&repo, tip.git).unwrap().is_some(), + "the chain keeps its frames, they just never go out" + ); + assert!( + log.replay(EventCursor::START, drain_bounds()) + .events + .is_empty(), + "no frame of a private repo reaches the ring" + ); + assert_eq!( + log.last_issued(), + untouched, + "a private write doesn't take a sequence number" + ); + } + + #[test] + fn a_private_repo_keeps_its_revs_rising() { + let (_dir, _layout, _did, repo, signer) = rig(); + let did = did(); + let mut emit = PrivateEmit::new(&did, UnixSeconds::new(1_000)); + let first = futures::executor::block_on(chain::materialize( + &repo, + &did, + &chain::in_tree(records(1)), + &signer, + &mut emit, + UnixSeconds::new(1), + )) + .unwrap(); + // the clock goes backwards here but the rev still has to move forward + let mut emit = PrivateEmit::new(&did, UnixSeconds::new(1)); + let second = futures::executor::block_on(chain::sync( + &repo, + &did, + &[knot_types::RecordCollection::new(COLLECTION).unwrap()], + &chain::in_tree(records(3)), + &signer, + &mut emit, + UnixSeconds::new(2), + )) + .unwrap(); + let revs: Vec = first + .iter() + .chain(&second) + .map(|chunk| chunk.rev.timestamp()) + .collect(); + assert!(revs.len() > 1); + assert!(revs.windows(2).all(|pair| pair[0] < pair[1]), "{revs:?}"); + } + + #[test] + fn standalone_frames_of_a_private_repo_are_never_published() { + let log = Arc::new(log()); + let did = did(); + let feed = PrivateFeed; + let signer = knot_runtime::K256Signer::from_slice(&[7u8; 32]).unwrap(); + let now = UnixSeconds::new(1); + feed.reserve_birth().announce(&did, now); + feed.publish_account(&did, false, Some("deleted"), now); + feed.publish_sync(&did, &Tid::from_time(1, 0), &[], now); + feed.publish_push( + &did, + &AccountDid::new("did:plc:octopus").unwrap(), + &[], + &PushOptions::new([]), + &signer, + now, + ); + assert!( + log.replay(EventCursor::START, drain_bounds()) + .events + .is_empty() + ); + } } diff --git a/knot2/crates/knot-xrpc/src/grants.rs b/knot2/crates/knot-xrpc/src/grants.rs index 04421e337..05cfc34f7 100644 --- a/knot2/crates/knot-xrpc/src/grants.rs +++ b/knot2/crates/knot-xrpc/src/grants.rs @@ -46,14 +46,15 @@ fn sync_grants_for( .map(|(subject, entry)| (subject, entry.created_at)), collection, )); - let mut emit = crate::firehose::FirehoseEmit::new(&state.firehose, did, now); + let feed = crate::firehose::feed_of(state, did); + let mut emit = feed.emit(did, now); let chunks = futures::executor::block_on(chain::settle( git, did, std::slice::from_ref(&collection.records()), &records, &signer, - &mut emit, + &mut *emit, now, ))?; emit.finish()?; diff --git a/knot2/crates/knot-xrpc/src/lib.rs b/knot2/crates/knot-xrpc/src/lib.rs index 4ca43ee1a..058f914ea 100644 --- a/knot2/crates/knot-xrpc/src/lib.rs +++ b/knot2/crates/knot-xrpc/src/lib.rs @@ -36,7 +36,7 @@ mod tests; pub use blobs::{BlobViews, RepoAttachments}; pub use error::XrpcError; -pub use firehose::{FirehoseEmit, persist_seq, replay_floor as firehose_floor}; +pub use firehose::{Feed, feed_of, persist_seq, replay_floor as firehose_floor}; pub use invite_codes::InviteCodes; pub use knot_pack::MaxWireBytes; pub use knot_resource::{ diff --git a/knot2/crates/knot-xrpc/src/materialize.rs b/knot2/crates/knot-xrpc/src/materialize.rs index f141b987f..a5671c546 100644 --- a/knot2/crates/knot-xrpc/src/materialize.rs +++ b/knot2/crates/knot-xrpc/src/materialize.rs @@ -16,7 +16,7 @@ fn announce_unframed( ) { match chain::unframed_tip(git) { Ok(Some(tip)) => { - crate::firehose::publish_sync(&state.firehose, did, &tip.rev, &tip.commit, now) + crate::firehose::feed_of(state, did).publish_sync(did, &tip.rev, &tip.commit, now) } Ok(None) => {} Err(error) => tracing::warn!( diff --git a/knot2/crates/knot-xrpc/src/migrate.rs b/knot2/crates/knot-xrpc/src/migrate.rs index 6c538de0b..b8ac38174 100644 --- a/knot2/crates/knot-xrpc/src/migrate.rs +++ b/knot2/crates/knot-xrpc/src/migrate.rs @@ -64,14 +64,15 @@ fn sweep_legacy_records( .map(|nsid| knot_types::RecordCollection::new(*nsid).expect("literal NSID parses")) .collect(); let signer = state.secrets.signer(&state.knot_did)?; - let mut emit = crate::firehose::FirehoseEmit::new(&state.firehose, did, now); + let feed = crate::firehose::feed_of(state, did); + let mut emit = feed.emit(did, now); let chunks = futures::executor::block_on(chain::settle( git, did, &collections, &records, &signer, - &mut emit, + &mut *emit, now, ))?; emit.finish()?; diff --git a/knot2/crates/knot-xrpc/src/refrecords.rs b/knot2/crates/knot-xrpc/src/refrecords.rs index aaf04428b..795257697 100644 --- a/knot2/crates/knot-xrpc/src/refrecords.rs +++ b/knot2/crates/knot-xrpc/src/refrecords.rs @@ -42,13 +42,14 @@ pub fn sync_refs( return 0; } }; - let mut emit = firehose::FirehoseEmit::new(&state.firehose, did, now); + let feed = firehose::feed_of(state, did); + let mut emit = feed.emit(did, now); let chunks = match futures::executor::block_on(async { match (chain::tip(git), refs_self.is_empty()) { (Ok(None), true) => Ok(Vec::new()), (Ok(None), false) => { let seeded = chain::in_tree(refs_self); - chain::materialize(git, did, &seeded, &signer, &mut emit, now).await + chain::materialize(git, did, &seeded, &signer, &mut *emit, now).await } (Ok(Some(_)), _) => { chain::sync( @@ -57,7 +58,7 @@ pub fn sync_refs( std::slice::from_ref(&git_refs::git_refs_collection()), &chain::in_tree(refs_self), &signer, - &mut emit, + &mut *emit, now, ) .await @@ -83,7 +84,7 @@ pub fn sync_refs( let last = chunks .last() .expect("a non-empty chunk list has a last chunk"); - firehose::publish_sync(&state.firehose, did, &last.rev, &last.commit, now); + feed.publish_sync(did, &last.rev, &last.commit, now); } chunks.len() } @@ -165,8 +166,7 @@ impl knot_receive::RefSurface for RefProjection { - firehose::publish_push( - &self.state.firehose, + firehose::feed_of(&self.state, repo_did).publish_push( repo_did, pusher, applied, diff --git a/knot2/crates/knot-xrpc/src/repos.rs b/knot2/crates/knot-xrpc/src/repos.rs index 9b8552ef4..5850f0938 100644 --- a/knot2/crates/knot-xrpc/src/repos.rs +++ b/knot2/crates/knot-xrpc/src/repos.rs @@ -7,7 +7,7 @@ use axum::response::{IntoResponse, Response}; use http::HeaderMap; use serde::{Deserialize, Serialize}; -use knot_acl::{KnotAcl, Lfs, ReadGrant, Repos, Viewer, can_admin_knot, can_delete_repo}; +use knot_acl::{KnotAcl, Lfs, ReadGrant, Repos, can_admin_knot, can_delete_repo}; use knot_atproto::{PreparedRepoDid, RecordPresence}; use knot_cob::{CobHome, CobId, CobStore}; use knot_cobs::{ @@ -151,9 +151,8 @@ pub(crate) async fn create_repo( method: crate::Method, body: Bytes, ) -> Result { - let authed = state.authenticate_write(&headers, &method).await?; - let viewer = Viewer::from(authed.clone()); - let actor = authed.into_did(); + let (actor, viewer) = + crate::actor_and_viewer(state.authenticate_write(&headers, &method).await?); crate::authorize_create( &state, &actor, @@ -241,6 +240,11 @@ pub(crate) async fn create_repo( let manifest = ManifestRecord::new(owner.clone(), input.name.clone(), meta, now, actor.clone())?.body()?; + let visibility = if input.private { + Visibility::Private + } else { + Visibility::Public + }; let registration = Registration { owner, rkey, @@ -248,11 +252,7 @@ pub(crate) async fn create_repo( repo: repo_did.clone(), created_at: now, policy: state.contribution_policy, - visibility: if input.private { - Visibility::Private - } else { - Visibility::Public - }, + visibility, }; let lfs_store = state.lfs.as_ref().map(|web| web.store().clone()); @@ -386,20 +386,20 @@ pub(crate) async fn create_repo( let knot_signer = state.secrets.signer(&state.knot_did)?; let repos = state.repos.clone(); let cob_locks = Arc::clone(&state.cob_locks); - let firehose = Arc::clone(&state.firehose); + let feed = crate::firehose::feed_for(&state.firehose, visibility); + let born = Arc::clone(&feed); let did = repo_did.clone(); let initialized = created.clone(); let chunk = run_blocking(move || -> Result { - let identity = firehose.reserve(); - let account = firehose.reserve(); + let birth = born.reserve_birth(); let _guard = cob_locks.repo(initialized.repo()); let git = repos.open(&initialized)?; - let mut emit = crate::FirehoseEmit::new(&firehose, &did, now); + let mut emit = born.emit(&did, now); let tail = futures::executor::block_on(knot_record::chain::initialize( &git, &did, &knot_signer, - &mut emit, + &mut *emit, now, ))?; let tail = match knot_record::git_refs::desired_record(&git) { @@ -409,7 +409,7 @@ pub(crate) async fn create_repo( std::slice::from_ref(&knot_record::git_refs::git_refs_collection()), &knot_record::chain::in_tree(record), &knot_signer, - &mut emit, + &mut *emit, now, )) { Ok(chunks) => chunks.into_iter().last().unwrap_or(tail), @@ -424,12 +424,11 @@ pub(crate) async fn create_repo( } }; emit.finish()?; - crate::firehose::fulfill_identity(identity, &did, now); - crate::firehose::fulfill_account(account, &did, true, None, now); + birth.announce(&did, now); Ok(tail) }) .await?; - crate::firehose::publish_sync(&state.firehose, &repo_did, &chunk.rev, &chunk.commit, now); + feed.publish_sync(&repo_did, &chunk.rev, &chunk.commit, now); crate::social::open_manifest(&state, actor, &viewer, repo_did.clone(), manifest, now).await } .await; @@ -665,15 +664,21 @@ pub(crate) async fn delete_repo( method: crate::Method, body: Bytes, ) -> Result { - let actor = state - .authenticate_write(&headers, &method) - .await? - .into_did(); + let (actor, viewer) = + crate::actor_and_viewer(state.authenticate_write(&headers, &method).await?); let DeleteInput { - repo: repo_did, + repo: requested, force, } = decode(&body)?; + let grant = match knot_acl::read_grant(&state.index, requested, &viewer) { + Resolved::Ready(Some(grant)) => grant, + Resolved::Ready(None) => return Ok(ok_empty()), + Resolved::Warming => return Err(crate::reads::warming()), + }; + let repo_did = grant.into_repo(); + let feed = crate::firehose::feed_of(&state, &repo_did); + let RepoRef { owner: did, rkey } = match state.index.ownership_of(&repo_did) { Resolved::Ready(Some(found)) => found, Resolved::Ready(None) => return Ok(ok_empty()), @@ -726,13 +731,15 @@ pub(crate) async fn delete_repo( let target = RepoRef { owner: did, rkey }; let deleted = repo_did.clone(); let home = CobHome::from(&state.knot_did); + let blobs = Arc::clone(&state.blobs); let lfs_store = state.lfs.as_ref().map(|lfs| lfs.store().clone()); + let tombstone = Arc::clone(&feed); run_blocking(move || { let _repo_guard = cob_locks.repo(&deleted); let _meta_guard = cob_locks.meta(); let meta = repos.open_meta(&knot)?; let store = CobStore::new(&meta); - crate::plc::record_deletion(&meta, &deleted, now_micros)?; + tombstone.record_deletion(&meta, &deleted, now_micros)?; crate::plc::drop_pending_genesis(&meta, &deleted)?; deregister(&store, &home, target, &deleted, &knot_signer, now)?; let removal = repos.remove(&deleted); @@ -750,6 +757,7 @@ pub(crate) async fn delete_repo( "lfs prefix removal on delete failed, orphan sweep will reclaim" ); } + blobs.evict(&deleted); removal.map_err(|error| { XrpcError::internal(format!( "repo {deleted} was deregistered but its on-disk directory couldn't be removed: {error}" @@ -758,13 +766,7 @@ pub(crate) async fn delete_repo( }) .await?; - crate::firehose::publish_account( - &state.firehose, - &repo_did, - false, - Some("deleted"), - state.now(), - ); + feed.publish_account(&repo_did, false, Some("deleted"), state.now()); Ok(ok_empty()) } diff --git a/knot2/crates/knot-xrpc/src/reservations.rs b/knot2/crates/knot-xrpc/src/reservations.rs index d1711a1d8..10ff0935c 100644 --- a/knot2/crates/knot-xrpc/src/reservations.rs +++ b/knot2/crates/knot-xrpc/src/reservations.rs @@ -1,5 +1,4 @@ use knot_cache::{Admitted, Expiring, GroupQuota, Quotas, Rejected, TotalQuota}; -use knot_runtime::UnixMicros; use knot_types::{AccountDid, RepoDid, UnixSeconds}; #[derive(Debug, Clone, Copy, PartialEq, Eq)] @@ -22,10 +21,6 @@ pub struct Reservations { ttl: ReservationTtl, } -fn micros(seconds: UnixSeconds) -> UnixMicros { - UnixMicros::new((seconds.get().max(0) as u64).saturating_mul(1_000_000)) -} - impl Reservations { pub fn new(ttl: ReservationTtl, per_actor: PerActorQuota, global: GlobalQuota) -> Self { Self { @@ -38,7 +33,7 @@ impl Reservations { } pub(crate) fn prune(&self, now: UnixSeconds) -> Vec { - self.held.prune(micros(now)) + self.held.prune(now.micros()) } pub(crate) fn try_reserve( @@ -47,13 +42,13 @@ impl Reservations { actor: &AccountDid, now: UnixSeconds, ) -> ReserveDecision { - let expires_at = micros(now.saturating_add_secs(self.ttl.get())); + let expires_at = now.saturating_add_secs(self.ttl.get()).micros(); match self.held.admit_or_renew( repo.clone(), actor.clone(), actor.clone(), expires_at, - micros(now), + now.micros(), ) { Ok(Admitted::Inserted) => ReserveDecision::Fresh, Ok(Admitted::Occupied(holder)) if &holder == actor => ReserveDecision::Renewed, @@ -65,7 +60,7 @@ impl Reservations { pub(crate) fn holder_is(&self, repo: &RepoDid, actor: &AccountDid, now: UnixSeconds) -> bool { self.held - .get(repo, micros(now)) + .get(repo, now.micros()) .is_some_and(|holder| &holder == actor) } diff --git a/knot2/crates/knot-xrpc/src/social.rs b/knot2/crates/knot-xrpc/src/social.rs index 65d2ac0e3..a7e6a29ec 100644 --- a/knot2/crates/knot-xrpc/src/social.rs +++ b/knot2/crates/knot-xrpc/src/social.rs @@ -41,8 +41,8 @@ use knot_runtime::{Clock, HttpTransport, Signer}; use knot_types::{ AccountDid, AtUri, BranchName, ChangeId, CobId, ContributionPermission, ContributionPolicy, MstKey, Oid, OwnerDid, RecordAddress, RecordBody, RecordCid, RecordCollection, RecordRkey, - RefName, RepoDid, RepoName, RepoPath, RepoRkey, ServiceDid, SubjectNumber, Tid, TypeName, - UnixSeconds, + RefName, RepoDid, RepoName, RepoPath, RepoPolicy, RepoRkey, ServiceDid, SubjectNumber, Tid, + TypeName, UnixSeconds, Visibility, }; use serde::de::DeserializeOwned; use serde::{Deserialize, Serialize}; @@ -632,6 +632,11 @@ async fn permission( ) -> Result { let policy = ready(state.index.contribution_policy(repo), "registry")? .ok_or_else(|| XrpcError::not_found("No such repository on this knot"))?; + let private = ready(state.index.visibility_of(repo), "registry")? == Some(Visibility::Private); + let policy = match private { + true => RepoPolicy::new(repo.clone(), ContributionPolicy::Collaborators), + false => policy, + }; project_collaborators(state, repo).await?; let collaboration = state.index.collaboration_consent(repo, actor); let membership = state.index.member_status(actor); @@ -726,14 +731,14 @@ impl Ledger<'_, H, C> { } fn publish(&self, address: &RecordAddress, op: RecordOp) -> Result { - let mut emit = - crate::firehose::FirehoseEmit::new(&self.state.firehose, self.repo(), self.now); + let feed = crate::firehose::feed_of(self.state, self.repo()); + let mut emit = feed.emit(self.repo(), self.now); let applied = futures::executor::block_on(chain::apply( self.git, self.repo(), &[(address.clone(), op)], self.signer, - &mut emit, + &mut *emit, self.now, )); match applied { @@ -986,14 +991,15 @@ pub(crate) fn reconcile( if chain::tip(git)?.is_none() && chain::Records::due(&records).is_empty() { return Ok(Reconciled { commits: 0 }); } - let mut emit = crate::firehose::FirehoseEmit::new(&state.firehose, repo, now); + let feed = crate::firehose::feed_of(state, repo); + let mut emit = feed.emit(repo, now); let chunks = futures::executor::block_on(chain::settle( git, repo, &knot_cobs::social_collections(), &records, signer, - &mut emit, + &mut *emit, now, ))?; emit.finish()?; @@ -1062,12 +1068,12 @@ pub(crate) async fn session( let authed = state .authenticate_write_for_repo(headers, method, &repo) .await?; - let viewer = Viewer::from(authed.clone()); - let actor = authed.into_did(); + let (actor, viewer) = crate::actor_and_viewer(authed); + let grant = crate::reads::require_hosted(state, repo.clone(), &viewer)?; let permission = permission(state, &actor, &repo).await?; Ok(Session { actor, - grant: crate::reads::require_hosted(state, repo, &viewer)?, + grant, permission, now: state.now(), }) @@ -1182,10 +1188,11 @@ pub(crate) async fn open_manifest( record: RecordBody, now: UnixSeconds, ) -> Result<(), XrpcError> { + let grant = crate::reads::require_hosted(state, repo.clone(), viewer)?; let permission = permission(state, &actor, &repo).await?; let session = Session { actor, - grant: crate::reads::require_hosted(state, repo, viewer)?, + grant, permission, now, }; @@ -1982,9 +1989,8 @@ async fn grant_open( let authed = state .authenticate_write_for_repo(headers, method, &repo) .await?; - let viewer = Viewer::from(authed.clone()); - crate::collaborators::collaborator_add(state, authed.did(), &viewer, &repo, subject) - .await + let (actor, viewer) = crate::actor_and_viewer(authed); + crate::collaborators::collaborator_add(state, &actor, &viewer, &repo, subject).await } }?; Ok(crate::ok_empty()) @@ -2020,9 +2026,8 @@ async fn grant_erase( let authed = state .authenticate_write_for_repo(headers, method, &repo) .await?; - let viewer = Viewer::from(authed.clone()); - crate::collaborators::collaborator_remove(state, authed.did(), &viewer, &repo, subject) - .await + let (actor, viewer) = crate::actor_and_viewer(authed); + crate::collaborators::collaborator_remove(state, &actor, &viewer, &repo, subject).await } }?; Ok(crate::ok_empty()) diff --git a/knot2/crates/knot-xrpc/src/tests.rs b/knot2/crates/knot-xrpc/src/tests.rs index 06dd3444a..56c85313a 100644 --- a/knot2/crates/knot-xrpc/src/tests.rs +++ b/knot2/crates/knot-xrpc/src/tests.rs @@ -192,22 +192,6 @@ async fn json_of(response: Response) -> serde_json::Value { serde_json::from_slice(&bytes).unwrap() } -async fn repo_status(world: &World, did: &RepoDid) -> serde_json::Value { - json_of( - crate::atproto::sync_get_repo_status( - State(Arc::clone(&world.state)), - HeaderMap::new(), - crate::Method::from_nsid("com.atproto.sync.getRepoStatus"), - axum::extract::Query(crate::atproto::DidParams { - did: crate::atproto::RepoSubject::Did(did.clone()), - }), - ) - .await - .unwrap(), - ) - .await -} - async fn repo_status_as(world: &World, did: &RepoDid, headers: HeaderMap) -> Response { into_response( crate::atproto::sync_get_repo_status( @@ -222,6 +206,12 @@ async fn repo_status_as(world: &World, did: &RepoDid, headers: HeaderMap) -> Res ) } +async fn repo_status(world: &World, did: &RepoDid) -> serde_json::Value { + let response = repo_status_as(world, did, HeaderMap::new()).await; + assert_eq!(response.status(), StatusCode::OK); + json_of(response).await +} + async fn call( world: &World, handler: F, @@ -1151,28 +1141,44 @@ fn manifest_rkey() -> RecordRkey { RecordRkey::new("self").unwrap() } -async fn live_manifest( +async fn manifest_with( world: &World, + headers: HeaderMap, repo: &RepoDid, ) -> (serde_json::Value, crate::out::CidString) { - let (status, served) = record_at( - world, - repo, - &at_of(repo, manifest_collection(), &manifest_rkey()), - None, - ) - .await; + let uri = at_of(repo, manifest_collection(), &manifest_rkey()); + let (status, served) = record_with(world, headers, repo, &uri, None).await; assert_eq!(status, StatusCode::OK, "the manifest serves: {served}"); (served["value"].clone(), cid_of(&served)) } +async fn live_manifest( + world: &World, + repo: &RepoDid, +) -> (serde_json::Value, crate::out::CidString) { + manifest_with(world, HeaderMap::new(), repo).await +} + +async fn manifest_as( + world: &World, + actor: &Actor, + repo: &RepoDid, +) -> (serde_json::Value, crate::out::CidString) { + let token = mint_for( + actor, + &nsid("com.atproto.repo.getRecord"), + &ServiceDid::new(repo.as_str()).unwrap(), + ); + manifest_with(world, bearer(&token), repo).await +} + async fn put_manifest_as( world: &World, actor: &Actor, repo: &RepoDid, edit: impl FnOnce(&mut serde_json::Value), ) -> (StatusCode, serde_json::Value) { - let (mut record, from) = live_manifest(world, repo).await; + let (mut record, from) = manifest_as(world, actor, repo).await; edit(&mut record); put_as( world, @@ -1631,11 +1637,21 @@ async fn record_at( repo: &RepoDid, uri: &AtUri, cid: Option<&crate::out::CidString>, +) -> (StatusCode, serde_json::Value) { + record_with(world, HeaderMap::new(), repo, uri, cid).await +} + +async fn record_with( + world: &World, + headers: HeaderMap, + repo: &RepoDid, + uri: &AtUri, + cid: Option<&crate::out::CidString>, ) -> (StatusCode, serde_json::Value) { let response = into_response( crate::atproto::repo_get_record( world.state(), - HeaderMap::new(), + headers, crate::Method::from_nsid("com.atproto.repo.getRecord"), crate::query::ValidatedQuery(record_of(repo, uri, cid.cloned())), ) @@ -3203,26 +3219,28 @@ async fn create_mints_a_did_plc_repo_and_refuses_a_duplicate_name() { ); } +async fn create_private_repo_helper(world: &World, name: &str) -> RepoDid { + assert_eq!( + create_status( + world, + &world.member, + json!({ "name": name, "private": true }) + ) + .await, + StatusCode::OK + ); + match resolve(world, &RepoName::new(name).unwrap()) { + Resolved::Ready(Some(did)) => did, + other => panic!("Private repo {name} wasn't registered: {other:?}"), + } +} + #[tokio::test] async fn create_marks_a_repo_private_only_when_asked() { let world = World::new(); world.grandfather(MEMBER_HOST); let public = create_repo_helper(&world, "anemone").await; - assert_eq!( - as_member( - &world, - crate::repos::create_repo, - &nsid(CREATE), - json!({ "name": "barnacle", "private": true }) - ) - .await - .status(), - StatusCode::OK - ); - let Resolved::Ready(Some(private)) = resolve(&world, &RepoName::new("barnacle").unwrap()) - else { - panic!("Private repo wasn't registered") - }; + let private = create_private_repo_helper(&world, "barnacle").await; assert_eq!( world.state.index.visibility_of(&public), Resolved::Ready(Some(Visibility::Public)) @@ -3231,7 +3249,327 @@ async fn create_marks_a_repo_private_only_when_asked() { world.state.index.visibility_of(&private), Resolved::Ready(Some(Visibility::Private)) ); - assert_eq!(world.state.index.public_repos(), vec![public]); + assert_eq!( + world.state.index.public_repos().map(|repos| repos + .iter() + .map(|public| public.repo().clone()) + .collect::>()), + Resolved::Ready(vec![public]) + ); +} + +async fn plainly(response: Response) -> (StatusCode, Vec) { + let status = response.status(); + let bytes = axum::body::to_bytes(response.into_body(), usize::MAX) + .await + .unwrap(); + (status, bytes.to_vec()) +} + +fn ghost() -> RepoDid { + RepoDid::new("did:plc:aaaaaaaaaaaaaaaaaaaaaaaa").unwrap() +} + +fn probed( + what: &'static str, + (status, body): (StatusCode, serde_json::Value), +) -> (&'static str, StatusCode, Vec) { + (what, status, serde_json::to_vec(&body).unwrap()) +} + +async fn write_probes( + world: &World, + actor: &Actor, + repo: &RepoDid, +) -> Vec<(&'static str, StatusCode, Vec)> { + let ticket = Ticket::on(world, repo) + .open(actor, "kelp", "a forest") + .await; + let grant = on_repo( + world, + crate::social::create_record, + actor, + &nsid(CREATE_RECORD), + repo, + grant_body(repo, &collaborator_collection(), &world.passerby.did), + ) + .await; + let erased = on_repo( + world, + crate::social::delete_record, + actor, + &nsid(DELETE_RECORD), + repo, + json!({ "repo": repo, "collection": COLLABORATOR, "rkey": world.passerby.did.as_str() }), + ) + .await; + let (status, body) = plainly( + call( + world, + crate::repos::delete_repo, + actor, + &nsid(DELETE), + json!({ "repo": repo }), + ) + .await, + ) + .await; + vec![ + probed("ticket", ticket), + probed("collaborator add", grant), + probed("collaborator remove", erased), + ("delete repo", status, body), + ] +} + +#[tokio::test] +async fn writes_to_a_private_repo_answer_a_stranger_like_a_missing_repo() { + let world = World::new(); + world.grandfather(MEMBER_HOST); + let private = create_private_repo_helper(&world, "barnacle").await; + for actor in [&world.stranger, &world.admin, &world.passerby] { + let missing = write_probes(&world, actor, &ghost()).await; + let hidden = write_probes(&world, actor, &private).await; + assert_eq!( + hidden, missing, + "{} can tell a private repo from a missing one", + actor.did + ); + } + assert!( + world.layout.open(&private).is_ok(), + "a stranger's delete left the private repo alone" + ); +} + +#[tokio::test] +async fn an_anonymous_write_to_a_private_repo_answers_like_one_to_a_missing_repo() { + let world = World::new(); + world.grandfather(MEMBER_HOST); + let private = create_private_repo_helper(&world, "barnacle").await; + let anonymous = |repo: RepoDid| { + let world = &world; + async move { + plainly(into_response( + crate::social::create_record( + world.state(), + HeaderMap::new(), + crate::Method::from_nsid(CREATE_RECORD), + body(json!({ + "repo": repo, + "collection": TICKET, + "record": ticket_record(&repo, "kelp", "a forest"), + })), + ) + .await, + )) + .await + } + }; + assert_eq!(anonymous(private).await, anonymous(ghost()).await); +} + +#[tokio::test] +async fn only_the_owner_and_collaborators_contribute_to_a_private_repo() { + let world = World::new(); + world.grandfather(MEMBER_HOST); + let private = create_private_repo_helper(&world, "barnacle").await; + let ticket = Ticket::on(&world, &private); + assert_eq!( + set_policy_as(&world, &world.member, &private, ContributionPolicy::Anyone).await, + StatusCode::OK + ); + assert_eq!( + ticket.open(&world.stranger, "kelp", "a forest").await.0, + StatusCode::NOT_FOUND, + "an open contribution policy doesn't let outsiders in" + ); + ok(ticket.open(&world.member, "kelp", "a forest").await); + assert_eq!( + offer_collaborator_as(&world, &world.member, &private, &world.passerby.did).await, + StatusCode::OK + ); + ok(ticket.open(&world.passerby, "reef", "a shelf").await); +} + +#[tokio::test] +async fn a_public_repo_still_answers_a_stranger_who_cannot_write_to_it() { + let world = World::new(); + world.grandfather(MEMBER_HOST); + let public = create_repo_helper(&world, "anemone").await; + let (status, _) = write_probes(&world, &world.passerby, &public) + .await + .into_iter() + .find_map(|(what, status, body)| (what == "delete repo").then_some((status, body))) + .unwrap(); + assert_eq!(status, StatusCode::FORBIDDEN); +} + +#[tokio::test] +async fn listing_repos_skips_private_ones_without_moving_the_pages() { + let world = World::new(); + world.grandfather(MEMBER_HOST); + let mut public = Vec::new(); + for name in ["anemone", "kelp", "urchin"] { + public.push(create_repo_helper(&world, name).await); + create_private_repo_helper(&world, &format!("{name}-hidden")).await; + } + let mut listed = Vec::new(); + let mut pages = 0; + let mut cursor: Option = None; + loop { + let query = match &cursor { + Some(cursor) => format!("limit=2&cursor={cursor}"), + None => "limit=2".to_owned(), + }; + let uri: http::Uri = format!("/xrpc/com.atproto.sync.listRepos?{query}") + .parse() + .unwrap(); + let response = crate::atproto::sync_list_repos( + world.state(), + axum::extract::Query::try_from_uri(&uri).unwrap(), + ) + .await + .unwrap(); + let page = json_of(response).await; + let repos = page["repos"].as_array().unwrap(); + pages += 1; + listed.extend( + repos + .iter() + .map(|repo| repo["did"].as_str().unwrap().to_owned()), + ); + match page["cursor"].as_str() { + Some(next) => cursor = Some(next.to_owned()), + None => break, + } + } + assert_eq!( + pages, 2, + "the knot's own repo and three public ones take two pages, the private ones take no slot" + ); + let mut expected: Vec = public.iter().map(|did| did.as_str().to_owned()).collect(); + listed.retain(|did| did != world.state.knot_repo.as_str()); + listed.sort(); + expected.sort(); + assert_eq!(listed, expected); +} + +fn firehose_events(world: &World) -> Vec> { + world + .state + .firehose + .replay( + knot_events::EventCursor::START, + knot_events::ReplayBounds::new( + knot_events::ReplayEvents::new(256).unwrap(), + knot_events::ReplayBytes::new(16 << 20).unwrap(), + ), + ) + .events +} + +fn names_repo(events: &[Arc], repo: &RepoDid) -> bool { + let needle = repo.as_str().as_bytes(); + events.iter().any(|event| { + event + .frame + .windows(needle.len()) + .any(|window| window == needle) + }) +} + +#[tokio::test] +async fn the_firehose_carries_a_public_repos_frames_and_none_of_a_private_ones() { + let world = World::new(); + world.grandfather(MEMBER_HOST); + let public = create_repo_helper(&world, "anemone").await; + let private = create_private_repo_helper(&world, "barnacle").await; + ok(Ticket::on(&world, &private) + .open(&world.member, "kelp", "a forest") + .await); + ok(Ticket::on(&world, &public) + .open(&world.member, "reef", "a shelf") + .await); + assert_eq!( + offer_collaborator_as(&world, &world.member, &private, &world.passerby.did).await, + StatusCode::OK + ); + + let events = firehose_events(&world); + assert!(names_repo(&events, &public), "public frames still flow"); + assert!( + !names_repo(&events, &private), + "a private repo's did is on no frame" + ); + let grant = knot_acl::ReadGrant::internal(private.clone()); + let git = world.state.repos.open(&grant).unwrap(); + assert!( + knot_record::chain::tip(&git).unwrap().is_some(), + "the private chain is still written for later reads" + ); + assert!( + events + .windows(2) + .all(|pair| pair[0].created < pair[1].created), + "seqs only climb" + ); +} + +#[tokio::test] +async fn deleting_a_private_repo_leaves_no_tombstone_or_frame() { + let world = World::new(); + world.grandfather(MEMBER_HOST); + let public = create_repo_helper(&world, "anemone").await; + let private = create_private_repo_helper(&world, "barnacle").await; + for repo in [&public, &private] { + assert_eq!( + as_member( + &world, + crate::repos::delete_repo, + &nsid(DELETE), + json!({ "repo": repo }) + ) + .await + .status(), + StatusCode::OK + ); + } + assert!(crate::plc::is_deleted(&world.state, &public).await.unwrap()); + assert!( + !crate::plc::is_deleted(&world.state, &private) + .await + .unwrap() + ); + let events = firehose_events(&world); + assert!(names_repo(&events, &public)); + assert!(!names_repo(&events, &private)); +} + +#[tokio::test] +async fn deleting_a_repo_drops_its_blob_view() { + let world = World::new(); + world.grandfather(MEMBER_HOST); + let repo = create_repo_helper(&world, "anemone").await; + let git = world + .state + .repos + .open(&knot_acl::ReadGrant::internal(repo.clone())) + .unwrap(); + world.state.blobs.get(&git, &repo).unwrap(); + assert!(world.state.blobs.holds(&repo)); + assert_eq!( + as_member( + &world, + crate::repos::delete_repo, + &nsid(DELETE), + json!({ "repo": repo }) + ) + .await + .status(), + StatusCode::OK + ); + assert!(!world.state.blobs.holds(&repo)); } #[tokio::test] @@ -5244,20 +5582,7 @@ mod fork_endpoints { async fn forking_a_private_source_is_not_found_like_a_missing_one() { let world = World::new(); world.grandfather(MEMBER_HOST); - assert_eq!( - as_member( - &world, - crate::repos::create_repo, - &nsid(CREATE), - json!({ "name": "kelp", "private": true }) - ) - .await - .status(), - StatusCode::OK - ); - let Resolved::Ready(Some(source)) = resolve(&world, &RepoName::new("kelp").unwrap()) else { - panic!("Private repo wasn't registered") - }; + let source = create_private_repo_helper(&world, "kelp").await; let source = Did::new_owned(source.as_str()).unwrap(); world.publish_plc_doc( &source, diff --git a/knot2/crates/knot-xrpc/tests/common/mod.rs b/knot2/crates/knot-xrpc/tests/common/mod.rs index 79d13ffff..4cc6cb992 100644 --- a/knot2/crates/knot-xrpc/tests/common/mod.rs +++ b/knot2/crates/knot-xrpc/tests/common/mod.rs @@ -855,7 +855,8 @@ impl World { .spawn(move || { let git = Repo::open(&path).unwrap(); let signer = self.state.secrets.signer(&self.state.knot_did).unwrap(); - let mut emit = knot_xrpc::FirehoseEmit::new(&self.state.firehose, did, now); + let feed = knot_xrpc::feed_of(&self.state, did); + let mut emit = feed.emit(did, now); let runtime = tokio::runtime::Builder::new_current_thread() .enable_all() .build() @@ -867,7 +868,7 @@ impl World { std::slice::from_ref(&fixture_collection()), &knot_record::chain::in_tree(records), &signer, - &mut emit, + &mut *emit, now, )) .unwrap(); @@ -1013,8 +1014,9 @@ pub fn enable_chain(world: &World, did: &RepoDid) { let now = UnixSeconds::new(1_000); let git = Repo::open(world.layout.repo_path(did).unwrap()).unwrap(); let signer = world.state.secrets.signer(&world.state.knot_did).unwrap(); - let mut emit = knot_xrpc::FirehoseEmit::new(&world.state.firehose, did, now); - let chain = knot_record::chain::initialize(&git, did, &signer, &mut emit, now); + let feed = knot_xrpc::feed_of(&world.state, did); + let mut emit = feed.emit(did, now); + let chain = knot_record::chain::initialize(&git, did, &signer, &mut *emit, now); futures::executor::block_on(chain).unwrap(); emit.finish().unwrap(); } diff --git a/knot2/crates/knot-xrpc/tests/ref_records.rs b/knot2/crates/knot-xrpc/tests/ref_records.rs index fe0998476..7e4dc430a 100644 --- a/knot2/crates/knot-xrpc/tests/ref_records.rs +++ b/knot2/crates/knot-xrpc/tests/ref_records.rs @@ -165,7 +165,11 @@ async fn a_push_stores_the_refs_snapshot_the_knot_serves_over_com_atproto() { http_push(work.path(), addr, &did, &owner(), &[":feature"]).await; let lines = packed_lines(&snapshot(&world, &did).await); - assert_eq!(lines.len(), 2, "the snapshot no longer lists the deleted branch"); + assert_eq!( + lines.len(), + 2, + "the snapshot no longer lists the deleted branch" + ); assert!( lines .iter() @@ -287,9 +291,10 @@ async fn the_legacy_sweep_deletes_old_ref_records_and_survives_its_chunking() { let git = layout.open(&did).unwrap(); let signer = state.secrets.signer(&state.knot_did).unwrap(); let now = UnixSeconds::new(2_000); - let mut emit = knot_xrpc::FirehoseEmit::new(&state.firehose, &did, now); + let feed = knot_xrpc::feed_of(&state, &did); + let mut emit = feed.emit(&did, now); let chunks = futures::executor::block_on(chain::apply( - &git, &did, &ops, &signer, &mut emit, now, + &git, &did, &ops, &signer, &mut *emit, now, )) .unwrap(); emit.finish().unwrap(); -- 2.54.0