From ca0b749199c39c8c3ea09a502dfa41506d81d375 Mon Sep 17 00:00:00 2001 From: Lewis Date: Fri, 23 Oct 2026 09:59:11 +0300 Subject: [PATCH] knot2/foster: post-ack foster forgetting repos Lewis: May this revision serve well! --- .slop-allow | 1 - crates/lexicons/Cargo.toml | 1 + knot2/crates/knot-acl/src/lib.rs | 2 + knot2/crates/knot-atproto/src/lib.rs | 12 + knot2/crates/knot-atproto/src/resolve.rs | 34 ++ knot2/crates/knot-bench/src/fixtures.rs | 2 + knot2/crates/knot-cobs/src/comment.rs | 30 +- knot2/crates/knot-cobs/src/label.rs | 6 + knot2/crates/knot-cobs/src/lib.rs | 11 +- knot2/crates/knot-cobs/src/manifest.rs | 6 + knot2/crates/knot-cobs/src/pull.rs | 6 + knot2/crates/knot-cobs/src/reaction.rs | 19 +- knot2/crates/knot-cobs/src/registry.rs | 160 +++++++ knot2/crates/knot-cobs/src/social.rs | 85 +++- knot2/crates/knot-cobs/src/ticket.rs | 43 +- knot2/crates/knot-cobs/tests/common/mod.rs | 1 + knot2/crates/knot-cobs/tests/social.rs | 7 + knot2/crates/knot-index/src/projections.rs | 1 + knot2/crates/knot-index/tests/common/mod.rs | 9 + knot2/crates/knot-index/tests/meta_repo.rs | 2 + knot2/crates/knot-index/tests/properties.rs | 1 + knot2/crates/knot-keyfill/tests/fill.rs | 1 + knot2/crates/knot-migrate/src/emit.rs | 2 + knot2/crates/knot-server/src/main.rs | 4 + knot2/crates/knot-sim/src/harness.rs | 9 +- knot2/crates/knot-sim/tests/lfs_chaos.rs | 1 + knot2/crates/knot-sim/tests/lfs_roundtrip.rs | 5 +- knot2/crates/knot-sim/tests/social_chaos.rs | 1 + knot2/crates/knot-sim/tests/ssh.rs | 1 + .../knot-ssh/examples/ephemeral_knot.rs | 2 + knot2/crates/knot-ssh/tests/ssh_push.rs | 5 + knot2/crates/knot-types/src/ids.rs | 20 +- knot2/crates/knot-xrpc/src/error.rs | 9 + knot2/crates/knot-xrpc/src/foster.rs | 401 +++++++++++++++++- knot2/crates/knot-xrpc/src/handover.rs | 190 +++++---- knot2/crates/knot-xrpc/src/repos.rs | 1 + knot2/crates/knot-xrpc/src/social.rs | 89 +++- .../crates/knot-xrpc/tests/atproto_interop.rs | 20 +- knot2/crates/knot-xrpc/tests/common/mod.rs | 87 ++-- knot2/crates/knot-xrpc/tests/handover.rs | 182 ++++++-- knot2/crates/knot-xrpc/tests/projection.rs | 6 +- knot2/crates/knot-xrpc/tests/retire.rs | 242 +++++++++++ knot2/interop/foster_test.go | 9 +- web/src/lib/api/lexicons/index.ts | 1 + .../types/com/bad-example/pocket/defs.ts | 2 +- .../types/org/tangled/temp/server/seedRepo.ts | 21 +- .../org/tangled/temp/server/uploadBlob.ts | 40 ++ 47 files changed, 1593 insertions(+), 197 deletions(-) delete mode 100644 .slop-allow create mode 100644 knot2/crates/knot-xrpc/tests/retire.rs create mode 100644 web/src/lib/api/lexicons/types/org/tangled/temp/server/uploadBlob.ts diff --git a/.slop-allow b/.slop-allow deleted file mode 100644 index 4e161fbc8..000000000 --- a/.slop-allow +++ /dev/null @@ -1 +0,0 @@ -named diff --git a/crates/lexicons/Cargo.toml b/crates/lexicons/Cargo.toml index 7246fcdb9..203cfff6e 100644 --- a/crates/lexicons/Cargo.toml +++ b/crates/lexicons/Cargo.toml @@ -33,3 +33,4 @@ streaming = ["jacquard-common/websocket"] [lints.clippy] unnecessary_lazy_evaluations = "allow" +unit_arg = "allow" diff --git a/knot2/crates/knot-acl/src/lib.rs b/knot2/crates/knot-acl/src/lib.rs index 5d04f6680..f21d07280 100644 --- a/knot2/crates/knot-acl/src/lib.rs +++ b/knot2/crates/knot-acl/src/lib.rs @@ -728,6 +728,8 @@ mod tests { created_at: UnixSeconds::new(at), policy: ContributionPolicy::default(), fostered: false, + + retired_at: None, } } diff --git a/knot2/crates/knot-atproto/src/lib.rs b/knot2/crates/knot-atproto/src/lib.rs index f0929fc2d..6187d15eb 100644 --- a/knot2/crates/knot-atproto/src/lib.rs +++ b/knot2/crates/knot-atproto/src/lib.rs @@ -278,6 +278,18 @@ impl Atproto { Ok(self.identity(did).await?.value) } + pub async fn resolve_knot_endpoint(&self, did: &RepoDid) -> Result { + let url = resolve::document_url(&AccountDid::from(did.clone()), &self.plc_directory)?; + resolve::guard_fetch_url(&url)?; + let response = self.http.execute(HttpRequest::get(url)).await?; + if !response.status.is_success() { + return Err(AtprotoError::Resolve(ResolveError::Status { + status: HttpStatus::from(response.status), + })); + } + resolve::knot_endpoint(did, &response.body).map_err(AtprotoError::Resolve) + } + async fn identity(&self, did: &AccountDid) -> Result, ResolveError> { let now = self.clock.now_unix_micros(); let filled = self diff --git a/knot2/crates/knot-atproto/src/resolve.rs b/knot2/crates/knot-atproto/src/resolve.rs index e9c2c398b..00d276cee 100644 --- a/knot2/crates/knot-atproto/src/resolve.rs +++ b/knot2/crates/knot-atproto/src/resolve.rs @@ -10,6 +10,8 @@ const LEGACY_K256_KIND: &str = "EcdsaSecp256k1VerificationKey2019"; const LEGACY_P256_KIND: &str = "EcdsaSecp256r1VerificationKey2019"; const SPINDLE_SERVICE_ID: &str = "#tangled_spindle"; const SPINDLE_SERVICE_TYPE: &str = "TangledSpindle"; +const KNOT_SERVICE_ID: &str = "#tangled_knot"; +const KNOT_SERVICE_TYPE: &str = "TangledKnot"; #[derive(Debug, Clone, thiserror::Error)] pub enum ResolveError { @@ -41,6 +43,10 @@ pub enum ResolveError { MissingSpindle, #[error("tangled_spindle endpoint {value:?} isn't valid URL")] BadSpindle { value: String }, + #[error("DID document declares no tangled_knot service endpoint")] + MissingKnot, + #[error("tangled_knot endpoint {value:?} isn't valid URL")] + BadKnot { value: String }, #[error("PLC directory {value:?} isn't valid http(s) base URL")] BadPlcDirectory { value: String }, #[error("identity {did} recently failed to resolve and is negatively cached")] @@ -90,6 +96,8 @@ impl ResolveError { | ResolveError::BadPds { .. } | ResolveError::MissingSpindle | ResolveError::BadSpindle { .. } + | ResolveError::MissingKnot + | ResolveError::BadKnot { .. } | ResolveError::BadPlcDirectory { .. } | ResolveError::RecentlyFailed { .. } | ResolveError::ExpectedKeyAbsent { .. } @@ -271,6 +279,32 @@ pub(crate) fn spindle_endpoint(did: &ServiceDid, body: &[u8]) -> Result Result { + let document: DidDocument = + serde_json::from_slice(body).map_err(|error| ResolveError::Malformed(error.to_string()))?; + if document.id.as_str() != did.as_str() { + return Err(ResolveError::IdMismatch { + requested: did.as_str().to_string(), + document: document.id.as_str().to_string(), + }); + } + let absolute = format!("{}{KNOT_SERVICE_ID}", did.as_str()); + let endpoint = document + .service + .iter() + .flatten() + .find(|service| { + let id = service.id.as_str(); + (id == KNOT_SERVICE_ID || id == absolute) + && service.r#type.as_str() == KNOT_SERVICE_TYPE + }) + .and_then(|service| service.service_endpoint.as_ref()?.as_str()) + .ok_or(ResolveError::MissingKnot)?; + Url::parse(endpoint).map_err(|_| ResolveError::BadKnot { + value: endpoint.to_string(), + }) +} + pub(crate) fn document_publishes_account_shape( did: &RepoDid, body: &[u8], diff --git a/knot2/crates/knot-bench/src/fixtures.rs b/knot2/crates/knot-bench/src/fixtures.rs index 906734baf..20a473e43 100644 --- a/knot2/crates/knot-bench/src/fixtures.rs +++ b/knot2/crates/knot-bench/src/fixtures.rs @@ -251,6 +251,8 @@ fn registration(index: RepoIndex) -> Registration { created_at: UnixSeconds::new(GENESIS_SECONDS + index.get() as i64), policy: knot_types::ContributionPolicy::default(), fostered: false, + + retired_at: None, } } diff --git a/knot2/crates/knot-cobs/src/comment.rs b/knot2/crates/knot-cobs/src/comment.rs index 2ed29e16d..ba2895794 100644 --- a/knot2/crates/knot-cobs/src/comment.rs +++ b/knot2/crates/knot-cobs/src/comment.rs @@ -1,7 +1,7 @@ use serde::{Deserialize, Serialize}; use knot_record::comment::CommentRef; -use knot_types::RecordCid; +use knot_types::{AtUri, RecordAddress, RecordBody, RecordCid}; use crate::Opening; @@ -11,12 +11,40 @@ pub struct Comment { pub reply: Option, } +#[derive(Deserialize)] +pub(crate) struct StoredRef { + pub(crate) uri: AtUri, + pub(crate) cid: RecordCid, +} + +fn stored_ref(reference: StoredRef) -> Result { + let (_, address) = RecordAddress::parse_at_uri(&reference.uri).map_err(|error| error.to_string())?; + Ok(CommentRef::new(address, reference.cid)) +} + +#[derive(Deserialize)] +struct StoredComment { + subject: StoredRef, + #[serde(rename = "replyTo", default)] + reply_to: Option, +} + +fn decode_stored(body: &RecordBody) -> Result { + let stored: StoredComment = serde_ipld_dagcbor::from_slice(body.as_bytes()) + .map_err(|error| error.to_string())?; + Ok(Comment { + subject: stored.subject.cid, + reply: stored.reply_to.map(stored_ref).transpose()?, + }) +} + crate::social_cob! { body = Comment, change = CommentChange, cob = CommentCob, collection = knot_record::comment::COMMENT_COLLECTION, opening = Opening::Contribution, + stored = decode_stored, } #[cfg(test)] diff --git a/knot2/crates/knot-cobs/src/label.rs b/knot2/crates/knot-cobs/src/label.rs index 403ac2772..d08e1b063 100644 --- a/knot2/crates/knot-cobs/src/label.rs +++ b/knot2/crates/knot-cobs/src/label.rs @@ -1,3 +1,4 @@ +use knot_types::RecordBody; use serde::{Deserialize, Serialize}; use knot_record::label::LABEL_DEF_COLLECTION; @@ -5,12 +6,17 @@ use knot_record::label::LABEL_DEF_COLLECTION; #[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)] pub struct LabelDef; +fn decode_stored(_: &RecordBody) -> Result { + Ok(LabelDef) +} + crate::social_cob! { body = LabelDef, change = LabelDefChange, cob = LabelDefCob, collection = LABEL_DEF_COLLECTION, opening = crate::Opening::Ownership, + stored = decode_stored, } #[cfg(test)] diff --git a/knot2/crates/knot-cobs/src/lib.rs b/knot2/crates/knot-cobs/src/lib.rs index 0f04fe6f5..019ba54fe 100644 --- a/knot2/crates/knot-cobs/src/lib.rs +++ b/knot2/crates/knot-cobs/src/lib.rs @@ -24,15 +24,16 @@ pub use pull::{Patch, PatchChange, PatchCob}; pub use reaction::{Reaction, ReactionChange, ReactionCob, ReactionKind}; pub use registry::{ Policy, Registration, RegistrationField, Registry, RegistryChange, RegistryError, Rename, - RepoRecord, RepoRef, RepoRegistryCob, deregister_repo, register_repo, rename_repo, - set_repo_policy, + RepoRecord, RepoRef, RepoRegistryCob, Retired, deregister_repo, register_repo, rename_repo, + retire_repo, set_repo_policy, }; pub use social::{ - AnyBody, AnySocialChange, Creation, Editing, Emission, Erasure, Opening, Social, SocialBody, - SocialChange, SocialCob, SocialObject, SocialState, decode_social, + AnyBody, AnySocialChange, Creation, Editing, Emission, Erasure, Opening, SettleError, Settled, + Social, SocialBody, SocialChange, SocialCob, SocialObject, SocialState, decode_social, }; pub use ticket::{ - Ticket, TicketChange, TicketCob, history_model_of, social_collections, social_state, + Ticket, TicketChange, TicketCob, history_model_of, settle_stored, social_collections, + social_state, }; #[doc(hidden)] diff --git a/knot2/crates/knot-cobs/src/manifest.rs b/knot2/crates/knot-cobs/src/manifest.rs index f790f0fb5..bf8dbbc83 100644 --- a/knot2/crates/knot-cobs/src/manifest.rs +++ b/knot2/crates/knot-cobs/src/manifest.rs @@ -1,3 +1,4 @@ +use knot_types::RecordBody; use serde::{Deserialize, Serialize}; use knot_record::manifest::MANIFEST_COLLECTION; @@ -5,12 +6,17 @@ use knot_record::manifest::MANIFEST_COLLECTION; #[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)] pub struct Manifest; +fn decode_stored(_: &RecordBody) -> Result { + Ok(Manifest) +} + crate::social_cob! { body = Manifest, change = ManifestChange, cob = ManifestCob, collection = MANIFEST_COLLECTION, opening = crate::Opening::Ownership, + stored = decode_stored, } #[cfg(test)] diff --git a/knot2/crates/knot-cobs/src/pull.rs b/knot2/crates/knot-cobs/src/pull.rs index 0619f9582..fe07c13cf 100644 --- a/knot2/crates/knot-cobs/src/pull.rs +++ b/knot2/crates/knot-cobs/src/pull.rs @@ -1,3 +1,4 @@ +use knot_types::RecordBody; use serde::{Deserialize, Serialize}; use crate::Opening; @@ -5,12 +6,17 @@ use crate::Opening; #[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)] pub struct Patch; +fn decode_stored(_: &RecordBody) -> Result { + Ok(Patch) +} + crate::social_cob! { body = Patch, change = PatchChange, cob = PatchCob, collection = knot_record::pull::PATCH_COLLECTION, opening = Opening::Contribution, + stored = decode_stored, } #[cfg(test)] diff --git a/knot2/crates/knot-cobs/src/reaction.rs b/knot2/crates/knot-cobs/src/reaction.rs index b0f1026cc..712b947f1 100644 --- a/knot2/crates/knot-cobs/src/reaction.rs +++ b/knot2/crates/knot-cobs/src/reaction.rs @@ -1,9 +1,10 @@ use serde::{Deserialize, Serialize}; pub use knot_record::reaction::ReactionKind; -use knot_types::RecordCid; +use knot_types::{RecordBody, RecordCid}; use crate::Opening; +use crate::comment::StoredRef; #[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)] pub struct Reaction { @@ -11,12 +12,28 @@ pub struct Reaction { pub kind: ReactionKind, } +#[derive(Deserialize)] +struct StoredReaction { + subject: StoredRef, + reaction: ReactionKind, +} + +fn decode_stored(body: &RecordBody) -> Result { + let stored: StoredReaction = serde_ipld_dagcbor::from_slice(body.as_bytes()) + .map_err(|error| error.to_string())?; + Ok(Reaction { + subject: stored.subject.cid, + kind: stored.reaction, + }) +} + crate::social_cob! { body = Reaction, change = ReactionChange, cob = ReactionCob, collection = knot_record::reaction::REACTION_COLLECTION, opening = Opening::Contribution, + stored = decode_stored, } #[cfg(test)] mod tests { diff --git a/knot2/crates/knot-cobs/src/registry.rs b/knot2/crates/knot-cobs/src/registry.rs index c54c4c282..e582169e5 100644 --- a/knot2/crates/knot-cobs/src/registry.rs +++ b/knot2/crates/knot-cobs/src/registry.rs @@ -20,6 +20,8 @@ pub struct Registration { pub policy: ContributionPolicy, #[serde(default)] pub fostered: bool, + #[serde(default)] + pub retired_at: Option, } #[derive(Debug, Clone, Copy, PartialEq, Eq)] @@ -87,6 +89,14 @@ pub struct RepoRef { pub rkey: RepoRkey, } +#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] +pub struct Retired { + pub owner: OwnerDid, + pub rkey: RepoRkey, + pub repo: RepoDid, + pub at: UnixSeconds, +} + #[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] #[serde(tag = "op", content = "data", rename_all = "snake_case")] pub enum RegistryChange { @@ -94,6 +104,7 @@ pub enum RegistryChange { Rename(Rename), Deregister(RepoRef), SetPolicy(Policy), + Retired(Retired), } impl ChangePayload for RegistryChange { @@ -112,6 +123,8 @@ pub struct RepoRecord { pub policy: ContributionPolicy, #[serde(default)] pub fostered: bool, + #[serde(default)] + pub retired_at: Option, } #[derive(Debug, Clone, PartialEq, Eq, Default, Serialize, Deserialize)] @@ -175,6 +188,7 @@ impl Registry { created_at: registration.created_at, policy: registration.policy, fostered: registration.fostered, + retired_at: registration.retired_at, }, ); self @@ -215,6 +229,21 @@ impl Registry { self } + fn retire(mut self, change: Retired) -> Self { + if let Some(record) = self + .records + .get_mut(&change.repo) + .filter(|record| record.owner == change.owner) + { + let at = match record.retired_at { + Some(existing) => existing.max(change.at), + None => change.at, + }; + record.retired_at = Some(at); + } + self + } + fn steal_alias(mut self, owner: &OwnerDid, rkey: &RepoRkey, target: &RepoDid) -> Self { match self.resolve(owner, rkey).cloned() { Some(holder) if holder != *target => { @@ -267,6 +296,7 @@ impl Evaluate for RepoRegistryCob { RegistryChange::Rename(rename) => state.rename(rename), RegistryChange::Deregister(target) => state.deregister(target), RegistryChange::SetPolicy(policy) => state.set_policy(policy), + RegistryChange::Retired(change) => state.retire(change), } } } @@ -454,6 +484,40 @@ pub fn set_repo_policy( ) } +pub fn retire_repo( + store: &CobStore, + home: &CobHome, + object: CobId, + change: Retired, + signer: &dyn Signer, + timestamp: UnixSeconds, +) -> Result, RegistryError> { + store.update_maybe_checkpointed::( + home, + object, + signer, + timestamp, + |registry| { + let record = + registry + .record_of(&change.repo) + .ok_or_else(|| RegistryError::NotHosted { + repo: change.repo.clone(), + })?; + if record.owner != change.owner { + return Err(RegistryError::OwnerMoved { + repo: change.repo.clone(), + expected: change.owner.clone(), + }); + } + match record.retired_at { + Some(existing) if existing >= change.at => Ok(None), + _ => Ok(Some(RegistryChange::Retired(change.clone()))), + } + }, + ) +} + #[cfg(test)] mod tests { use super::*; @@ -483,6 +547,7 @@ mod tests { created_at: UnixSeconds::new(at), policy: ContributionPolicy::default(), fostered: false, + retired_at: None, }) } @@ -502,6 +567,15 @@ mod tests { }) } + fn retired(owner_id: &str, key: &str, repo_id: &str, at: i64) -> RegistryChange { + RegistryChange::Retired(Retired { + owner: owner(owner_id), + rkey: rkey(key), + repo: repo(repo_id), + at: UnixSeconds::new(at), + }) + } + fn fold(changes: Vec) -> Registry { let author = ActorId::from_secp256k1(&[0x02; 33]); changes @@ -729,4 +803,90 @@ mod tests { "Fold answers to the owner the registry has on file, not to what the change claims" ); } + + #[test] + fn retire_records_acked_at_and_keeps_latest() { + let state = fold(vec![ + register("nel", "anemone", "squid", 1), + retired("nel", "anemone", "squid", 10), + ]); + assert_eq!( + state.record_of(&repo("squid")).unwrap().retired_at, + Some(UnixSeconds::new(10)) + ); + + let later = fold(vec![ + register("nel", "anemone", "squid", 1), + retired("nel", "anemone", "squid", 10), + retired("nel", "anemone", "squid", 20), + ]); + assert_eq!( + later.record_of(&repo("squid")).unwrap().retired_at, + Some(UnixSeconds::new(20)), + "a later ack wins" + ); + + let stale = fold(vec![ + register("nel", "anemone", "squid", 1), + retired("nel", "anemone", "squid", 20), + retired("nel", "anemone", "squid", 5), + ]); + assert_eq!( + stale.record_of(&repo("squid")).unwrap().retired_at, + Some(UnixSeconds::new(20)), + "an old ack can't retire a fresh shadow" + ); + } + + #[test] + fn retire_under_mismatched_owner_is_no_op() { + let state = fold(vec![ + register("nel", "anemone", "squid", 1), + retired("olaren", "anemone", "squid", 10), + ]); + assert_eq!( + state.record_of(&repo("squid")).unwrap().retired_at, + None, + "only the owner on file can retire" + ); + } + + #[test] + fn registration_before_retired_field_still_applies() { + #[derive(Serialize)] + struct Before<'a> { + owner: &'a str, + rkey: &'a str, + name: &'a str, + repo: &'a str, + created_at: i64, + policy: ContributionPolicy, + fostered: bool, + } + + #[derive(Serialize)] + #[serde(tag = "op", content = "data", rename_all = "snake_case")] + enum BeforeChange<'a> { + Register(Before<'a>), + } + + let bytes = serde_ipld_dagcbor::to_vec(&BeforeChange::Register(Before { + owner: "did:plc:nel", + rkey: "anemone", + name: "anemone", + repo: "did:plc:squid", + created_at: 5, + policy: ContributionPolicy::Collaborators, + fostered: false, + })) + .unwrap(); + let decoded = RegistryChange::decode(&bytes) + .expect("Change from before this field existed still applies"); + let state = fold(vec![decoded]); + assert_eq!( + state.record_of(&repo("squid")).unwrap().retired_at, + None, + "a repo registered before this field existed isn't retired" + ); + } } diff --git a/knot2/crates/knot-cobs/src/social.rs b/knot2/crates/knot-cobs/src/social.rs index 6ab6692b2..efe9e3974 100644 --- a/knot2/crates/knot-cobs/src/social.rs +++ b/knot2/crates/knot-cobs/src/social.rs @@ -2,12 +2,14 @@ use std::fmt::Debug; use std::marker::PhantomData; use knot_cob::{ - Append, Authored, Authorize, ChangePayload, Checkpoint, Emits, Evaluate, HistoryModel, - SnapshotStride, StateSize, WriteContext, + Approved, Append, Authored, Authorize, ChangePayload, Checkpoint, CobHome, CobStore, Emits, + Evaluate, HistoryModel, Materialized, Record, SnapshotStride, StateSize, WriteContext, }; +use knot_runtime::Signer; use knot_types::{ - AccountDid, ActorId, Blocked, ContributionPermission, Decision, RecordAddress, RecordCid, - RecordCollection, RepoRole, SubjectNumber, UnixSeconds, + AccountDid, ActorId, Blocked, ContributionPermission, ContributionPolicy, Decision, + RecordAddress, RecordBody, RecordCid, RecordCollection, RepoDid, RepoRole, SubjectNumber, + UnixSeconds, }; use serde::de::{DeserializeOwned, DeserializeSeed, MapAccess, SeqAccess, Visitor}; use serde::{Deserialize, Serialize}; @@ -88,6 +90,67 @@ pub trait SocialBody: { const COLLECTION: &'static str; const OPENING: Opening; + fn from_stored(body: &RecordBody) -> Result; +} + +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub enum Settled { + AtSeed, + Superseded, +} + +#[derive(Debug, thiserror::Error)] +pub enum SettleError { + #[error(transparent)] + Cob(#[from] knot_cob::CobError), + #[error("{0}")] + Stored(String), +} + +pub(crate) fn settle( + store: &CobStore, + home: &CobHome, + repo: &RepoDid, + signer: &dyn Signer, + creation: Creation, + record: RecordBody, +) -> Result { + let created_at = creation.created_at; + let body = B::from_stored(&record).map_err(SettleError::Stored)?; + let permission = + ContributionPermission::new(Blocked::No, RepoRole::Owner, ContributionPolicy::Anyone); + let author = creation.author.clone(); + let cx = WriteContext::new(&author, repo, permission); + let existing = store + .list::>()? + .into_iter() + .map(|object| { + store + .root_address::>(object) + .map(|found| found.filter(|found| found == &creation.address).map(|_| object)) + }) + .find_map(Result::transpose) + .transpose()?; + match existing { + None => { + let materialized = Materialized { + change: SocialChange::Open { creation, body }, + record: Record::version(record), + }; + let approved = Approved::signed_by(&cx, materialized)?; + store.create_approved::>(home, &approved, signer, created_at)?; + Ok(Settled::AtSeed) + } + Some(object) => { + let state = store.get::>(object)?; + let current = state.state().object().and_then(|live| live.version()); + if current != Some(creation.produces) { + Ok(Settled::Superseded) + } else { + Ok(Settled::AtSeed) + } + } + } } #[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] #[serde(tag = "op", content = "data", rename_all = "snake_case")] @@ -195,11 +258,16 @@ macro_rules! social_cob { change = $change:ident, cob = $cob:ident, collection = $collection:expr, - opening = $opening:expr $(,)? + opening = $opening:expr, + stored = $stored:path $(,)? ) => { impl $crate::SocialBody for $body { const COLLECTION: &'static str = $collection; const OPENING: $crate::Opening = $opening; + + fn from_stored(body: &knot_types::RecordBody) -> Result { + $stored(body) + } } pub type $change = $crate::SocialChange<$body>; @@ -498,12 +566,19 @@ mod tests { text: String, } + fn note_stored(_: &RecordBody) -> Result { + Ok(NoteBody { + text: String::new(), + }) + } + social_cob! { body = NoteBody, change = NoteChange, cob = NoteCob, collection = "org.tangled.test.note", opening = Opening::Contribution, + stored = note_stored, } fn who(suffix: u8) -> AccountDid { diff --git a/knot2/crates/knot-cobs/src/ticket.rs b/knot2/crates/knot-cobs/src/ticket.rs index c98dd076a..f516f6d5e 100644 --- a/knot2/crates/knot-cobs/src/ticket.rs +++ b/knot2/crates/knot-cobs/src/ticket.rs @@ -1,5 +1,6 @@ +use knot_record::label::LabelToken; use knot_record::ticket::{TICKET_COLLECTION, TicketState}; -use knot_types::LabelKey; +use knot_types::{LabelKey, RecordBody}; use serde::{Deserialize, Serialize}; use crate::Opening; @@ -11,12 +12,34 @@ pub struct Ticket { pub tangled_labels: Option>, } +#[derive(Deserialize)] +struct StoredTicket { + state: TicketState, + #[serde(rename = "tangledLabels", default)] + tangled_labels: Option>, +} + +fn decode_stored(body: &RecordBody) -> Result { + let stored: StoredTicket = serde_ipld_dagcbor::from_slice(body.as_bytes()) + .map_err(|error| error.to_string())?; + Ok(Ticket { + state: stored.state, + tangled_labels: stored.tangled_labels.map(|tokens| { + tokens + .into_iter() + .map(|token| token.key().clone()) + .collect() + }), + }) +} + crate::social_cob! { body = Ticket, change = TicketChange, cob = TicketCob, collection = TICKET_COLLECTION, opening = Opening::Contribution, + stored = decode_stored, } macro_rules! social_kinds { @@ -51,6 +74,24 @@ macro_rules! social_kinds { _ => None, } } + + pub fn settle_stored( + store: &knot_cob::CobStore, + home: &knot_cob::CobHome, + repo: &knot_types::RepoDid, + signer: &dyn knot_runtime::Signer, + collection: &str, + creation: $crate::Creation, + record: knot_types::RecordBody, + ) -> Result, $crate::SettleError> { + match collection { + $(<$body as $crate::SocialBody>::COLLECTION => $crate::social::settle::<$body>( + store, home, repo, signer, creation, record, + ) + .map(Some),)+ + _ => Ok(None), + } + } }; } diff --git a/knot2/crates/knot-cobs/tests/common/mod.rs b/knot2/crates/knot-cobs/tests/common/mod.rs index 064c27fa8..d0bdfff63 100644 --- a/knot2/crates/knot-cobs/tests/common/mod.rs +++ b/knot2/crates/knot-cobs/tests/common/mod.rs @@ -65,6 +65,7 @@ pub fn registration(owner: &str, key: &str, repo_id: &str, ts: i64) -> Registrat created_at: at(ts), policy: knot_types::ContributionPolicy::default(), fostered: false, + retired_at: None, } } diff --git a/knot2/crates/knot-cobs/tests/social.rs b/knot2/crates/knot-cobs/tests/social.rs index 4f7137cd1..1bcd58a04 100644 --- a/knot2/crates/knot-cobs/tests/social.rs +++ b/knot2/crates/knot-cobs/tests/social.rs @@ -17,12 +17,19 @@ struct NoteBody { text: String, } +fn note_stored(_: &RecordBody) -> Result { + Ok(NoteBody { + text: String::new(), + }) +} + social_cob! { body = NoteBody, change = NoteChange, cob = NoteCob, collection = "org.tangled.test.note", opening = knot_cobs::Opening::Contribution, + stored = note_stored, } fn squid() -> RepoDid { diff --git a/knot2/crates/knot-index/src/projections.rs b/knot2/crates/knot-index/src/projections.rs index b63c58276..84347626e 100644 --- a/knot2/crates/knot-index/src/projections.rs +++ b/knot2/crates/knot-index/src/projections.rs @@ -767,6 +767,7 @@ impl RegistryProjection { self.apply_deregister(interner, target, displaced) } RegistryChange::SetPolicy(policy) => self.apply_policy(interner, policy, displaced), + RegistryChange::Retired(_) => displaced, }) } diff --git a/knot2/crates/knot-index/tests/common/mod.rs b/knot2/crates/knot-index/tests/common/mod.rs index 4298251ee..bfb1c20e1 100644 --- a/knot2/crates/knot-index/tests/common/mod.rs +++ b/knot2/crates/knot-index/tests/common/mod.rs @@ -81,6 +81,7 @@ pub fn registration(owner_id: &str, key: &str, repo: &RepoDid, seconds: i64) -> created_at: at(seconds), policy: ContributionPolicy::default(), fostered: false, + retired_at: None, } } @@ -99,6 +100,7 @@ pub fn registration_with_name( created_at: at(seconds), policy: ContributionPolicy::default(), fostered: false, + retired_at: None, } } @@ -567,6 +569,13 @@ knot_cobs::social_cob! { cob = NoteCob, collection = "org.tangled.test.note", opening = knot_cobs::Opening::Contribution, + stored = note_stored, +} + +fn note_stored(_: &knot_types::RecordBody) -> Result { + Ok(NoteBody { + text: String::new(), + }) } macro_rules! addressed { diff --git a/knot2/crates/knot-index/tests/meta_repo.rs b/knot2/crates/knot-index/tests/meta_repo.rs index 8cad69c0c..2deab342c 100644 --- a/knot2/crates/knot-index/tests/meta_repo.rs +++ b/knot2/crates/knot-index/tests/meta_repo.rs @@ -48,6 +48,7 @@ fn the_meta_repo_round_trips_every_projection_from_git() { created_at: at(1), policy: knot_types::ContributionPolicy::default(), fostered: false, + retired_at: None, }), &signer, at(1), @@ -65,6 +66,7 @@ fn the_meta_repo_round_trips_every_projection_from_git() { created_at: at(2), policy: knot_types::ContributionPolicy::default(), fostered: true, + retired_at: None, }), &signer, at(2), diff --git a/knot2/crates/knot-index/tests/properties.rs b/knot2/crates/knot-index/tests/properties.rs index 37b635b02..2bb1acd59 100644 --- a/knot2/crates/knot-index/tests/properties.rs +++ b/knot2/crates/knot-index/tests/properties.rs @@ -77,6 +77,7 @@ fn registry_change(op: u8, who: u8, rkey: u8, repo: u8, t: i64) -> RegistryChang created_at: UnixSeconds::new(t), policy: knot_types::ContributionPolicy::default(), fostered: false, + retired_at: None, }), 1 => RegistryChange::Rename(Rename { owner: owner(who), diff --git a/knot2/crates/knot-keyfill/tests/fill.rs b/knot2/crates/knot-keyfill/tests/fill.rs index 9dcf5f8d8..e5fa2558c 100644 --- a/knot2/crates/knot-keyfill/tests/fill.rs +++ b/knot2/crates/knot-keyfill/tests/fill.rs @@ -84,6 +84,7 @@ fn hosted_index(scratch: &TempDir, repos: &[Hosted], members: &[AccountDid]) -> created_at: UnixSeconds::new(1), policy: knot_types::ContributionPolicy::default(), fostered: false, + retired_at: None, }) }; let (first, rest) = repos.split_first().expect("a knot under test hosts a repo"); diff --git a/knot2/crates/knot-migrate/src/emit.rs b/knot2/crates/knot-migrate/src/emit.rs index 4f0044dbd..f2b597138 100644 --- a/knot2/crates/knot-migrate/src/emit.rs +++ b/knot2/crates/knot-migrate/src/emit.rs @@ -322,6 +322,8 @@ fn registration(repo: &AdoptRepo) -> Registration { created_at: repo.created_at, policy: knot_types::ContributionPolicy::default(), fostered: false, + + retired_at: None, } } diff --git a/knot2/crates/knot-server/src/main.rs b/knot2/crates/knot-server/src/main.rs index fbc0d3687..817f7bc3d 100644 --- a/knot2/crates/knot-server/src/main.rs +++ b/knot2/crates/knot-server/src/main.rs @@ -826,6 +826,10 @@ async fn run(config: knot_config::Validated) -> anyhow::Result<()> { Arc::clone(&xrpc_state), shutdown.clone(), )); + tokio::spawn(knot_xrpc::foster::run_retire_gc( + Arc::clone(&xrpc_state), + shutdown.clone(), + )); let mut edge_task = tokio::spawn(knot_edge::serve( edge_config, app, diff --git a/knot2/crates/knot-sim/src/harness.rs b/knot2/crates/knot-sim/src/harness.rs index b6c3a6782..742b7715d 100644 --- a/knot2/crates/knot-sim/src/harness.rs +++ b/knot2/crates/knot-sim/src/harness.rs @@ -28,10 +28,9 @@ use knot_types::{ KnotServiceUrl, Oid, OwnerDid, RecordRkey, RefName, RepoDid, RepoName, RepoRkey, UnixSeconds, }; use knot_xrpc::{ - BlobReadBudget, BodyLimit, Budgets, ByteLimits, CobLocks, GlobalInflight, - GlobalQuota, LanguagesReadBudget, LimitConfig, MaxWireBytes, PerActorQuota, PerPeerInflight, - PreAuthLimiter, ReadBudget, ReservationTtl, Reservations, ResponseLimit, TreeReadBudget, - XrpcState, + BlobReadBudget, BodyLimit, Budgets, ByteLimits, CobLocks, GlobalInflight, GlobalQuota, + LanguagesReadBudget, LimitConfig, MaxWireBytes, PerActorQuota, PerPeerInflight, PreAuthLimiter, + ReadBudget, ReservationTtl, Reservations, ResponseLimit, TreeReadBudget, XrpcState, }; use crate::realdata::{ @@ -711,6 +710,8 @@ fn register_seed_repo( created_at: UnixSeconds::new(1), policy: knot_types::ContributionPolicy::default(), fostered: false, + + retired_at: None, }), &signer, UnixSeconds::new(1), diff --git a/knot2/crates/knot-sim/tests/lfs_chaos.rs b/knot2/crates/knot-sim/tests/lfs_chaos.rs index 1b6bc7c32..e423ce7b7 100644 --- a/knot2/crates/knot-sim/tests/lfs_chaos.rs +++ b/knot2/crates/knot-sim/tests/lfs_chaos.rs @@ -74,6 +74,7 @@ fn build_fixture(root: &Path) -> Paths { created_at: UnixSeconds::new(1), policy: knot_types::ContributionPolicy::default(), fostered: false, + retired_at: None, }), &signer, UnixSeconds::new(1), diff --git a/knot2/crates/knot-sim/tests/lfs_roundtrip.rs b/knot2/crates/knot-sim/tests/lfs_roundtrip.rs index 518a9498b..493b23f93 100644 --- a/knot2/crates/knot-sim/tests/lfs_roundtrip.rs +++ b/knot2/crates/knot-sim/tests/lfs_roundtrip.rs @@ -21,8 +21,8 @@ use knot_runtime::{ }; use knot_secrets::{MasterKey, SealedStore}; use knot_types::{ - AccountDid, ActorId, AdmissionPolicy, Did, KnotHostname, KnotId, - KnotServiceUrl, Nsid, OwnerDid, RepoDid, RepoName, RepoRkey, ServiceDid, UnixSeconds, + AccountDid, ActorId, AdmissionPolicy, Did, KnotHostname, KnotId, KnotServiceUrl, Nsid, + OwnerDid, RepoDid, RepoName, RepoRkey, ServiceDid, UnixSeconds, }; use tempfile::TempDir; use tokio::net::TcpListener; @@ -305,6 +305,7 @@ async fn spawn(published_line: String, with_h3: bool) -> World { created_at: UnixSeconds::new(1), policy: knot_types::ContributionPolicy::default(), fostered: false, + retired_at: None, }), &knot_signer, UnixSeconds::new(1), diff --git a/knot2/crates/knot-sim/tests/social_chaos.rs b/knot2/crates/knot-sim/tests/social_chaos.rs index 56cdd3cb3..592d34d6b 100644 --- a/knot2/crates/knot-sim/tests/social_chaos.rs +++ b/knot2/crates/knot-sim/tests/social_chaos.rs @@ -79,6 +79,7 @@ fn build_target(root: &Path) { created_at: UnixSeconds::new(1), policy: knot_types::ContributionPolicy::default(), fostered: false, + retired_at: None, }), &signer, UnixSeconds::new(1), diff --git a/knot2/crates/knot-sim/tests/ssh.rs b/knot2/crates/knot-sim/tests/ssh.rs index 1ccd125a7..f281ac0d8 100644 --- a/knot2/crates/knot-sim/tests/ssh.rs +++ b/knot2/crates/knot-sim/tests/ssh.rs @@ -172,6 +172,7 @@ async fn spawn(published_line: String) -> Server { created_at: UnixSeconds::new(1), policy: knot_types::ContributionPolicy::default(), fostered: false, + retired_at: None, }), &signer, UnixSeconds::new(1), diff --git a/knot2/crates/knot-ssh/examples/ephemeral_knot.rs b/knot2/crates/knot-ssh/examples/ephemeral_knot.rs index 3bc1aa2c3..5c3104fe2 100644 --- a/knot2/crates/knot-ssh/examples/ephemeral_knot.rs +++ b/knot2/crates/knot-ssh/examples/ephemeral_knot.rs @@ -180,6 +180,8 @@ async fn main() { created_at: UnixSeconds::new(1), policy: knot_types::ContributionPolicy::default(), fostered: false, + + retired_at: None, }), &signer, UnixSeconds::new(1), diff --git a/knot2/crates/knot-ssh/tests/ssh_push.rs b/knot2/crates/knot-ssh/tests/ssh_push.rs index 0e0bfd7dc..db4421751 100644 --- a/knot2/crates/knot-ssh/tests/ssh_push.rs +++ b/knot2/crates/knot-ssh/tests/ssh_push.rs @@ -360,6 +360,7 @@ async fn spawn_server_core( created_at: UnixSeconds::new(1), policy: knot_types::ContributionPolicy::default(), fostered: false, + retired_at: None, }), &signer, UnixSeconds::new(1), @@ -381,6 +382,7 @@ async fn spawn_server_core( created_at: UnixSeconds::new(2), policy: knot_types::ContributionPolicy::default(), fostered: false, + retired_at: None, }), &signer, UnixSeconds::new(2), @@ -1583,6 +1585,7 @@ fn registered_index( created_at: UnixSeconds::new(1), policy: knot_types::ContributionPolicy::default(), fostered: false, + retired_at: None, }), &signer, UnixSeconds::new(1), @@ -1728,6 +1731,7 @@ async fn a_collaborator_pushes_its_repo_but_is_denied_on_a_repo_it_doesnt_collab created_at: UnixSeconds::new(1), policy: knot_types::ContributionPolicy::default(), fostered: false, + retired_at: None, }), &signer, UnixSeconds::new(1), @@ -1745,6 +1749,7 @@ async fn a_collaborator_pushes_its_repo_but_is_denied_on_a_repo_it_doesnt_collab created_at: UnixSeconds::new(2), policy: knot_types::ContributionPolicy::default(), fostered: false, + retired_at: None, }), &signer, UnixSeconds::new(2), diff --git a/knot2/crates/knot-types/src/ids.rs b/knot2/crates/knot-types/src/ids.rs index 38f6fb994..1a09cd1b8 100644 --- a/knot2/crates/knot-types/src/ids.rs +++ b/knot2/crates/knot-types/src/ids.rs @@ -290,8 +290,11 @@ fn parse_record_collection(value: &str) -> Option> { fn parse_did_rkey(value: &str) -> Option { let did = parse_did(value)?; Rkey::::new_owned(did.as_str()).ok()?; - let ported = matches!(did.as_str().strip_prefix("did:web:"), Some(host) if host.contains(':')); - (!ported).then_some(did) + (!ported_did_web(value)).then_some(did) +} + +fn ported_did_web(value: &str) -> bool { + matches!(value.strip_prefix("did:web:"), Some(host) if host.contains(':')) } fn parse_ref_name(value: &str) -> Option { @@ -473,6 +476,7 @@ impl RecordRkey { let value = value.into(); match parse_tid(&value) { Some(tid) => Ok(Self::Minted(tid)), + None if ported_did_web(&value) => Err(ParseError::invalid("record key", value)), None => DidRkey::new(value.clone()) .map(Self::Subject) .or_else(|_| LabelKey::new(value).map(Self::Composed)), @@ -1422,6 +1426,18 @@ mod tests { assert_eq!(RepoDid::from(&rkey).as_str(), rkey.as_str()); } + #[test] + fn ported_did_web_record_key_rejected() { + assert!( + RecordRkey::new("did:web:oyster.cafe:3333").is_err(), + "a ported did:web is nobody's record key" + ); + assert!( + RecordRkey::new("did:web:oyster.cafe").is_ok(), + "an unported did:web stays a subject record key" + ); + } + #[test] fn a_ci_logs_address_splits_into_a_bare_host_and_a_nonzero_port() { let addr = CiLogsAddr::new("logs.oyster.cafe:3333").unwrap(); diff --git a/knot2/crates/knot-xrpc/src/error.rs b/knot2/crates/knot-xrpc/src/error.rs index 412e39f4c..22d1b7fb2 100644 --- a/knot2/crates/knot-xrpc/src/error.rs +++ b/knot2/crates/knot-xrpc/src/error.rs @@ -246,6 +246,15 @@ impl From for XrpcError { } } +impl From for XrpcError { + fn from(error: knot_cobs::SettleError) -> Self { + match error { + knot_cobs::SettleError::Cob(cob) => cob.into(), + knot_cobs::SettleError::Stored(message) => Self::internal(message), + } + } +} + impl From for XrpcError { fn from(error: knot_types::RecordBodyError) -> Self { use knot_types::RecordBodyError; diff --git a/knot2/crates/knot-xrpc/src/foster.rs b/knot2/crates/knot-xrpc/src/foster.rs index 534a6239f..45ebae083 100644 --- a/knot2/crates/knot-xrpc/src/foster.rs +++ b/knot2/crates/knot-xrpc/src/foster.rs @@ -1,4 +1,5 @@ use std::collections::BTreeMap; +use std::path::Path; use std::str::FromStr; use std::sync::Arc; @@ -9,33 +10,45 @@ use axum::extract::{DefaultBodyLimit, State}; use axum::middleware::from_fn_with_state; use axum::response::{IntoResponse, Response}; use axum::routing::post; +use base64::Engine as _; +use base64::engine::general_purpose::URL_SAFE_NO_PAD; use cid::Cid; -use http::HeaderMap; +use http::{HeaderMap, StatusCode}; use ipld_core::ipld::Ipld; use serde::{Deserialize, Serialize}; +use knot_capability::Capability; use knot_cob::{ ChangePayload, CobHome, CobId, CobStore, Emits, HistoryModel, KnotAuthored, Materialized, Record, }; +use knot_cobs::{ + Registry, RepoRef, RepoRegistryCob, Retired, deregister_repo, retire_repo, +}; use knot_git::Repo; use knot_index::Resolved; use knot_record::chain; use knot_record::manifest::ManifestMeta; -use knot_runtime::{Clock, HttpTransport, Signer}; +use knot_runtime::{Clock, HttpTransport, PublicKeyBytes, Signature, Signer}; +use knot_types::crypto::{KeyCodec, PublicKey as CryptoKey}; use knot_types::{ AccountDid, ActorId, ContributionPolicy, MstKey, OwnerDid, RecordAddress, RecordBody, RecordBodyError, RecordCid, RecordCollection, RecordRkey, RepoDid, RepoName, RepoRkey, UnixSeconds, }; +use url::Url; use crate::XrpcState; use crate::error::XrpcError; +use crate::handover::{advertises, get_json, knot_identity}; use crate::query::ValidatedQuery; use crate::{decode, enforce_pre_auth_limit, run_blocking}; pub const SEED_ROUTE: &str = "/xrpc/org.tangled.temp.server.seedRepo"; pub const UPLOAD_BLOB_ROUTE: &str = "/xrpc/org.tangled.temp.server.uploadBlob"; +pub const ACK_ROUTE: &str = "/xrpc/org.tangled.temp.server.ackHandover"; + +pub(crate) const ACK_CONTEXT: &str = "org.tangled.temp.handover.ack.v1"; const MAX_RECORDS_PER_REQUEST: usize = 200; @@ -85,6 +98,41 @@ struct RecordResult { error: Option, } +#[derive(Deserialize)] +struct AckInput { + context: String, + #[serde(rename = "repoDid")] + repo_did: RepoDid, + records: RecordCount, + #[serde(rename = "ackedAt")] + acked_at: UnixSeconds, + signature: String, +} + +#[derive(Serialize)] +pub(crate) struct AckSigned<'a> { + pub(crate) context: &'static str, + #[serde(rename = "repoDid")] + pub(crate) repo_did: &'a RepoDid, + pub(crate) records: RecordCount, + #[serde(rename = "ackedAt")] + pub(crate) acked_at: UnixSeconds, +} + +#[derive(Clone, Copy, Debug, Serialize, Deserialize)] +#[serde(transparent)] +pub(crate) struct RecordCount(usize); + +impl RecordCount { + pub(crate) const fn new(records: usize) -> Self { + Self(records) + } + + pub(crate) const fn get(self) -> usize { + self.0 + } +} + struct CheckedRecord { address: RecordAddress, author: AccountDid, @@ -207,6 +255,7 @@ pub fn router(state: Arc>) -> Router let limiter = Arc::clone(&state); Router::new() .route(SEED_ROUTE, post(seed_repo::)) + .route(ACK_ROUTE, post(ack_handover::)) .route( UPLOAD_BLOB_ROUTE, post(upload_blob::).layer(DefaultBodyLimit::max( @@ -250,6 +299,13 @@ async fn seed_repo( .map_err(|error| XrpcError::invalid_request(error.to_string()))?; let rkey = RepoRkey::from_repo_name(&name) .map_err(|error| XrpcError::invalid_request(error.to_string()))?; + if let Some(record) = read_registry_record(state, &repo_did).await? + && record.retired_at.is_some() + { + return Err(XrpcError::conflict( + "the repo's shadow was retired by a handover ack, so it no longer accepts seeds", + )); + } let (repo_did, key) = match hosted(state, &owner, &rkey)? { Some(existing) => { if existing != repo_did { @@ -268,7 +324,10 @@ async fn seed_repo( let public = state.secrets.public_key(&state.knot_did)?; (existing, ActorId::from_secp256k1(public.as_bytes())) } - None => foster_repo(state, owner, name, rkey, repo_did, manifest, policy).await?, + None => { + seed_check(state, &repo_did).await?; + foster_repo(state, owner, name, rkey, repo_did, manifest, policy).await? + } }; let results = seed_records(state, &repo_did, records).await?; Ok(( @@ -306,6 +365,240 @@ async fn upload_blob( crate::blobs::store_upload(state, params.repo_did, &headers, body).await } +async fn ack_handover( + State(foster): State>>, + body: Bytes, +) -> Result { + let state = &foster.state; + let input: AckInput = decode(&body)?; + if input.context != ACK_CONTEXT { + return Err(XrpcError::invalid_request( + "the ack context isn't the handover ack context", + )); + } + if input.acked_at.get() < 0 { + return Err(XrpcError::invalid_request( + "the ack's ackedAt predates the epoch", + )); + } + + let Some(record) = read_registry_record(state, &input.repo_did) + .await? + .filter(|record| record.fostered) + else { + return Err(XrpcError::not_found("the repo isn't fostered on this knot")); + }; + if let Some(retired_at) = record.retired_at + && retired_at >= input.acked_at + { + return Ok((StatusCode::OK, Json(serde_json::json!({}))).into_response()); + } + + let origin = origin_knot_url(state, &input.repo_did).await?; + let key = origin_key(state, &origin).await?; + let signed = AckSigned { + context: ACK_CONTEXT, + repo_did: &input.repo_did, + records: input.records, + acked_at: input.acked_at, + }; + let bytes = serde_ipld_dagcbor::to_vec(&signed) + .map_err(|error| XrpcError::internal(error.to_string()))?; + let signature = Signature::from_bytes( + URL_SAFE_NO_PAD + .decode(&input.signature) + .map_err(|_| XrpcError::invalid_request("the ack signature isn't base64url"))?, + ); + if !knot_runtime::verify(&key, &bytes, &signature) { + return Err(XrpcError::forbidden( + "the ack signature doesn't verify against the origin knot's key", + )); + } + spot_check_record(state, &origin, &input.repo_did).await?; + retire(state, &input.repo_did, input.acked_at).await?; + Ok((StatusCode::OK, Json(serde_json::json!({}))).into_response()) +} + +async fn seed_check( + state: &XrpcState, + repo_did: &RepoDid, +) -> Result<(), XrpcError> { + let Ok(origin) = origin_knot_url(state, repo_did).await else { + return Ok(()); + }; + let Ok(url) = origin_url(&origin, "xrpc/org.tangled.knot.describeKnot") else { + return Ok(()); + }; + let Ok(described) = get_json(state.git_http.as_ref(), url, "origin knot").await else { + return Ok(()); + }; + if advertises(&described, &Capability::SocialWrites) { + return Err(XrpcError::conflict( + "the origin knot already serves COBs, so this repo can't be shadowed", + )); + } + Ok(()) +} + +async fn origin_knot_url( + state: &XrpcState, + repo_did: &RepoDid, +) -> Result { + let url = state + .atproto + .resolve_knot_endpoint(repo_did) + .await + .map_err(|error| match error { + knot_atproto::AtprotoError::Resolve(_) => XrpcError::invalid_request(error.to_string()), + other => other.into(), + })?; + if !crate::handover::url_is_trusted(&url) { + return Err(XrpcError::invalid_request(format!( + "the repo's knot endpoint {} must be https, or http on loopback", + url.as_str() + ))); + } + Ok(url) +} + +async fn origin_key( + state: &XrpcState, + origin: &Url, +) -> Result { + let multibase = knot_identity( + state, origin, "origin knot", &Capability::SocialWrites, + "doesn't serve COBs, so it can't be the origin of a retired shadow", + ) + .await?; + let key = CryptoKey::decode_owned(multibase) + .map_err(|_| XrpcError::bad_gateway("the origin knot's multikey doesn't parse as a key"))?; + if key.codec != KeyCodec::Secp256k1 { + return Err(XrpcError::bad_gateway("the origin knot's #atproto key isn't secp256k1")); + } + Ok(PublicKeyBytes::from_bytes(key.bytes.as_ref().to_vec())) +} + +async fn spot_check_record( + state: &XrpcState, + origin: &Url, + repo: &RepoDid, +) -> Result<(), XrpcError> { + let layout = state.layout.clone(); + let repo = repo.clone(); + let versions = run_blocking({ + let repo = repo.clone(); + move || { + let git = layout.open(&repo)?; + let store = CobStore::new(&git); + seeded_versions(&store) + } + }) + .await?; + let Some(version) = versions.first() else { + return Ok(()); + }; + let mut url = origin_url(origin, "xrpc/com.atproto.repo.getRecord")?; + url.query_pairs_mut() + .append_pair("repo", repo.as_str()) + .append_pair("collection", version.address.collection().as_str()) + .append_pair("rkey", version.address.rkey().as_str()); + let served = get_json(state.git_http.as_ref(), url, "origin knot").await?; + let cid = served["cid"] + .as_str() + .ok_or_else(|| XrpcError::bad_gateway("the origin knot served getRecord without a cid"))?; + if cid != version.produces.cid().to_string() { + return Err(XrpcError::bad_gateway(format!( + "the origin knot serves {} at {cid} while the shadow recorded {}", + version.address.at_uri(&repo), + version.produces.cid() + ))); + } + Ok(()) +} + +async fn retire( + state: &Arc>, + repo: &RepoDid, + at: UnixSeconds, +) -> Result<(), XrpcError> { + let signer = state.secrets.signer(&state.knot_did)?; + let meta_path = state.meta_path.clone(); + let cob_locks = Arc::clone(&state.cob_locks); + let knot_did = state.knot_did.clone(); + let repo = repo.clone(); + run_blocking(move || { + let _guard = cob_locks.meta(); + let home = CobHome::from(&knot_did); + with_registry(&meta_path, |store, object, registry| { + let record = registry + .record_of(&repo) + .ok_or_else(|| XrpcError::not_found("the repo isn't registered on this knot"))?; + let change = Retired { + owner: record.owner.clone(), + rkey: record.rkey.clone(), + repo: repo.clone(), + at, + }; + retire_repo(store, &home, object, change, &signer, at) + .map(|_| ()) + .map_err(XrpcError::from) + })? + .ok_or_else(|| XrpcError::not_found("no git repos registered on this knot")) + }) + .await +} + +async fn read_registry_record( + state: &XrpcState, + repo: &RepoDid, +) -> Result, XrpcError> { + let meta_path = state.meta_path.clone(); + let repo = repo.clone(); + run_blocking(move || { + with_registry(&meta_path, |_, _, registry| { + Ok(registry.record_of(&repo).cloned()) + }) + .map(Option::flatten) + }) + .await +} + +fn with_registry( + meta_path: &Path, + f: impl FnOnce(&CobStore, CobId, &Registry) -> Result, +) -> Result, XrpcError> { + let meta = Repo::open(meta_path)?; + let store = CobStore::new(&meta); + let Some(object) = registry_object(&store)? else { + return Ok(None); + }; + let (registry, _) = store.materialize::(object)?; + f(&store, object, ®istry).map(Some) +} + +fn registry_object(store: &CobStore) -> Result, XrpcError> { + match store + .list::() + .map_err(XrpcError::from)? + .as_slice() + { + [] => Ok(None), + [object] => Ok(Some(*object)), + many => Err(XrpcError::internal(format!( + "{} repo registry objects share meta-repo", + many.len() + ))), + } +} + +pub(crate) fn origin_url(base: &Url, path: &str) -> Result { + let mut url = base.clone(); + url.path_segments_mut() + .map_err(|_| XrpcError::bad_gateway("the origin knot's endpoint can't take a path"))? + .extend(path.split('/')); + Ok(url) +} + fn hosted( state: &XrpcState, owner: &OwnerDid, @@ -428,9 +721,12 @@ pub(crate) struct SeededVersion { pub(crate) produces: RecordCid, } +pub(crate) fn seed_type_name() -> knot_types::TypeName { + knot_types::TypeName::new(Seeded::TYPE).expect("the seed type is a valid type name") +} + pub(crate) fn seeded_versions(store: &CobStore) -> Result, XrpcError> { - let seed_type = - knot_types::TypeName::new(Seeded::TYPE).expect("the seed type is a valid type name"); + let seed_type = seed_type_name(); fostered_objects(store)? .into_iter() .map(|(address, object)| { @@ -532,3 +828,98 @@ fn seed_one( })?; Ok(()) } + +pub async fn run_retire_gc( + state: Arc>, + shutdown: tokio_util::sync::CancellationToken, +) { + let mut tick = tokio::time::interval(std::time::Duration::from_secs(5)); + tick.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Skip); + let _first = tick.tick().await; + loop { + tokio::select! { + () = shutdown.cancelled() => return, + _ = tick.tick() => gc_tick(&state).await, + } + } +} + +pub async fn gc_tick(state: &Arc>) { + let now = state.now(); + const RETIRE_GRACE_SECS: u64 = 3_600; + let meta_path = state.meta_path.clone(); + let candidates = run_blocking(move || { + with_registry(&meta_path, |_, _, registry| { + Ok(registry + .records() + .filter_map(|(repo, record)| { + record.retired_at.and_then(|at| { + u64::try_from(now.get().saturating_sub(at.get())) + .is_ok_and(|age| age >= RETIRE_GRACE_SECS) + .then(|| repo.clone()) + }) + }) + .collect::>()) + }) + }) + .await + .unwrap_or_else(|error: XrpcError| { + tracing::error!(%error, "retire GC couldn't read the registry"); + None + }) + .unwrap_or_default(); + for repo in candidates { + if let Err(error) = gc_repo(state, &repo).await { + tracing::error!(repo = %repo, %error, "retire GC couldn't remove the shadow"); + } + } +} + +async fn gc_repo( + state: &Arc>, + repo: &RepoDid, +) -> Result<(), XrpcError> { + let signer = state.secrets.signer(&state.knot_did)?; + let layout = state.layout.clone(); + let index = Arc::clone(&state.index); + let cob_locks = Arc::clone(&state.cob_locks); + let meta_path = state.meta_path.clone(); + let lfs_store = state.lfs.as_ref().map(|lfs| Arc::clone(&lfs.handle.store)); + let deleted = repo.clone(); + let home = CobHome::from(&state.knot_did); + let now = state.now(); + run_blocking(move || { + let _repo_guard = cob_locks.repo(&deleted); + let _meta_guard = cob_locks.meta(); + with_registry(&meta_path, |store, object, registry| { + let record = registry + .record_of(&deleted) + .ok_or_else(|| XrpcError::not_found("the repo isn't registered on this knot"))?; + let target = RepoRef { + owner: record.owner.clone(), + rkey: record.rkey.clone(), + }; + layout + .remove(&deleted) + .map_err(|error| XrpcError::internal(format!("repo {deleted} couldn't be removed from disk, so it stays registered for the next sweep: {error}")))?; + deregister_repo(store, &home, object, target, deleted.clone(), &signer, now)?; + index.refresh_registry().map_err(|error| { + XrpcError::internal(format!( + "repo {deleted} was deregistered but registry projection refresh failed: {error}" + )) + })?; + if let Some(lfs) = &lfs_store + && let Err(error) = lfs.remove_repo(&deleted) + { + tracing::warn!( + repo = %deleted, + %error, + "lfs prefix removal on retire GC failed, orphan sweep will reclaim" + ); + } + Ok(()) + })? + .ok_or_else(|| XrpcError::not_found("no git repos registered on this knot")) + }) + .await +} diff --git a/knot2/crates/knot-xrpc/src/handover.rs b/knot2/crates/knot-xrpc/src/handover.rs index 2976ffcd3..d6a791f03 100644 --- a/knot2/crates/knot-xrpc/src/handover.rs +++ b/knot2/crates/knot-xrpc/src/handover.rs @@ -34,7 +34,6 @@ pub(crate) const SIDECAR: &str = "tangled.handover.json"; pub(crate) const SETTLED: &str = "tangled.handover.done"; const COBS_PREFIX: &str = "refs/cobs/"; -const ACK_CONTEXT: &str = "org.tangled.temp.handover.ack.v1"; const BOOT_WIDTH: usize = 4; const FOSTER_DOWN_AFTER: usize = 3; @@ -89,9 +88,7 @@ pub struct FosterKnot(Url); impl FosterKnot { pub fn new(raw: &str) -> Result { let url = Url::parse(raw).map_err(|_| format!("foster url {raw} isn't a url"))?; - let trusted = matches!(url.scheme(), "https") - || matches!(url.scheme(), "http") && host_is_loopback(&url); - if !trusted { + if !url_is_trusted(&url) { return Err(format!( "foster url {raw} must be https, or http on loopback. a foster knot is \ trusted over tls or not at all" @@ -109,6 +106,10 @@ impl FosterKnot { } } +pub(crate) fn url_is_trusted(url: &Url) -> bool { + matches!(url.scheme(), "https") || matches!(url.scheme(), "http") && host_is_loopback(url) +} + fn host_is_loopback(url: &Url) -> bool { match url.host() { Some(url::Host::Domain(name)) => name.eq_ignore_ascii_case("localhost"), @@ -250,15 +251,15 @@ async fn drive( state: Arc>, repo: RepoDid, lease: HandoverLease, -) -> Result { +) -> Result { tokio::select! { () = lease.token.cancelled() => { tracing::info!(repo = %repo, "the handover was canceled. its job is abandoned"); - Ok(0) + Ok(crate::foster::RecordCount::new(0)) } outcome = finish(&state, &repo) => { match &outcome { - Ok(records) => tracing::info!(repo = %repo, records = *records, "the handover finished. every fostered record of this repo is home"), + Ok(records) => tracing::info!(repo = %repo, records = records.get(), "the handover finished. every fostered record of this repo is home"), Err(error) => tracing::error!(repo = %repo, %error, "the handover didn't complete. its job stays for a replay"), }; outcome @@ -269,7 +270,7 @@ async fn drive( pub(crate) async fn finish( state: &Arc>, repo: &RepoDid, -) -> Result { +) -> Result { let repo_path = state .layout .repo_path(repo) @@ -315,7 +316,7 @@ async fn handover( state: &Arc>, repo: &RepoDid, foster: &FosterKnot, -) -> Result { +) -> Result { let url = foster.url(repo.as_str()); let (object_format, advertised) = match advertised_cobs(state, &url).await? { Shadow::Absent => { @@ -323,7 +324,7 @@ async fn handover( repo = %repo, "the foster knot doesn't shadow this repo, so its foster era is settled" ); - return Ok(0); + return Ok(crate::foster::RecordCount::new(0)); } Shadow::Hosted(object_format, advertised) => (object_format, advertised), }; @@ -391,15 +392,48 @@ async fn handover( } apply_cob_refs(&git, &graphs, &placed)?; let signer = settled.secrets.signer(&settled.knot_did)?; + let converted = crate::social::convert_seeded(&git, &target, &signer)?; crate::social::reconcile(&settled, &git, &target, &signer, settled.now())?; - let seeded = crate::foster::seeded_versions(&store)?; - confirm_local_leaves(&git, &seeded)?; - Ok(seeded) + confirm_local_leaves(&git, &converted)?; + crate::foster::seeded_versions(&store) }) .await?; confirm_foster_serving(state, foster, repo, &seeded).await?; - send_ack(state, foster, repo, seeded.len(), state.now()).await?; - Ok(seeded.len()) + prune_seeded(state, repo).await?; + let records = crate::foster::RecordCount::new(seeded.len()); + send_ack(state, foster, repo, records, state.now()).await?; + Ok(records) +} + +async fn prune_seeded( + state: &Arc>, + repo: &RepoDid, +) -> Result<(), XrpcError> { + let layout = state.layout.clone(); + let target = repo.clone(); + let settled = Arc::clone(state); + run_blocking(move || { + let _guard = settled.cob_locks.repo(&target); + let git = layout.open(&target)?; + let store = CobStore::new(&git); + let seed_type = crate::foster::seed_type_name(); + let updates = crate::foster::fostered_objects(&store)? + .into_values() + .filter_map(|object| seeded_ref_delete(&git, &seed_type, object).transpose()) + .collect::, XrpcError>>()?; + git.with_ref_txn(|txn| txn.update_refs(&updates)) + .map_err(XrpcError::from) + }) + .await +} + +fn seeded_ref_delete( + git: &Repo, + seed_type: &TypeName, + object: CobId, +) -> Result, XrpcError> { + let name = RefName::new(cob_ref(seed_type, object)).expect("a served cob ref name parses"); + Ok(git.find_ref(&name)?.map(|old| RefUpdate::Delete { name, old })) } fn cob_ref(type_name: &TypeName, object: CobId) -> String { @@ -480,10 +514,13 @@ fn apply_cob_refs( fn confirm_local_leaves( git: &Repo, - seeded: &[crate::foster::SeededVersion], + converted: &[crate::social::Converted], ) -> Result<(), XrpcError> { let leaves = futures::executor::block_on(chain::leaves(git))?; - seeded.iter().try_for_each(|version| { + converted.iter().try_for_each(|crate::social::Converted { version, settled }| { + if matches!(settled, knot_cobs::Settled::Superseded) { + return Ok(()); + } (leaves.get(&version.address.mst_key()) == Some(&version.produces.cid())) .then_some(()) .ok_or_else(|| { @@ -497,52 +534,55 @@ fn confirm_local_leaves( }) } +pub(crate) fn advertises(described: &serde_json::Value, capability: &Capability) -> bool { + described["capabilities"].as_array().is_some_and(|capabilities| { + capabilities.iter().any(|token| token.as_str() == Some(capability.token())) + }) +} + +pub(crate) async fn knot_identity( + state: &XrpcState, + base: &Url, + who: &str, + capability: &Capability, + lacking: &str, +) -> Result { + let describe_url = crate::foster::origin_url(base, "xrpc/org.tangled.knot.describeKnot")?; + let described = get_json(state.git_http.as_ref(), describe_url, who).await?; + let did = described["did"] + .as_str() + .ok_or_else(|| XrpcError::bad_gateway("describeKnot leaves out the knot's own did"))? + .to_owned(); + if !advertises(&described, capability) { + return Err(XrpcError::invalid_request(format!("{did} {lacking}"))); + } + let document_url = crate::foster::origin_url(base, ".well-known/did.json")?; + let document = get_json(state.git_http.as_ref(), document_url, who).await?; + if document["id"].as_str() != Some(did.as_str()) { + return Err(XrpcError::bad_gateway(format!( + "the did document served by {base} belongs to a different knot than describeKnot reported" + ))); + } + let wanted = format!("{did}#{}", knot_atproto::ATPROTO_METHOD); + let multibase = document["verificationMethod"].as_array().and_then(|methods| { + methods.iter().find(|method| method["id"].as_str() == Some(wanted.as_str())) + }) + .and_then(|method| method["publicKeyMultibase"].as_str()) + .ok_or_else(|| { + XrpcError::bad_gateway(format!("{did} publishes no multikey under #atproto")) + })? + .to_owned(); + Ok(multibase) +} + async fn foster_key( state: &Arc>, foster: &FosterKnot, ) -> Result { - let described = get_json( - state.git_http.as_ref(), - foster.url("xrpc/org.tangled.knot.describeKnot"), + let multibase = knot_identity( + state, &foster.0, "foster knot", &Capability::Foster, "doesn't take fostered repos", ) .await?; - let did = described["did"].as_str().ok_or_else(|| { - XrpcError::bad_gateway("describeKnot on the foster knot leaves out its own did") - })?; - let fostering = described["capabilities"] - .as_array() - .is_some_and(|capabilities| { - capabilities - .iter() - .any(|token| token.as_str() == Some(Capability::Foster.token())) - }); - if !fostering { - return Err(XrpcError::invalid_request(format!( - "{did} doesn't take fostered repos" - ))); - } - let document = get_json(state.git_http.as_ref(), foster.url(".well-known/did.json")).await?; - if document["id"].as_str() != Some(did) { - return Err(XrpcError::bad_gateway(format!( - "the did document served by {} belongs to a different knot than the one that \ - served describeKnot", - foster.0.as_str() - ))); - } - let wanted = format!("{did}#{}", knot_atproto::ATPROTO_METHOD); - let multibase = document["verificationMethod"] - .as_array() - .and_then(|methods| { - methods - .iter() - .find(|method| method["id"].as_str() == Some(wanted.as_str())) - }) - .and_then(|method| method["publicKeyMultibase"].as_str()) - .ok_or_else(|| { - XrpcError::bad_gateway( - "the foster knot's did document publishes no multikey under #atproto", - ) - })?; ActorId::new(multibase) .map_err(|_| XrpcError::bad_gateway("the foster knot's multikey doesn't parse as a key")) } @@ -559,7 +599,7 @@ async fn confirm_foster_serving( .append_pair("repo", repo.as_str()) .append_pair("collection", version.address.collection().as_str()) .append_pair("rkey", version.address.rkey().as_str()); - let served = get_json(state.git_http.as_ref(), url).await?; + let served = get_json(state.git_http.as_ref(), url, "foster knot").await?; let cid = served["cid"].as_str().ok_or_else(|| { XrpcError::bad_gateway(format!( "the foster knot served getRecord for {} without a cid", @@ -568,7 +608,7 @@ async fn confirm_foster_serving( })?; if cid != version.produces.cid().to_string() { return Err(XrpcError::internal(format!( - "the foster knot serves {} at {cid} while its own graph recorded {}", + "the foster knot serves {} at {cid} while the knot's graph recorded {}", version.address.at_uri(repo), version.produces.cid() ))); @@ -577,20 +617,10 @@ async fn confirm_foster_serving( Ok(()) } -#[derive(Serialize)] -struct AckSigned<'a> { - context: &'static str, - #[serde(rename = "repoDid")] - repo_did: &'a RepoDid, - records: usize, - #[serde(rename = "ackedAt")] - acked_at: UnixSeconds, -} - #[derive(Serialize)] struct AckBody<'a> { #[serde(flatten)] - signed: AckSigned<'a>, + signed: crate::foster::AckSigned<'a>, signature: String, } @@ -598,12 +628,12 @@ async fn send_ack( state: &Arc>, foster: &FosterKnot, repo: &RepoDid, - records: usize, + records: crate::foster::RecordCount, now: UnixSeconds, ) -> Result<(), XrpcError> { let signer = state.secrets.signer(&state.knot_did)?; - let signed = AckSigned { - context: ACK_CONTEXT, + let signed = crate::foster::AckSigned { + context: crate::foster::ACK_CONTEXT, repo_did: repo, records, acked_at: now, @@ -637,20 +667,22 @@ async fn send_ack( ))) } -async fn get_json(http: &dyn HttpTransport, url: Url) -> Result { +pub(crate) async fn get_json( + http: &dyn HttpTransport, + url: Url, + who: &str, +) -> Result { let response = http.execute(HttpRequest::get(url)).await.map_err(|error| { - XrpcError::bad_gateway(format!("the foster knot is unreachable: {error}")) + XrpcError::bad_gateway(format!("the {who} is unreachable: {error}")) })?; if !response.status.is_success() { return Err(XrpcError::bad_gateway(format!( - "the foster knot returned {} where a document belonged", + "the {who} returned {} instead of the did document", response.status.as_u16() ))); } serde_json::from_slice(&response.body).map_err(|error| { - XrpcError::bad_gateway(format!( - "the foster knot returned bytes that aren't json: {error}" - )) + XrpcError::bad_gateway(format!("the {who} returned non-json bytes: {error}")) }) } diff --git a/knot2/crates/knot-xrpc/src/repos.rs b/knot2/crates/knot-xrpc/src/repos.rs index dedbd4785..72114514e 100644 --- a/knot2/crates/knot-xrpc/src/repos.rs +++ b/knot2/crates/knot-xrpc/src/repos.rs @@ -288,6 +288,7 @@ pub(crate) async fn seed_repo( created_at: now, policy, fostered: matches!(seeding, Seeding::Fostered { .. }), + retired_at: None, }; let lfs_store = state.lfs.as_ref().map(|web| Arc::clone(&web.handle.store)); diff --git a/knot2/crates/knot-xrpc/src/social.rs b/knot2/crates/knot-xrpc/src/social.rs index 7ff52d6a7..35abf3f24 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, Decision, MstKey, Oid, OwnerDid, RecordAddress, RecordBody, RecordCid, RecordCollection, - RecordRkey, RepoDid, RepoName, RepoRkey, ServiceDid, SubjectNumber, Tid, - TypeName, UnixSeconds, + RecordRkey, RepoDid, RepoName, RepoRkey, ServiceDid, SubjectNumber, Tid, TypeName, + UnixSeconds, }; use serde::de::DeserializeOwned; use serde::{Deserialize, Serialize}; @@ -940,6 +940,73 @@ impl chain::Records for Kept { &self.due } } + +#[derive(Deserialize)] +struct SeededView { + author: AccountDid, + created_at: UnixSeconds, +} + +pub(crate) struct Converted { + pub(crate) version: crate::foster::SeededVersion, + pub(crate) settled: knot_cobs::Settled, +} + +pub(crate) fn convert_seeded( + git: &Repo, + repo: &RepoDid, + signer: &dyn Signer, +) -> Result, XrpcError> { + let store = CobStore::new(git); + let home = CobHome::from(repo); + let seeded = crate::foster::seeded_versions(&store)?; + let seed_type = crate::foster::seed_type_name(); + let mut settled = Vec::with_capacity(seeded.len()); + let mut numbers: BTreeMap = BTreeMap::new(); + for version in seeded { + let changes = + store.changes_of(&seed_type, version.object, knot_cob::HistoryModel::Convergent, None)?; + let tip = changes + .changes + .last() + .ok_or_else(|| XrpcError::internal("a seeded graph has no changes"))?; + let SeededView { author, created_at } = serde_ipld_dagcbor::from_slice(tip.payload()) + .map_err(|error| XrpcError::internal(error.to_string()))?; + let body = tree_body(&store, &seed_type, version.object, version.produces)? + .ok_or_else(|| XrpcError::internal("a seeded record has no body"))?; + let number = numbers + .get(version.address.collection()) + .copied() + .unwrap_or(SubjectNumber::FIRST); + numbers.insert( + version.address.collection().clone(), + number + .after() + .ok_or_else(|| XrpcError::internal("a collection has numbered past what fits"))?, + ); + let collection = version.address.collection().as_str().to_owned(); + let creation = Creation { + address: version.address.clone(), + author, + number, + parent: None, + created_at, + produces: version.produces, + }; + let outcome = knot_cobs::settle_stored(&store, &home, repo, signer, &collection, creation, body)? + .ok_or_else(|| { + XrpcError::bad_gateway(format!( + "the foster knot serves {collection} under an unknown history model" + )) + })?; + settled.push(Converted { + version, + settled: outcome, + }); + } + Ok(settled) +} + pub(crate) fn reconcile( state: &Arc>, git: &Repo, @@ -954,13 +1021,13 @@ pub(crate) fn reconcile( .filter_map(|entry| entry.state.serving().map(|version| (entry, version))) .collect(); let store = CobStore::new(git); - let standing = futures::executor::block_on(chain::leaves(git))?; + let leaves = futures::executor::block_on(chain::leaves(git))?; let mut collections: Vec = knot_cobs::social_collections().to_vec(); let mut records = serving .iter() .try_fold(Kept::default(), |kept, (entry, version)| { let key = entry.address.mst_key(); - let leaf = standing.get(&key).copied(); + let leaf = leaves.get(&key).copied(); let due = resolve_due( &store, &collection_type(entry.address.collection()), @@ -974,12 +1041,14 @@ pub(crate) fn reconcile( None => kept, }) })?; - for (key, collection, due) in fostered_versions(&store, &standing)? { - if !collections.contains(&collection) { - collections.push(collection); - } - if !records.due.contains_key(&key) { - records = records.taking(key, due); + if matches!(state.index.is_fostered(repo), Resolved::Ready(true)) { + for (key, collection, due) in fostered_versions(&store, &leaves)? { + if !collections.contains(&collection) { + collections.push(collection); + } + if !records.due.contains_key(&key) { + records = records.taking(key, due); + } } } if chain::tip(git)?.is_none() && chain::Records::due(&records).is_empty() { diff --git a/knot2/crates/knot-xrpc/tests/atproto_interop.rs b/knot2/crates/knot-xrpc/tests/atproto_interop.rs index 0a7607d54..7e1ecfd64 100644 --- a/knot2/crates/knot-xrpc/tests/atproto_interop.rs +++ b/knot2/crates/knot-xrpc/tests/atproto_interop.rs @@ -205,8 +205,11 @@ async fn foster_functionality_doesnt_activate_if_no_secret() { let world = World::new(); let credentials = common::basic_auth("admin", "denied-anyway"); let seed = serde_json::json!({ - "name": "boltfulboy", - "ownerDid": FOSTER_OWNER, + "repoDid": FOSTER_REPO_DID, + "manifest": { + "$type": "org.tangled.repo.manifest", + "declaration": { "owner": FOSTER_OWNER, "name": "boltfulboy" }, + }, "records": [], }); let (status, body) = common::post_raw( @@ -282,9 +285,11 @@ async fn a_fostering_knot_says_so_in_its_capabilities() { fn foster_seed_body(repo_did: &str) -> serde_json::Value { serde_json::json!({ - "name": "boltfulboy", - "ownerDid": FOSTER_OWNER, "repoDid": repo_did, + "manifest": { + "$type": "org.tangled.repo.manifest", + "declaration": { "owner": FOSTER_OWNER, "name": "boltfulboy" }, + }, "records": [{ "collection": "org.tangled.track.ticket", "rkey": FOSTER_RKEY, @@ -356,10 +361,11 @@ async fn a_fostered_record_survives_a_restart() { let world = World::fostering(FOSTER_SECRET); let credentials = common::basic_auth("admin", FOSTER_SECRET); let seed = serde_json::json!({ - "name": "boltfulboy", - "ownerDid": FOSTER_OWNER, "repoDid": FOSTER_REPO_DID, - "defaultBranch": "main", + "manifest": { + "$type": "org.tangled.repo.manifest", + "declaration": { "owner": FOSTER_OWNER, "name": "boltfulboy" }, + }, "records": [{ "collection": "org.tangled.track.ticket", "rkey": FOSTER_RKEY, diff --git a/knot2/crates/knot-xrpc/tests/common/mod.rs b/knot2/crates/knot-xrpc/tests/common/mod.rs index 9300d374a..36a0f70d1 100644 --- a/knot2/crates/knot-xrpc/tests/common/mod.rs +++ b/knot2/crates/knot-xrpc/tests/common/mod.rs @@ -37,7 +37,7 @@ pub async fn serve_local_plc(world: &World) -> std::net::SocketAddr { let endpoint = endpoint.clone(); async move { let did = Did::new_owned(did).expect("the plc request path is a DID"); - let doc = did_doc(&did, &multikey, &endpoint); + let doc = did_doc(&did, &multikey, &endpoint, None); let mut headers = axum::http::HeaderMap::new(); headers.insert( axum::http::header::CONTENT_TYPE, @@ -204,9 +204,9 @@ use knot_runtime::{ }; use knot_secrets::{MasterKey, SealedStore}; use knot_types::{ - AccountDid, ActorId, AtUri, Did, DidRkey, KnotHostname, KnotId, - KnotServiceUrl, Nsid, ObjectFormat, Oid, OwnerDid, RecordAddress, RecordCid, RecordCollection, - RecordRkey, RepoDid, RepoName, RepoRkey, ServiceDid, UnixSeconds, + AccountDid, ActorId, AtUri, Did, DidRkey, KnotHostname, KnotId, KnotServiceUrl, Nsid, + ObjectFormat, Oid, OwnerDid, RecordAddress, RecordCid, RecordCollection, RecordRkey, RepoDid, + RepoName, RepoRkey, ServiceDid, UnixSeconds, }; use knot_xrpc::{ ArchiveLimit, Budgets, ByteLimits, CobLocks, GlobalQuota, LimitConfig, PerActorQuota, @@ -414,8 +414,10 @@ pub struct World { pub layout: Layout, pub lfs_dir: std::path::PathBuf, pub router: Router, - pub state: Arc, ManualClock>>, + pub state: Arc, Arc>>, + clock: Arc, plc_pds: Arc>>, + plc_knots: Arc>>, } struct WorldSetup { @@ -446,6 +448,17 @@ impl World { .insert(did.clone(), pds.clone()); } + pub fn publish_plc_knot(&self, did: &str, knot: &str) { + self.plc_knots + .lock() + .unwrap() + .insert(Did::new_owned(did).unwrap(), knot.to_string()); + } + + pub fn advance(&self, delta: std::time::Duration) { + self.clock.advance(delta); + } + pub fn new() -> Self { Self::build(true, ByteLimits::default(), ObjectFormat::SHA1) } @@ -513,6 +526,18 @@ impl World { }) } + pub fn fostering_with_grace(secret: &str) -> Self { + let limits = knot_runtime::HttpLimits { + block_private_addresses: false, + ..knot_runtime::HttpLimits::default() + }; + Self::tailored(WorldSetup { + legacy_admin_secret: knot_xrpc::legacy_admin::LegacyAdminSecret::new(secret).ok(), + git_http: Some(Arc::new(knot_runtime::ReqwestHttp::new(limits).unwrap())), + ..WorldSetup::default() + }) + } + pub fn pulling(git_http: Arc) -> Self { Self::tailored(WorldSetup { admins: BTreeSet::from([owner()]), @@ -584,7 +609,9 @@ impl World { } let plc_pds = Arc::new(std::sync::Mutex::new(std::collections::HashMap::new())); + let plc_knots = Arc::new(std::sync::Mutex::new(std::collections::HashMap::new())); let pds_of = Arc::clone(&plc_pds); + let knots_of = Arc::clone(&plc_knots); let responder: Responder = Box::new(move |request| { let host = request.url.host_str().unwrap_or_default(); let did = match host { @@ -596,15 +623,14 @@ impl World { } .and_then(|did| Did::new_owned(did).ok()); let body = match did { - Some(did) => did_doc_at( - &did, - &pds_of - .lock() - .unwrap() - .get(&did) - .cloned() - .unwrap_or_else(default_pds), - ), + Some(did) => { + let knot = knots_of.lock().unwrap().get(&did).cloned(); + did_doc_at( + &did, + &pds_of.lock().unwrap().get(&did).cloned().unwrap_or_else(default_pds), + knot.as_deref(), + ) + } None => Bytes::new(), }; Ok(HttpResponse { @@ -613,9 +639,10 @@ impl World { body, }) }); + let clock = Arc::new(ManualClock::new(UnixMicros::new(1_000_000_000))); let atproto = Arc::new(Atproto::new( FakeHttp::new(responder), - ManualClock::new(UnixMicros::new(1_000_000_000)), + Arc::clone(&clock), knot.clone(), knot_atproto::PlcDirectory::new(url::Url::parse("https://plc.directory/").unwrap()) .unwrap(), @@ -648,7 +675,7 @@ impl World { let shutdown = tokio::sync::watch::channel(false); let fostering = legacy_admin_secret.is_some(); let firehose = Arc::new(knot_events::EventLog::new( - ManualClock::new(UnixMicros::new(1_000_000_000)), + Arc::clone(&clock), knot_events::ReplayBounds::new( knot_events::ReplayEvents::new(if fostering { 64 } else { 8 }).unwrap(), knot_events::ReplayBytes::new(16 << 20).unwrap(), @@ -745,7 +772,9 @@ impl World { lfs_dir: lfs_store, router, state, + clock, plc_pds, + plc_knots, } } @@ -762,6 +791,8 @@ impl World { created_at: UnixSeconds::new(1_000), policy: self.state.contribution_policy, fostered: false, + + retired_at: None, }; match store.list::().unwrap().as_slice() { [] => { @@ -929,7 +960,18 @@ fn actor_key() -> SigningKey { SigningKey::from_bytes(&[9u8; 32].into()).unwrap() } -fn did_doc(did: &Did, multikey: &ActorId, endpoint: &KnotServiceUrl) -> serde_json::Value { +fn did_doc( + did: &Did, + multikey: &ActorId, + endpoint: &KnotServiceUrl, + knot: Option<&str>, +) -> serde_json::Value { + let mut services = vec![serde_json::json!({"id": "#atproto_pds", + "type": "AtprotoPersonalDataServer", "serviceEndpoint": endpoint.as_str()})]; + if let Some(knot) = knot { + services.push(serde_json::json!({"id": "#tangled_knot", + "type": "TangledKnot", "serviceEndpoint": knot})); + } serde_json::json!({ "id": did.as_str(), "alsoKnownAs": [], @@ -939,22 +981,18 @@ fn did_doc(did: &Did, multikey: &ActorId, endpoint: &KnotServiceUrl) -> serde_js "controller": did.as_str(), "publicKeyMultibase": multikey.as_str() }], - "service": [{ - "id": "#atproto_pds", - "type": "AtprotoPersonalDataServer", - "serviceEndpoint": endpoint.as_str() - }] + "service": services }) } -fn did_doc_at(did: &Did, endpoint: &KnotServiceUrl) -> Bytes { +fn did_doc_at(did: &Did, endpoint: &KnotServiceUrl, knot: Option<&str>) -> Bytes { let sec1 = actor_key() .verifying_key() .to_encoded_point(true) .as_bytes() .to_vec(); let multikey = ActorId::new(knot_types::crypto::multikey(0xe7, &sec1)).unwrap(); - let doc = did_doc(did, &multikey, endpoint); + let doc = did_doc(did, &multikey, endpoint, knot); Bytes::from(serde_json::to_vec(&doc).unwrap()) } @@ -1040,6 +1078,7 @@ pub fn repo_did_document(world: &World, did: &RepoDid) -> serde_json::Value { &Did::new_owned(did.as_str()).unwrap(), &knot_multikey(world), &world.state.knot_service_url, + None, ) } diff --git a/knot2/crates/knot-xrpc/tests/handover.rs b/knot2/crates/knot-xrpc/tests/handover.rs index 22760ad36..457b98141 100644 --- a/knot2/crates/knot-xrpc/tests/handover.rs +++ b/knot2/crates/knot-xrpc/tests/handover.rs @@ -98,7 +98,7 @@ fn ticket_record(text: &str) -> serde_json::Value { "record": { "$type": COLLECTION, "title": {"$type": "org.tangled.markup.markdown#inline", "text": text}, - "state": "feral", + "state": "open", "createdAt": CREATED_AT, "x-tngl-editor": FOSTER_OWNER, }, @@ -124,10 +124,9 @@ async fn post_seed(world: &World, records: serde_json::Value) -> (StatusCode, se "/xrpc/org.tangled.temp.server.seedRepo", common::basic_auth("admin", FOSTER_SECRET), serde_json::json!({ - "name": "feralpearl", - "ownerDid": FOSTER_OWNER, "repoDid": REPO, - "defaultBranch": "main", + "manifest": {"$type": "org.tangled.repo.manifest", + "declaration": { "owner": FOSTER_OWNER, "name": "feralpearl" }}, "records": records, }), ) @@ -137,9 +136,11 @@ async fn post_seed(world: &World, records: serde_json::Value) -> (StatusCode, se } async fn seed_foster_record(world: &World) { + let origin_addr = serve_router(fake_origin_edge()).await; + world.publish_plc_knot(REPO, &format!("http://{origin_addr}")); let (status, body) = post_seed(world, serde_json::json!([ticket_record("a feral pearl")])).await; - assert_eq!(status, StatusCode::OK, "the seed must succeed: {body}"); + assert_eq!(status, StatusCode::OK, "seed failed: {body}"); } fn host_locally(world: &World) { @@ -158,13 +159,24 @@ fn local_knot() -> World { upgrade } +fn fake_origin_edge() -> axum::Router { + axum::Router::new().route( + "/xrpc/org.tangled.knot.describeKnot", + axum::routing::get(|| async { + axum::Json(serde_json::json!({"did": "did:web:origin.test", "capabilities": []})) + }), + ) +} + async fn handover_pair( records: serde_json::Value, refuse_first: bool, ) -> (World, std::net::SocketAddr, AckSink, World) { let foster = fostering(); + let origin_addr = serve_router(fake_origin_edge()).await; + foster.publish_plc_knot(REPO, &format!("http://{origin_addr}")); let (status, body) = post_seed(&foster, records).await; - assert_eq!(status, StatusCode::OK, "the seed must succeed: {body}"); + assert_eq!(status, StatusCode::OK, "seed failed: {body}"); let sink = AckSink { refuse_first, ..ok_sink() @@ -607,10 +619,12 @@ async fn a_cob_type_with_an_unknown_history_model_blocks_the_handover() { } #[tokio::test] -async fn an_empty_seed_is_refused() { +async fn empty_seed_still_fosters_manifest() { let foster = fostering(); let (status, body) = post_seed(&foster, serde_json::json!([])).await; - assert_eq!(status, StatusCode::BAD_REQUEST, "{body}"); + assert_eq!(status, StatusCode::OK, "{body}"); + assert_eq!(body["did"], REPO, "the shadow hosts the supplied repo DID"); + assert_eq!(body["results"], serde_json::json!([]), "no records, no results"); } #[tokio::test] @@ -664,22 +678,140 @@ async fn a_social_write_onto_a_shadow_repo_is_refused() { } #[tokio::test] -async fn a_record_outside_the_social_collections_comes_home_too() { - let (foster, foster_addr, sink, upgrade) = handover_pair( +async fn record_outside_social_collections_blocks_handover() { + let (_foster, foster_addr, _, upgrade) = handover_pair( serde_json::json!([ticket_record("a feral pearl"), markdown_record()]), false, ) .await; let outcome = trigger(&upgrade, foster_addr).await; + assert_eq!(outcome["state"], "running", "{outcome}"); + assert!(outcome["error"].as_str().unwrap().contains("history model"), + "the error reports the unknown history model: {outcome}"); +} + +async fn ticketed_pair(refuse_first: bool) -> (World, std::net::SocketAddr, AckSink, World) { + handover_pair(serde_json::json!([ticket_record("a feral pearl")]), refuse_first).await +} + +fn ticket_path() -> String { + format!("/xrpc/com.atproto.repo.getRecord?repo={REPO}&collection={COLLECTION}&rkey={RKEY}") +} + +async fn ticket_cid(world: &World) -> String { + common::get_json(world, &ticket_path()).await["cid"].as_str().unwrap().to_string() +} + +async fn delete_ticket(world: &World) { + let (status, body) = common::post_as( + world, + &nsid("com.atproto.repo.deleteRecord"), + &owner(), + &ServiceDid::new(REPO).unwrap(), + serde_json::json!({ + "repo": REPO, "rkey": RKEY, "collection": common::ticket_collection(), + "swapRecord": ticket_cid(world).await, + }), + ) + .await; + assert_eq!(status, StatusCode::OK, "deleteRecord failed: {body}"); +} + +async fn edit_ticket(world: &World) { + let (status, body) = common::post_as( + world, + &nsid("com.atproto.repo.putRecord"), + &owner(), + &ServiceDid::new(REPO).unwrap(), + serde_json::json!({ + "repo": REPO, "rkey": RKEY, "collection": common::ticket_collection(), + "swapRecord": ticket_cid(world).await, + "record": {"$type": COLLECTION, + "title": {"$type": "org.tangled.markup.markdown#inline", "text": "an edited pearl"}, + "state": "completed", "createdAt": CREATED_AT}, + }), + ) + .await; + assert_eq!(status, StatusCode::OK, "putRecord failed: {body}"); +} + +#[tokio::test] +async fn edit_to_pulled_ticket_succeeds() { + let (_foster, foster_addr, _, upgrade) = ticketed_pair(false).await; + let outcome = trigger(&upgrade, foster_addr).await; assert_eq!(outcome["state"], "done", "{outcome}"); - assert_eq!( - sink.bodies.lock()[0]["records"], - 2, - "both records come home" - ); - for (collection, rkey) in [(COLLECTION, RKEY), (MARKDOWN, NOTES_RKEY)] { - assert_cid_parity(&foster, &upgrade, collection, rkey).await; - } + + edit_ticket(&upgrade).await; +} + +#[tokio::test] +async fn deleted_pulled_ticket_stays_deleted_through_reconcile() { + let (_foster, foster_addr, _, upgrade) = ticketed_pair(false).await; + let outcome = trigger(&upgrade, foster_addr).await; + assert_eq!(outcome["state"], "done", "{outcome}"); + + delete_ticket(&upgrade).await; + + let state = std::sync::Arc::clone(&upgrade.state); + let surfaces = knot_xrpc::materialize::enable_surfaces; + let failed = tokio::task::spawn_blocking(move || surfaces(&state)).await.unwrap(); + assert_eq!(failed, 0, "the reconcile pass after a delete stays green"); + + let (status, error) = common::get_error(&upgrade, &ticket_path()).await; + assert_eq!(status, StatusCode::BAD_REQUEST, "{error}"); + assert_eq!(error, "RecordNotFound", "the deleted record stays deleted"); +} + +#[tokio::test] +async fn delete_during_crashed_handover_doesnt_wedge_replay() { + let (_foster, foster_addr, sink, upgrade) = ticketed_pair(true).await; + let first = trigger(&upgrade, foster_addr).await; + assert_eq!(first["state"], "running", "the rejected ack leaves it pending"); + + delete_ticket(&upgrade).await; + + let second = trigger(&upgrade, foster_addr).await; + assert_eq!(second["state"], "done", "the replay converges past the delete: {second}"); + assert_eq!(sink.len(), 2, "the ack goes out once the delete settles"); + assert_retired(&upgrade); + + let (status, error) = common::get_error(&upgrade, &ticket_path()).await; + assert_eq!(status, StatusCode::BAD_REQUEST, "{error}"); + assert_eq!(error, "RecordNotFound", "the origin's delete outlives the handover"); +} + +#[tokio::test] +async fn edit_during_crashed_handover_survives_replay() { + let (foster, foster_addr, sink, upgrade) = ticketed_pair(true).await; + let first = trigger(&upgrade, foster_addr).await; + assert_eq!(first["state"], "running", "the rejected ack leaves it pending"); + + edit_ticket(&upgrade).await; + + let second = trigger(&upgrade, foster_addr).await; + assert_eq!(second["state"], "done", "the replay goes green over the edit: {second}"); + assert_eq!(sink.len(), 2, "the ack goes out once the edit settles"); + assert_retired(&upgrade); + + let fetched = common::get_json(&upgrade, &ticket_path()).await; + assert_eq!(fetched["value"]["title"]["text"], "an edited pearl", + "the origin's edit outlives the handover"); + let shadow = common::get_json(&foster, &ticket_path()).await; + assert_ne!(fetched["cid"], shadow["cid"], + "the origin serves the edit, not the shadow's version"); +} + +#[tokio::test] +async fn finished_handover_drops_seed_refs() { + let (foster, foster_addr, _sink, upgrade) = ticketed_pair(false).await; + let outcome = trigger(&upgrade, foster_addr).await; + assert_eq!(outcome["state"], "done", "{outcome}"); + assert_cid_parity(&foster, &upgrade, COLLECTION, RKEY).await; + + let refs = local_cob_refs(&upgrade); + assert!(refs.keys().all(|name| !name.contains("temp.seed")), + "the transport refs go away once the handover is green: {refs:?}"); + assert!(!refs.is_empty(), "the repo's own manifest and ticket cobs stay behind"); } fn regress_foster_refs(world: &World, prior: &BTreeMap) { @@ -725,15 +857,11 @@ async fn a_regressed_foster_ref_cant_pull_the_local_ref_backwards() { regress_foster_refs(&foster, &after_first); let outcome = trigger(&upgrade, foster_addr).await; assert_eq!(outcome["state"], "running", "{outcome}"); - assert!( - outcome["error"].as_str().unwrap().contains("isn't ahead"), - "the error reports the ref wasn't moved backwards: {outcome}" - ); - assert_eq!( - local_cob_refs(&upgrade), - after_second, - "the local refs keep the newer tips" - ); + assert!(outcome["error"].as_str().unwrap().contains("while the knot's graph recorded"), + "the green check catches the regressed foster ref: {outcome}"); + assert!(after_second.iter().filter(|(name, _)| !name.contains("temp.seed")) + .all(|(name, target)| local_cob_refs(&upgrade).get(name) == Some(target)), + "the local refs keep the newer tips"); } fn hanging_ack_edge() -> axum::Router { diff --git a/knot2/crates/knot-xrpc/tests/projection.rs b/knot2/crates/knot-xrpc/tests/projection.rs index 650055109..cdb24bd4e 100644 --- a/knot2/crates/knot-xrpc/tests/projection.rs +++ b/knot2/crates/knot-xrpc/tests/projection.rs @@ -146,7 +146,11 @@ async fn the_full_lifecycle_keeps_the_mst_projection_exact() { }), ) .await; - assert_eq!(status, http::StatusCode::OK, "completing failed: {completed}"); + assert_eq!( + status, + http::StatusCode::OK, + "completing failed: {completed}" + ); assert_projection(&world, &did).await; let (status, deleted) = post_authed( diff --git a/knot2/crates/knot-xrpc/tests/retire.rs b/knot2/crates/knot-xrpc/tests/retire.rs new file mode 100644 index 000000000..6dd19dc14 --- /dev/null +++ b/knot2/crates/knot-xrpc/tests/retire.rs @@ -0,0 +1,242 @@ +mod common; + +use base64::{engine::general_purpose::URL_SAFE_NO_PAD, Engine as _}; +use common::{World, basic_auth, get_json, serve_router}; +use http::StatusCode; +use knot_cob::CobStore; +use knot_cobs::RepoRegistryCob; +use knot_runtime::{K256Signer, OsEntropy, Signer as _}; +use knot_types::{KnotId, KnotServiceUrl, RepoDid, UnixSeconds}; + +const FOSTER_SECRET: &str = "foster-pearl"; +const FOSTER_OWNER: &str = "did:plc:boltfulboy"; +const REPO: &str = "did:plc:feralpearl"; +const COLLECTION: &str = "org.tangled.track.ticket"; +const RKEY: &str = "the-feral-pearl-ledger"; +const ACK_PATH: &str = "/xrpc/org.tangled.temp.server.ackHandover"; +const ORIGIN_DID: &str = "did:web:origin.test"; +const ACK_CONTEXT: &str = "org.tangled.temp.handover.ack.v1"; + +fn repo_did() -> RepoDid { + RepoDid::new(REPO).unwrap() +} + +fn ticket_record(text: &str) -> serde_json::Value { + serde_json::json!({"collection": COLLECTION, "rkey": RKEY, "author": FOSTER_OWNER, "createdAt": 1100, + "record": {"$type": COLLECTION, + "title": {"$type": "org.tangled.markup.markdown#inline", "text": text}, + "state": "feral", "createdAt": "1970-01-01T00:18:20Z", "x-tngl-editor": FOSTER_OWNER}}) +} + +async fn post_seed(world: &World, records: serde_json::Value) -> (StatusCode, serde_json::Value) { + let (status, body) = common::post_raw( + world, + "/xrpc/org.tangled.temp.server.seedRepo", + basic_auth("admin", FOSTER_SECRET), + serde_json::json!({ + "repoDid": REPO, "records": records, + "manifest": {"$type": "org.tangled.repo.manifest", + "declaration": { "owner": FOSTER_OWNER, "name": "feralpearl" }}, + }), + ) + .await; + let parsed = serde_json::from_slice(&body).unwrap_or(serde_json::Value::Null); + (status, parsed) +} + +async fn seed_foster_record(world: &World) { + let (status, body) = + post_seed(world, serde_json::json!([ticket_record("a feral pearl")])).await; + assert_eq!(status, StatusCode::OK, "seed failed: {body}"); +} + +fn json_get(payload: serde_json::Value) -> axum::routing::MethodRouter { + axum::routing::get(move || { + let payload = payload.clone(); + async move { axum::Json(payload) } + }) +} + +fn knot_social_origin(signer: &K256Signer) -> axum::Router { + let document = knot_atproto::knot_did_document( + &KnotId::new(ORIGIN_DID).unwrap(), + &signer.public_key(), + &KnotServiceUrl::new("https://origin.test").unwrap(), + ); + axum::Router::new() + .route( + "/xrpc/org.tangled.knot.describeKnot", + json_get(serde_json::json!({"did": ORIGIN_DID, "capabilities": ["knot-social-writes"]})), + ) + .route("/.well-known/did.json", json_get(document)) +} + +async fn serve_origin(world: &World, signer: &K256Signer) -> String { + let path = format!("/xrpc/com.atproto.repo.getRecord?repo={REPO}&collection={COLLECTION}&rkey={RKEY}"); + let fetched = get_json(world, &path).await; + let cid = fetched["cid"].as_str().unwrap().to_string(); + let origin = knot_social_origin(signer).route("/xrpc/com.atproto.repo.getRecord", + json_get(serde_json::json!({"uri": format!("at://{REPO}/{COLLECTION}/{RKEY}"), + "cid": cid, "value": fetched["value"].clone()}))); + format!("http://{}", serve_router(origin).await) +} + +#[derive(serde::Serialize)] +struct AckSignedView<'a> { + context: &'a str, + #[serde(rename = "repoDid")] + repo_did: &'a str, + records: u64, + #[serde(rename = "ackedAt")] + acked_at: i64, +} + +fn signed_ack(signer: &K256Signer, acked_at: i64) -> serde_json::Value { + let view = AckSignedView { context: ACK_CONTEXT, repo_did: REPO, records: 1, acked_at }; + let bytes = serde_ipld_dagcbor::to_vec(&view).unwrap(); + let signature = URL_SAFE_NO_PAD.encode(signer.sign(&bytes).as_bytes()); + serde_json::json!({"context": ACK_CONTEXT, "repoDid": REPO, "records": 1, + "ackedAt": acked_at, "signature": signature}) +} + +async fn post_ack(world: &World, ack: serde_json::Value) -> (StatusCode, serde_json::Value) { + let (status, body) = common::post_raw(world, ACK_PATH, http::HeaderMap::new(), ack).await; + let parsed = serde_json::from_slice(&body).unwrap_or(serde_json::Value::Null); + (status, parsed) +} + +fn retired_at(world: &World) -> Option { + let meta = knot_git::Repo::open(&world.state.meta_path).unwrap(); + let store = CobStore::new(&meta); + let object = store.list::().unwrap()[0]; + let (registry, _) = store.materialize::(object).unwrap(); + registry.record_of(&repo_did()).and_then(|record| record.retired_at.map(UnixSeconds::get)) +} + +async fn seeded_origin() -> (World, K256Signer, String) { + let foster = World::fostering_with_grace(FOSTER_SECRET); + seed_foster_record(&foster).await; + let signer = K256Signer::generate(&OsEntropy); + let origin_url = serve_origin(&foster, &signer).await; + foster.publish_plc_knot(REPO, &origin_url); + (foster, signer, origin_url) +} + +#[tokio::test] +async fn valid_ack_retires_shadow() { + let (foster, signer, _) = seeded_origin().await; + + let (status, body) = post_ack(&foster, signed_ack(&signer, 1100)).await; + assert_eq!(status, StatusCode::OK, "{body}"); + assert_eq!(retired_at(&foster), Some(1100), "retiredAt follows the ack's ackedAt"); +} + +#[tokio::test] +async fn retired_shadow_returns_409_to_further_seeds() { + let (foster, signer, _) = seeded_origin().await; + let (status, body) = post_ack(&foster, signed_ack(&signer, 1100)).await; + assert_eq!(status, StatusCode::OK, "{body}"); + + let (status, body) = post_seed(&foster, serde_json::json!([ticket_record("a second pearl")])).await; + assert_eq!(status, StatusCode::CONFLICT, "{body}"); +} + +#[tokio::test] +async fn forged_ack_rejected() { + let (foster, _, _) = seeded_origin().await; + + let stranger = K256Signer::generate(&OsEntropy); + let (status, _) = post_ack(&foster, signed_ack(&stranger, 1100)).await; + assert_eq!(status, StatusCode::FORBIDDEN, "the ack's signer isn't the foster key"); + assert_eq!(retired_at(&foster), None, "a rejected ack doesn't retire the repo"); +} + +#[tokio::test] +async fn plain_http_origin_off_loopback_rejected() { + let (foster, signer, origin_url) = seeded_origin().await; + let off_loopback = origin_url.replace("127.0.0.1", "foster.nel.pet"); + foster.publish_plc_knot(REPO, &off_loopback); + + let (status, body) = post_ack(&foster, signed_ack(&signer, 1100)).await; + assert_eq!(status, StatusCode::BAD_REQUEST, "{body}"); + assert_eq!(retired_at(&foster), None, "no retire over plain http off loopback"); +} + +#[tokio::test] +async fn data_url_origin_rejected() { + let foster = World::fostering_with_grace(FOSTER_SECRET); + foster.publish_plc_knot(REPO, "data:text/plain,a-knot"); + seed_foster_record(&foster).await; + let signer = K256Signer::generate(&OsEntropy); + + let (status, body) = post_ack(&foster, signed_ack(&signer, 1100)).await; + assert_eq!(status, StatusCode::BAD_REQUEST, "{body}"); + assert_eq!(retired_at(&foster), None, "a data url origin doesn't retire the shadow or panic the knot"); +} + +#[tokio::test] +async fn pre_epoch_ack_rejected() { + let (foster, signer, _) = seeded_origin().await; + + let (status, body) = post_ack(&foster, signed_ack(&signer, -1)).await; + assert_eq!(status, StatusCode::BAD_REQUEST, "{body}"); + assert_eq!(retired_at(&foster), None, "a pre-epoch ackedAt can't retire anything"); +} + +#[tokio::test] +async fn old_ack_after_newer_is_no_op() { + let (foster, signer, _) = seeded_origin().await; + + let (status, body) = post_ack(&foster, signed_ack(&signer, 20)).await; + assert_eq!(status, StatusCode::OK, "the newer ack retires first: {body}"); + + let (status, _) = post_ack(&foster, signed_ack(&signer, 10)).await; + assert_eq!(status, StatusCode::OK, "an old ack is a no-op, not an error"); + assert_eq!(retired_at(&foster), Some(20), "an old ack can't retire a fresh shadow"); +} + +#[tokio::test] +async fn replay_after_retire_returns_200() { + let (foster, signer, _) = seeded_origin().await; + + let ack = signed_ack(&signer, 1100); + let (status, body) = post_ack(&foster, ack.clone()).await; + assert_eq!(status, StatusCode::OK, "the first ack retires the shadow: {body}"); + + let (status, _) = post_ack(&foster, ack).await; + assert_eq!(status, StatusCode::OK, "a replay after retire is a 200, not an error"); + assert_eq!(retired_at(&foster), Some(1100), "the replay doesn't move the retired state"); +} + +#[tokio::test] +async fn seed_from_knot_social_origin_gets_409() { + let foster = World::fostering_with_grace(FOSTER_SECRET); + let signer = K256Signer::generate(&OsEntropy); + let origin = serve_router(knot_social_origin(&signer)).await; + foster.publish_plc_knot(REPO, &format!("http://{origin}")); + + let (status, body) = post_seed(&foster, serde_json::json!([ticket_record("a feral pearl")])).await; + assert_eq!(status, StatusCode::CONFLICT, + "the origin already serves COBs: {body}"); +} + +#[tokio::test] +async fn gc_takes_retired_repo_past_grace_window() { + let (foster, signer, _) = seeded_origin().await; + + let (status, body) = post_ack(&foster, signed_ack(&signer, 0)).await; + assert_eq!(status, StatusCode::OK, "retire before the grace window: {body}"); + + foster.advance(std::time::Duration::from_secs(3_601)); + knot_xrpc::foster::gc_tick(&foster.state).await; + + let meta = knot_git::Repo::open(&foster.state.meta_path).unwrap(); + let store = CobStore::new(&meta); + let object = store.list::().unwrap()[0]; + let (registry, _) = store.materialize::(object).unwrap(); + assert!(registry.record_of(&repo_did()).is_none(), "the retired shadow is deregistered"); + assert!(!foster.layout.repo_path(&repo_did()).unwrap().exists(), "the on-disk shadow is removed"); + let tombstones = knot_xrpc::plc::deleted_dids(&meta).unwrap(); + assert!(!tombstones.contains(&repo_did()), + "the repo is alive on the origin, so the retire GC writes no PLC tombstone"); +} diff --git a/knot2/interop/foster_test.go b/knot2/interop/foster_test.go index 1cd285a16..c5daa73f8 100644 --- a/knot2/interop/foster_test.go +++ b/knot2/interop/foster_test.go @@ -64,10 +64,11 @@ func seedRequest(t *testing.T) (map[string]any, syntax.RecordKey) { "KNOT_TEST_FOSTER_REPO_DID is the foreign repo DID the knot fosters", ) return map[string]any{ - "name": "boltfulboy", - "ownerDid": fosterOwner, - "repoDid": fosterRepoDid, - "defaultBranch": "main", + "repoDid": fosterRepoDid, + "manifest": map[string]any{ + "$type": "org.tangled.repo.manifest", + "declaration": map[string]any{"owner": fosterOwner, "name": "boltfulboy"}, + }, "records": []map[string]any{{ "collection": ticketCollection, "rkey": rkey.String(), diff --git a/web/src/lib/api/lexicons/index.ts b/web/src/lib/api/lexicons/index.ts index 4bb6aacfc..7a5749a1e 100644 --- a/web/src/lib/api/lexicons/index.ts +++ b/web/src/lib/api/lexicons/index.ts @@ -193,6 +193,7 @@ export * as OrgTangledTempServerCreateInviteCode from "./types/org/tangled/temp/ export * as OrgTangledTempServerHandoverRepo from "./types/org/tangled/temp/server/handoverRepo.js"; export * as OrgTangledTempServerRedeemInviteCode from "./types/org/tangled/temp/server/redeemInviteCode.js"; export * as OrgTangledTempServerSeedRepo from "./types/org/tangled/temp/server/seedRepo.js"; +export * as OrgTangledTempServerUploadBlob from "./types/org/tangled/temp/server/uploadBlob.js"; export * as OrgTangledTempSiteClaimDomain from "./types/org/tangled/temp/site/claimDomain.js"; export * as OrgTangledTempSiteGetDomainClaim from "./types/org/tangled/temp/site/getDomainClaim.js"; export * as OrgTangledTempSiteReleaseDomain from "./types/org/tangled/temp/site/releaseDomain.js"; diff --git a/web/src/lib/api/lexicons/types/com/bad-example/pocket/defs.ts b/web/src/lib/api/lexicons/types/com/bad-example/pocket/defs.ts index a60611ee3..d93208555 100644 --- a/web/src/lib/api/lexicons/types/com/bad-example/pocket/defs.ts +++ b/web/src/lib/api/lexicons/types/com/bad-example/pocket/defs.ts @@ -4,9 +4,9 @@ import * as OrgTangledTempActorDefs from "../../../org/tangled/temp/actor/defs.j const _preferencesSchema = /*#__PURE__*/ v.array(() => { return /*#__PURE__*/ v.variant([ + OrgTangledTempActorDefs.onboardingPrefSchema, OrgTangledTempActorDefs.savedKnotPrefSchema, OrgTangledTempActorDefs.savedSpindlePrefSchema, - OrgTangledTempActorDefs.onboardingPrefSchema, ]); }); diff --git a/web/src/lib/api/lexicons/types/org/tangled/temp/server/seedRepo.ts b/web/src/lib/api/lexicons/types/org/tangled/temp/server/seedRepo.ts index 442bb08f3..c78be1ceb 100644 --- a/web/src/lib/api/lexicons/types/org/tangled/temp/server/seedRepo.ts +++ b/web/src/lib/api/lexicons/types/org/tangled/temp/server/seedRepo.ts @@ -9,37 +9,24 @@ const _mainSchema = /*#__PURE__*/ v.procedure( input: { type: "lex", schema: /*#__PURE__*/ v.object({ - defaultBranch: /*#__PURE__*/ v.optional(/*#__PURE__*/ v.string()), /** - * Name of the repository: mustn't contain '/', '\', '..', control/whitespace characters, and mustn't be just '.' or '..'. The knot will force lowercase it into the repo rkey. - * @maxLength 100 + * Raw `org.tangled.repo.manifest` record body */ - name: /*#__PURE__*/ v.constrain(/*#__PURE__*/ v.string(), [ - /*#__PURE__*/ v.stringLength(0, 100), - ]), + manifest: /*#__PURE__*/ v.unknown(), /** - * DID of the git repo owner, who doesn't have to necessarily be a member of this particular knot. - */ - ownerDid: /*#__PURE__*/ v.didString(), - /** - * Records to write. Max 200 per req. - * @minLength 1 + * Records to write besides the manifest. Max 200 per req. * @maxLength 200 */ get records() { return /*#__PURE__*/ v.constrain( /*#__PURE__*/ v.array(seedRecordSchema), - [/*#__PURE__*/ v.arrayLength(1, 200)], + [/*#__PURE__*/ v.arrayLength(0, 200)], ); }, /** * Existing repo DID which our COB records belong to. The foster knot will serve this DID as a shadow, and will give the record (authority) back to the principal knot once that given knot that *actually* owns the repo DID upgrades! */ repoDid: /*#__PURE__*/ v.didString(), - /** - * https URL of git repo to import content from. - */ - source: /*#__PURE__*/ v.optional(/*#__PURE__*/ v.genericUriString()), }), }, output: { diff --git a/web/src/lib/api/lexicons/types/org/tangled/temp/server/uploadBlob.ts b/web/src/lib/api/lexicons/types/org/tangled/temp/server/uploadBlob.ts new file mode 100644 index 000000000..48301d95c --- /dev/null +++ b/web/src/lib/api/lexicons/types/org/tangled/temp/server/uploadBlob.ts @@ -0,0 +1,40 @@ +import type {} from "@atcute/lexicons"; +import * as v from "@atcute/lexicons/validations"; +import type {} from "@atcute/lexicons/ambient"; + +const _mainSchema = /*#__PURE__*/ v.procedure( + "org.tangled.temp.server.uploadBlob", + { + params: /*#__PURE__*/ v.object({ + /** + * The Repo DID that a given blob belongs to + */ + repoDid: /*#__PURE__*/ v.didString(), + }), + input: { + type: "blob", + }, + output: { + type: "lex", + schema: /*#__PURE__*/ v.object({ + blob: /*#__PURE__*/ v.blob(), + }), + }, + }, +); + +type main$schematype = typeof _mainSchema; + +export interface mainSchema extends main$schematype {} + +export const mainSchema = _mainSchema as mainSchema; + +export interface $params extends v.InferInput {} +export type $input = v.InferXRPCBodyInput; +export interface $output extends v.InferXRPCBodyInput {} + +declare module "@atcute/lexicons/ambient" { + interface XRPCProcedures { + "org.tangled.temp.server.uploadBlob": mainSchema; + } +} -- 2.54.0