From ebb681abe4459ec2c3be01beb9596d06541b64db Mon Sep 17 00:00:00 2001 From: Lewis Date: Thu, 29 Oct 2026 17:23:22 +0200 Subject: [PATCH] knot2: small tweaks Lewis: May this revision serve well! --- bobbin/crates/resolver/src/lib.rs | 25 +- bobbin/crates/types/src/edges.rs | 14 +- bobbin/crates/xrpc/src/lib.rs | 11 +- bobbin/crates/xrpc/src/repo.rs | 5 +- docs/src/content/docs/upgrading/knot-2.md | 3 + knot2/crates/knot-atproto/src/auth.rs | 1 - knot2/crates/knot-cob/src/backend.rs | 6 +- knot2/crates/knot-cob/src/graph.rs | 2 +- knot2/crates/knot-cobs/src/comment.rs | 7 +- knot2/crates/knot-cobs/src/reaction.rs | 4 +- knot2/crates/knot-cobs/src/registry.rs | 4 +- knot2/crates/knot-cobs/src/social.rs | 7 +- knot2/crates/knot-cobs/src/ticket.rs | 4 +- knot2/crates/knot-cobs/tests/social.rs | 3 +- knot2/crates/knot-config/src/lib.rs | 40 ++-- knot2/crates/knot-git/src/reads.rs | 2 +- knot2/crates/knot-index/src/lib.rs | 2 +- knot2/crates/knot-index/src/social.rs | 8 +- knot2/crates/knot-index/tests/projections.rs | 32 ++- knot2/crates/knot-maintenance/tests/chaos.rs | 4 +- .../knot-maintenance/tests/common/mod.rs | 4 +- knot2/crates/knot-migrate/src/emit.rs | 5 + knot2/crates/knot-migrate/src/main.rs | 36 ++- knot2/crates/knot-migrate/tests/migrate.rs | 4 + knot2/crates/knot-pack/src/objects.rs | 5 +- knot2/crates/knot-pack/tests/common/mod.rs | 1 - knot2/crates/knot-receive/src/lib.rs | 6 +- knot2/crates/knot-record/src/artifact.rs | 4 +- knot2/crates/knot-record/src/chain.rs | 35 ++- knot2/crates/knot-record/src/reaction.rs | 4 +- knot2/crates/knot-record/src/ticket.rs | 5 +- knot2/crates/knot-secrets/src/lib.rs | 1 - knot2/crates/knot-ssh/tests/ssh_push.rs | 12 +- knot2/crates/knot-types/src/lib.rs | 12 +- knot2/crates/knot-workflow/src/lib.rs | 1 - knot2/crates/knot-xrpc/src/atproto.rs | 18 +- knot2/crates/knot-xrpc/src/blocklist.rs | 2 +- knot2/crates/knot-xrpc/src/foster.rs | 85 ++++--- knot2/crates/knot-xrpc/src/handover.rs | 148 +++++++----- knot2/crates/knot-xrpc/src/keep_sweep.rs | 56 +++-- knot2/crates/knot-xrpc/src/lfs.rs | 51 ++-- knot2/crates/knot-xrpc/src/lib.rs | 4 +- knot2/crates/knot-xrpc/src/members.rs | 4 +- knot2/crates/knot-xrpc/src/plc.rs | 43 ++-- knot2/crates/knot-xrpc/src/reads.rs | 20 +- knot2/crates/knot-xrpc/src/social.rs | 74 +++--- knot2/crates/knot-xrpc/src/tests.rs | 16 +- knot2/crates/knot-xrpc/tests/common/mod.rs | 15 +- knot2/crates/knot-xrpc/tests/handover.rs | 112 ++++++--- knot2/crates/knot-xrpc/tests/retire.rs | 219 ++++++++++++++---- knot2/example.toml | 7 - knotmirror/crates/knotmirror-index/src/lib.rs | 2 +- .../crates/knotmirror-ingest/src/lib.rs | 2 +- .../crates/knotmirror-maintenance/src/lib.rs | 4 +- knotmirror/crates/knotmirror-sync/src/lib.rs | 2 +- knotmirror/crates/knotmirror-sync/src/task.rs | 8 +- knotmirror/crates/knotmirror-xrpc/src/lib.rs | 2 +- .../crates/knotmirror-xrpc/src/routes/git.rs | 4 +- knotmirror/crates/knotmirror/src/config.rs | 5 +- 59 files changed, 775 insertions(+), 447 deletions(-) diff --git a/bobbin/crates/resolver/src/lib.rs b/bobbin/crates/resolver/src/lib.rs index 0706514d5..883e79c9e 100644 --- a/bobbin/crates/resolver/src/lib.rs +++ b/bobbin/crates/resolver/src/lib.rs @@ -411,11 +411,8 @@ impl RepoIdResolver { let result = cell .get_or_init(|| async { - match tokio::time::timeout( - RESOLVE_TIMEOUT, - self.fetch_repo_did(owner, rkey, &key), - ) - .await + match tokio::time::timeout(RESOLVE_TIMEOUT, self.fetch_repo_did(owner, rkey, &key)) + .await { Ok(resolution) => resolution, Err(_elapsed) => { @@ -424,10 +421,12 @@ impl RepoIdResolver { rkey = rkey.as_ref(), "repo resolution hung, caching as unresolvable under short TTL", ); - let probe = self.probe.as_ref().expect( - "fetch_repo_did is only called when a probe is present", - ); - self.stats.record_miss(MissKind::Transient, Some(RESOLVE_TIMEOUT)); + let probe = self + .probe + .as_ref() + .expect("fetch_repo_did is only called when a probe is present"); + self.stats + .record_miss(MissKind::Transient, Some(RESOLVE_TIMEOUT)); let expires_at = probe.clock.now_instant() + TRANSIENT_TTL; self.fill_transient(key.clone(), expires_at).await; Resolution::Unresolvable @@ -668,7 +667,10 @@ mod tests { struct HangingHttp; impl bobbin_runtime::HttpTransport for HangingHttp { - fn execute(&self, _request: bobbin_runtime::HttpRequest) -> bobbin_runtime::HttpResponseFuture { + fn execute( + &self, + _request: bobbin_runtime::HttpRequest, + ) -> bobbin_runtime::HttpResponseFuture { Box::pin(std::future::pending()) } } @@ -697,7 +699,8 @@ mod tests { .await; assert_eq!(again, Resolution::Unresolvable); assert_eq!( - resolver.stats().hits, 1, + resolver.stats().hits, + 1, "transient entry must stop retries during TTL such that wedgy upstream can't stall every frame refing to ident" ); } diff --git a/bobbin/crates/types/src/edges.rs b/bobbin/crates/types/src/edges.rs index 77111c45f..c8d8681f1 100644 --- a/bobbin/crates/types/src/edges.rs +++ b/bobbin/crates/types/src/edges.rs @@ -93,9 +93,9 @@ fn upgrade_bare_string_titles(value: &mut serde_json::Value) -> bool { } upgraded } - serde_json::Value::Array(items) => items - .iter_mut() - .fold(false, |upgraded, item| upgrade_bare_string_titles(item) | upgraded), + serde_json::Value::Array(items) => items.iter_mut().fold(false, |upgraded, item| { + upgrade_bare_string_titles(item) | upgraded + }), _ => false, } } @@ -115,13 +115,13 @@ impl Record { ) -> Result { macro_rules! parse { ($variant:ident) => { - Ok(Self::$variant( - serde_json::from_slice(bytes).or_else(|canon_err| { + Ok(Self::$variant(serde_json::from_slice(bytes).or_else( + |canon_err| { normalize_legacy_titles(bytes) .and_then(|upgraded| serde_json::from_slice(&upgraded).ok()) .ok_or(canon_err) - })?, - )) + }, + )?)) }; } match nsid.as_ref() { diff --git a/bobbin/crates/xrpc/src/lib.rs b/bobbin/crates/xrpc/src/lib.rs index 4f4e2c2cc..5dc5649ff 100644 --- a/bobbin/crates/xrpc/src/lib.rs +++ b/bobbin/crates/xrpc/src/lib.rs @@ -34,7 +34,9 @@ use bobbin_resolver::{HydrantClient, IdentityResolveError, IdentityResolver, Rep use bobbin_runtime::{ReqwestHttp, UnixMicros}; use bobbin_search::{ActorIndex, ActorIndexError, SearchReader}; use bobbin_slingshot_client::{SlingshotClient, SlingshotError}; -use bobbin_types::edges::{REPO_SOURCE_EDGE_KIND, SUBSCRIPTION_IGNORE_KIND, normalize_legacy_titles}; +use bobbin_types::edges::{ + REPO_SOURCE_EDGE_KIND, SUBSCRIPTION_IGNORE_KIND, normalize_legacy_titles, +}; use bobbin_types::edges::{TICKET_ISSUE_KIND, TICKET_PULL_KIND, ticket_editor}; use bobbin_types::ids::{EdgeKey, SubjectRef, nsid_static, owner_did_from_aturi}; use bobbin_types::org_tangled::actor::profile::{Profile, ProfileGetRecordOutput, ProfileRecord}; @@ -52,6 +54,7 @@ use bobbin_types::org_tangled::key::ssh_key::{SshKey, SshKeyGetRecordOutput, Ssh use bobbin_types::org_tangled::label::definition::{ Definition as LabelDefinition, DefinitionRecord as LabelDefinitionRecord, }; +use bobbin_types::org_tangled::repo::artifact::{Artifact, ArtifactRecord}; use bobbin_types::org_tangled::repo::manifest::{ Manifest, ManifestGetRecordOutput, ManifestRecord, }; @@ -59,7 +62,6 @@ use bobbin_types::org_tangled::review::patch::{Patch, PatchRecord}; use bobbin_types::org_tangled::snippet::string::{ SnippetString, SnippetStringGetRecordOutput, SnippetStringRecord, }; -use bobbin_types::org_tangled::repo::artifact::{Artifact, ArtifactRecord}; use bobbin_types::org_tangled::track::ticket::{Ticket, TicketGetRecordOutput, TicketRecord}; use bobbin_types::record::RecordBody; @@ -3722,7 +3724,10 @@ async fn list_tags_with_artifacts( }; let mut body: serde_json::Value = read_upstream_json(upstream, MAX_TAGS_BYTES).await?; let mut artifacts = artifacts_by_oid(&state, repo).await; - if let Some(tags) = body.get_mut("tags").and_then(serde_json::Value::as_array_mut) { + if let Some(tags) = body + .get_mut("tags") + .and_then(serde_json::Value::as_array_mut) + { for tag in tags { let Some(views) = tag .get("oid") diff --git a/bobbin/crates/xrpc/src/repo.rs b/bobbin/crates/xrpc/src/repo.rs index fdcd654ff..d27e5a3ff 100644 --- a/bobbin/crates/xrpc/src/repo.rs +++ b/bobbin/crates/xrpc/src/repo.rs @@ -29,7 +29,10 @@ pub(crate) async fn get_repo_by_name( ) -> Result, XrpcError> { let repo_did = state .claims - .repo_by_name(params.owner.as_str(), ¶ms.name.as_str().to_ascii_lowercase()) + .repo_by_name( + params.owner.as_str(), + ¶ms.name.as_str().to_ascii_lowercase(), + ) .and_then(|r| Did::new_owned(r).ok()) .ok_or(XrpcError::NotFound)?; let (manifest_uri, refs_uri) = (view::manifest_uri(&repo_did), view::refs_uri(&repo_did)); diff --git a/docs/src/content/docs/upgrading/knot-2.md b/docs/src/content/docs/upgrading/knot-2.md index 8dcd09f10..a40f4fc1c 100644 --- a/docs/src/content/docs/upgrading/knot-2.md +++ b/docs/src/content/docs/upgrading/knot-2.md @@ -90,12 +90,15 @@ sudo sh -c 'set -a; . /etc/knot/knot.env; set +a; exec knot-migrate \ --env-file /home/git/.knot.env \ --host-key /etc/ssh/ssh_host_ed25519_key \ --plc-url https://plc.directory \ + --contact-email ops@oyster.cafe \ --target /var/lib/knot \ --dry-run' ``` The `--env-file` gives your hostname and repository scan path, so you don't have to repeat them. +The contact email lands in the rendered config's `[contact]` section, +and comes out of `KNOT_CONTACT_EMAIL` in the env file when the flag stays off. `--dry-run` won't touch the target, though SQLite will still add `knotserver.db-shm` and `knotserver.db-wal` beside a write-ahead-log database even to read it. diff --git a/knot2/crates/knot-atproto/src/auth.rs b/knot2/crates/knot-atproto/src/auth.rs index 29b3a7c62..e936afd64 100644 --- a/knot2/crates/knot-atproto/src/auth.rs +++ b/knot2/crates/knot-atproto/src/auth.rs @@ -58,4 +58,3 @@ impl PointerAuthorizer for ServiceAuth<'_> { Ok(()) } } - diff --git a/knot2/crates/knot-cob/src/backend.rs b/knot2/crates/knot-cob/src/backend.rs index 23fe31dc0..0e2107887 100644 --- a/knot2/crates/knot-cob/src/backend.rs +++ b/knot2/crates/knot-cob/src/backend.rs @@ -59,11 +59,7 @@ fn prefixed_ref_name( type_name: &TypeName, object: CobId, ) -> Result { - let raw = format!( - "{prefix}{}/{}", - type_name.as_str(), - object.oid().to_hex() - ); + let raw = format!("{prefix}{}/{}", type_name.as_str(), object.oid().to_hex()); RefName::new(raw.as_str()).map_err(|_| CobError::RefName(raw)) } diff --git a/knot2/crates/knot-cob/src/graph.rs b/knot2/crates/knot-cob/src/graph.rs index ac25ef11d..64e1ffd78 100644 --- a/knot2/crates/knot-cob/src/graph.rs +++ b/knot2/crates/knot-cob/src/graph.rs @@ -1,7 +1,7 @@ use std::cmp::Reverse; -use std::collections::{BTreeMap, BinaryHeap}; #[cfg(test)] use std::collections::BTreeSet; +use std::collections::{BTreeMap, BinaryHeap}; use knot_types::{ChangeId, CobId, UnixSeconds}; diff --git a/knot2/crates/knot-cobs/src/comment.rs b/knot2/crates/knot-cobs/src/comment.rs index 7152dccf4..1f4d10ed3 100644 --- a/knot2/crates/knot-cobs/src/comment.rs +++ b/knot2/crates/knot-cobs/src/comment.rs @@ -35,7 +35,8 @@ impl StoredCid { } fn stored_ref(reference: StoredRef) -> Result { - let (_, address) = RecordAddress::parse_at_uri(&reference.uri).map_err(|error| error.to_string())?; + let (_, address) = + RecordAddress::parse_at_uri(&reference.uri).map_err(|error| error.to_string())?; Ok(CommentRef::new(address, reference.cid.resolve()?)) } @@ -58,8 +59,8 @@ struct StoredComment { } fn decode_stored(body: &RecordBody) -> Result { - let stored: StoredComment = serde_ipld_dagcbor::from_slice(body.as_bytes()) - .map_err(|error| error.to_string())?; + let stored: StoredComment = + serde_ipld_dagcbor::from_slice(body.as_bytes()).map_err(|error| error.to_string())?; Ok(Comment { subject: stored.subject.cid.resolve()?, reply: stored.reply_to.map(stored_ref).transpose()?, diff --git a/knot2/crates/knot-cobs/src/reaction.rs b/knot2/crates/knot-cobs/src/reaction.rs index 13f138524..9e10b2aff 100644 --- a/knot2/crates/knot-cobs/src/reaction.rs +++ b/knot2/crates/knot-cobs/src/reaction.rs @@ -19,8 +19,8 @@ struct StoredReaction { } fn decode_stored(body: &RecordBody) -> Result { - let stored: StoredReaction = serde_ipld_dagcbor::from_slice(body.as_bytes()) - .map_err(|error| error.to_string())?; + let stored: StoredReaction = + serde_ipld_dagcbor::from_slice(body.as_bytes()).map_err(|error| error.to_string())?; Ok(Reaction { subject: stored.subject.cid.resolve()?, kind: stored.reaction, diff --git a/knot2/crates/knot-cobs/src/registry.rs b/knot2/crates/knot-cobs/src/registry.rs index afd2fb9a2..48f0baffd 100644 --- a/knot2/crates/knot-cobs/src/registry.rs +++ b/knot2/crates/knot-cobs/src/registry.rs @@ -2,8 +2,8 @@ use std::collections::BTreeMap; use std::fmt; use knot_cob::{ - ChangePayload, Checkpoint, CobError, CobHome, CobStore, Evaluate, HistoryModel, - KnotAuthored, SnapshotStride, StateSize, + ChangePayload, Checkpoint, CobError, CobHome, CobStore, Evaluate, HistoryModel, KnotAuthored, + SnapshotStride, StateSize, }; use knot_runtime::Signer; use knot_types::{ diff --git a/knot2/crates/knot-cobs/src/social.rs b/knot2/crates/knot-cobs/src/social.rs index a3f8752d9..8a1c8f54f 100644 --- a/knot2/crates/knot-cobs/src/social.rs +++ b/knot2/crates/knot-cobs/src/social.rs @@ -2,7 +2,7 @@ use std::fmt::Debug; use std::marker::PhantomData; use knot_cob::{ - Approved, Append, Authored, Authorize, ChangePayload, Checkpoint, CobHome, CobStore, Emits, + Append, Approved, Authored, Authorize, ChangePayload, Checkpoint, CobHome, CobStore, Emits, Evaluate, HistoryModel, Materialized, Record, SnapshotStride, StateSize, WriteContext, }; use knot_runtime::Signer; @@ -145,7 +145,10 @@ pub(crate) fn settle( .state() .object() .and_then(|live| live.version()); - let Some(at) = versions.iter().rposition(|version| Some(*version) == current) else { + let Some(at) = versions + .iter() + .rposition(|version| Some(*version) == current) + else { return Ok(Settled::Superseded); }; for (editing, record) in edits.into_iter().skip(at) { diff --git a/knot2/crates/knot-cobs/src/ticket.rs b/knot2/crates/knot-cobs/src/ticket.rs index 91b4b30fb..2c71e9bae 100644 --- a/knot2/crates/knot-cobs/src/ticket.rs +++ b/knot2/crates/knot-cobs/src/ticket.rs @@ -24,8 +24,8 @@ struct StoredTicket { } fn decode_stored(body: &RecordBody) -> Result { - let stored: StoredTicket = serde_ipld_dagcbor::from_slice(body.as_bytes()) - .map_err(|error| error.to_string())?; + 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| { diff --git a/knot2/crates/knot-cobs/tests/social.rs b/knot2/crates/knot-cobs/tests/social.rs index ce4313c72..a18b5b5bf 100644 --- a/knot2/crates/knot-cobs/tests/social.rs +++ b/knot2/crates/knot-cobs/tests/social.rs @@ -302,8 +302,7 @@ fn a_forked_social_history_written_outside_the_knot_replays_through_the_converge .unwrap(); let replayed = CobStore::new(&repo).get::(object).unwrap(); - let knot_cob::Emits::Version(left_version) = - edited(&nel, opened_at, "left", 2).change.emits() + let knot_cob::Emits::Version(left_version) = edited(&nel, opened_at, "left", 2).change.emits() else { panic!("an edit emits a version") }; diff --git a/knot2/crates/knot-config/src/lib.rs b/knot2/crates/knot-config/src/lib.rs index b281b5082..19e3506a7 100644 --- a/knot2/crates/knot-config/src/lib.rs +++ b/knot2/crates/knot-config/src/lib.rs @@ -519,16 +519,16 @@ pub struct SocialConfig { /// If this knot is being spun up for the first time and the current /// day is *after* 2026-10-28, please disregard. This /// setting will be gutted once all knots are upgraded. - #[config(env = "KNOT_SOCIAL_FOSTER_URL", default = "https://knot2.mock.tangled.org")] + #[config( + env = "KNOT_SOCIAL_FOSTER_URL", + default = "https://knot2.mock.tangled.org" + )] pub foster_url: String, /// How many handovers that the upgrade of a knot2 at boot-time from a foster knot pull will run concurrently. #[config(env = "KNOT_SOCIAL_FOSTER_BOOT_WIDTH", default = 4)] pub foster_boot_width: u32, - /// Takes a handover ack even when the repo's DID document declares no - /// tangled_knot service endpoint, for the one-time 2026-10-28 upgrade, so - /// a slow PLC-ops tail can't stall the foster retire. Such an ack skips - /// the origin knot's signature and record checks, so leave this off - /// outside that upgrade. + + // The following are for the "foster knot", you don't need to touch. #[config(env = "KNOT_SOCIAL_FOSTER_ACK_SKIP_KNOT_SERVICE", default = false)] pub foster_ack_skip_knot_service: bool, #[config(env = "KNOT_SOCIAL_FOSTER_ACK_SKIP_ORIGIN_CHECKS", default = false)] @@ -1043,7 +1043,7 @@ fn is_http_token(value: &str) -> bool { .all(|byte| byte.is_ascii_alphanumeric() || b"!#$%&'*+-.^_`|~".contains(&byte)) } -fn is_contact_email(value: &str) -> bool { +pub fn is_contact_email(value: &str) -> bool { let mut parts = value.splitn(2, '@'); matches!( (parts.next(), parts.next()), @@ -1095,16 +1095,16 @@ pub struct Validated { impl Validated { fn new(config: KnotConfig) -> Result { config.validate()?; - let catalog = knot_messages::Catalog::parse(&config.messages) - .map_err(|error| ConfigError { errors: vec![error.to_string()] })?; + let catalog = + knot_messages::Catalog::parse(&config.messages).map_err(|error| ConfigError { + errors: vec![error.to_string()], + })?; let trusted_proxies = TrustedProxies::parse( - config - .xrpc - .trusted_proxies - .iter() - .map(String::as_str), + config.xrpc.trusted_proxies.iter().map(String::as_str), ) - .map_err(|error| ConfigError { errors: vec![format!("xrpc.trusted_proxies: {error}")] })?; + .map_err(|error| ConfigError { + errors: vec![format!("xrpc.trusted_proxies: {error}")], + })?; let ssh_proxy_protocol_from = TrustedProxies::parse( config .server @@ -1112,10 +1112,8 @@ impl Validated { .iter() .map(String::as_str), ) - .map_err(|error| { - ConfigError { - errors: vec![format!("server.ssh_proxy_protocol_from: {error}")], - } + .map_err(|error| ConfigError { + errors: vec![format!("server.ssh_proxy_protocol_from: {error}")], })?; let crawlers = config .atproto @@ -1123,7 +1121,9 @@ impl Validated { .iter() .map(|raw| KnotServiceUrl::new(raw.as_str())) .collect::, _>>() - .map_err(|error| ConfigError { errors: vec![format!("atproto.crawlers: {error}")] })?; + .map_err(|error| ConfigError { + errors: vec![format!("atproto.crawlers: {error}")], + })?; Ok(Self { config, crawlers, diff --git a/knot2/crates/knot-git/src/reads.rs b/knot2/crates/knot-git/src/reads.rs index 1b26be45d..50559153b 100644 --- a/knot2/crates/knot-git/src/reads.rs +++ b/knot2/crates/knot-git/src/reads.rs @@ -3,7 +3,7 @@ use std::ops::ControlFlow; use std::time::Instant; use gix::bstr::ByteSlice; -use knot_types::{BranchName, OriginUrl, Oid, RepoPath, TagName, UnixSeconds}; +use knot_types::{BranchName, Oid, OriginUrl, RepoPath, TagName, UnixSeconds}; use crate::error::{GitError, backend}; use crate::objects::{Commit, CommitRange, EntryKind, Identity, identity, map_kind}; diff --git a/knot2/crates/knot-index/src/lib.rs b/knot2/crates/knot-index/src/lib.rs index 7c9b1076d..1b4e853a6 100644 --- a/knot2/crates/knot-index/src/lib.rs +++ b/knot2/crates/knot-index/src/lib.rs @@ -19,11 +19,11 @@ use knot_cobs::{ MembersCob, RegistryChange, RepoRef, RepoRegistryCob, }; use knot_git::{Layout, Repo}; +use knot_resource::StructureBytes; use knot_types::{ AccountDid, ClonePath, CobId, OfferedKey, OwnerDid, RecordAddress, RecordCollection, RepoDid, RepoPolicy, RepoRkey, SubjectNumber, TypeName, UnixSeconds, }; -use knot_resource::StructureBytes; use tokio::sync::watch; use intern::{Interner, RepoKey}; diff --git a/knot2/crates/knot-index/src/social.rs b/knot2/crates/knot-index/src/social.rs index 0c3f6c237..23cb3947f 100644 --- a/knot2/crates/knot-index/src/social.rs +++ b/knot2/crates/knot-index/src/social.rs @@ -68,6 +68,10 @@ impl SocialEntryState { Self::Erased | Self::Contested => None, } } + + pub const fn erased(self) -> bool { + matches!(self, Self::Erased) + } } #[derive(Debug, Clone, PartialEq, Eq)] @@ -260,9 +264,7 @@ impl RepoSocial { } } - fn reaction_key( - objected: &Objected, - ) -> Option<(Slot, AccountKey, knot_cobs::ReactionKind)> { + fn reaction_key(objected: &Objected) -> Option<(Slot, AccountKey, knot_cobs::ReactionKind)> { Some(( objected.parent?, objected.reacted?.actor, diff --git a/knot2/crates/knot-index/tests/projections.rs b/knot2/crates/knot-index/tests/projections.rs index 3bf534ea5..87547a3cf 100644 --- a/knot2/crates/knot-index/tests/projections.rs +++ b/knot2/crates/knot-index/tests/projections.rs @@ -1924,9 +1924,21 @@ fn reaction_index_answers_live_reactions_only() { cold.refresh_social(&repo).unwrap(); for (actor, kind, expected) in [ - (&acc("nel"), ReactionKind::new("👍").unwrap(), Some(nels_like)), - (&acc("olaren"), ReactionKind::new("👍").unwrap(), Some(olarens_like)), - (&acc("nel"), ReactionKind::new("❤️").unwrap(), Some(nels_heart)), + ( + &acc("nel"), + ReactionKind::new("👍").unwrap(), + Some(nels_like), + ), + ( + &acc("olaren"), + ReactionKind::new("👍").unwrap(), + Some(olarens_like), + ), + ( + &acc("nel"), + ReactionKind::new("❤️").unwrap(), + Some(nels_heart), + ), (&acc("olaren"), ReactionKind::new("❤️").unwrap(), None), (&acc("nel"), ReactionKind::new("👀").unwrap(), None), ] { @@ -1951,12 +1963,22 @@ fn reaction_index_answers_live_reactions_only() { ); index.refresh_social(&repo).unwrap(); assert_eq!( - index.social_live_reaction(&repo, &parent, &acc("nel"), ReactionKind::new("👍").unwrap()), + index.social_live_reaction( + &repo, + &parent, + &acc("nel"), + ReactionKind::new("👍").unwrap() + ), Resolved::Ready(None), "a deleted reaction should probably stop being found" ); assert_eq!( - index.social_live_reaction(&repo, &parent, &acc("olaren"), ReactionKind::new("👍").unwrap()), + index.social_live_reaction( + &repo, + &parent, + &acc("olaren"), + ReactionKind::new("👍").unwrap() + ), Resolved::Ready(Some(olarens_like)), "another actor's same-kind reaction should still be found" ); diff --git a/knot2/crates/knot-maintenance/tests/chaos.rs b/knot2/crates/knot-maintenance/tests/chaos.rs index 0f3d5b361..b76d0ff3e 100644 --- a/knot2/crates/knot-maintenance/tests/chaos.rs +++ b/knot2/crates/knot-maintenance/tests/chaos.rs @@ -8,9 +8,7 @@ use knot_git::{ use knot_maintenance::{ FullRepackInterval, GeometricFactor, Options, PruneGrace, ReflogRetention, run_repo, }; -use knot_types::{ - AuthorName, BranchName, Email, ObjectCount, Oid, RefName, RepoDid, UnixSeconds, -}; +use knot_types::{AuthorName, BranchName, Email, ObjectCount, Oid, RefName, RepoDid, UnixSeconds}; const DID: &str = "did:plc:squid"; const EMPTY_TREE: &str = "4b825dc642cb6eb9a060e54bf8d69288fbee4904"; diff --git a/knot2/crates/knot-maintenance/tests/common/mod.rs b/knot2/crates/knot-maintenance/tests/common/mod.rs index b0af1d039..2270ef9ce 100644 --- a/knot2/crates/knot-maintenance/tests/common/mod.rs +++ b/knot2/crates/knot-maintenance/tests/common/mod.rs @@ -7,9 +7,7 @@ use std::path::Path; use knot_git::{ EntryKind, Identity, Layout, NewCommit, RefUpdate, Repo, StagedAction, StagedChange, }; -use knot_maintenance::{ - FullRepackInterval, GeometricFactor, Options, PruneGrace, ReflogRetention, -}; +use knot_maintenance::{FullRepackInterval, GeometricFactor, Options, PruneGrace, ReflogRetention}; use knot_types::{ AuthorName, BranchName, Email, ObjectCount, ObjectFormat, Oid, RefName, RepoDid, UnixSeconds, }; diff --git a/knot2/crates/knot-migrate/src/emit.rs b/knot2/crates/knot-migrate/src/emit.rs index f2b597138..1bcde4971 100644 --- a/knot2/crates/knot-migrate/src/emit.rs +++ b/knot2/crates/knot-migrate/src/emit.rs @@ -661,6 +661,7 @@ impl std::fmt::Display for MasterKeyEnv { pub struct ConfigValues { pub hostname: KnotHostname, pub admins: Vec, + pub contact_email: String, pub scan_path: PathBuf, pub ssh_host_key_file: PathBuf, pub sealed_key_file: PathBuf, @@ -706,6 +707,10 @@ pub fn render_config(values: &ConfigValues) -> Result { ("atproto", "plc_directory"), quote(values.plc_directory.as_str()), ), + ( + ("contact", "email"), + quote(&values.contact_email), + ), ] .into_iter() .collect(); diff --git a/knot2/crates/knot-migrate/src/main.rs b/knot2/crates/knot-migrate/src/main.rs index 35908f110..040df796a 100644 --- a/knot2/crates/knot-migrate/src/main.rs +++ b/knot2/crates/knot-migrate/src/main.rs @@ -34,6 +34,7 @@ options: --target knot data directory to create --hostname knot hostname, defaults to KNOT_SERVER_HOSTNAME --plc-url PLC directory, defaults to KNOT_SERVER_PLC_URL + --contact-email knot contact email, defaults to KNOT_CONTACT_EMAIL --object-format sha1 or sha256 for repos knot creates, default sha1 --master-key-env env var holding the base64 master key, default KNOT_MASTER_KEY --consume-source move the source repos into place instead of copying them, @@ -78,9 +79,7 @@ enum MigrateError { Refused(Refusals), #[error("the rehearsal lists what the migration still needs")] RehearsalIncomplete, - #[error( - "the migration skipped {count} repos, each with its reason in the skip list above" - )] + #[error("the migration skipped {count} repos, each with its reason in the skip list above")] SkippedRepos { count: usize }, #[error("{context}: {source}")] Io { @@ -146,6 +145,7 @@ struct Args { target: PathBuf, hostname: Option, plc_url: Option, + contact_email: Option, object_format: ObjectFormat, master_key_env: MasterKeyEnv, source_policy: SourcePolicy, @@ -162,7 +162,7 @@ struct Switches { force_host_key: bool, } -const KNOWN_FLAGS: [&str; 9] = [ +const KNOWN_FLAGS: [&str; 10] = [ "source-db", "source-repos", "env-file", @@ -170,6 +170,7 @@ const KNOWN_FLAGS: [&str; 9] = [ "target", "hostname", "plc-url", + "contact-email", "object-format", "master-key-env", ]; @@ -260,6 +261,7 @@ fn parse_args(args: &[String]) -> Result { target: required("target", take("target"))?.into(), hostname: take("hostname"), plc_url: take("plc-url"), + contact_email: take("contact-email"), object_format, master_key_env: take("master-key-env") .map_or_else(|| MasterKeyEnv::new("KNOT_MASTER_KEY"), MasterKeyEnv::new) @@ -342,6 +344,20 @@ fn run(args: &[String]) -> Result<(), MigrateError> { .ok_or_else(|| { MigrateError::Usage(format!("PLC directory {plc_url} isn't an https URL")) })?; + let contact_email = args + .contact_email + .clone() + .or_else(|| env.get("KNOT_CONTACT_EMAIL").map(str::to_string)) + .ok_or_else(|| { + MigrateError::Usage( + "--contact-email is required when the env file specifies none".to_string(), + ) + })?; + if !knot_config::is_contact_email(&contact_email) { + return Err(MigrateError::Usage(format!( + "--contact-email must be a contact email, not {contact_email}" + ))); + } let source_repos = args .source_repos .clone() @@ -432,7 +448,14 @@ fn run(args: &[String]) -> Result<(), MigrateError> { Err(MigrateError::Refused(refusals)) } (false, true) => { - let written = materialize(&args, &hostname, &plc_directory, &source_repos, &mapping)?; + let written = materialize( + &args, + &hostname, + &plc_directory, + &contact_email, + &source_repos, + &mapping, + )?; report( &mapping, orphan_alias_count, @@ -511,6 +534,7 @@ fn materialize( args: &Args, hostname: &KnotHostname, plc_directory: &Url, + contact_email: &str, source_repos: &Path, mapping: &Mapping, ) -> Result { @@ -567,6 +591,7 @@ fn materialize( let config = emit::render_config(&ConfigValues { hostname: hostname.clone(), admins: vec![mapping.knot_owner.clone()], + contact_email: contact_email.to_string(), scan_path, ssh_host_key_file: host_key_destination, sealed_key_file, @@ -620,6 +645,7 @@ mod tests { fn every_switch_is_off_until_it_is_named() { let off = parse(&["--source-db=/db", "--target=/t"]).unwrap(); assert!(!off.dry_run); + assert_eq!(off.contact_email, None); assert!(!off.skip_unreadable); assert_eq!(off.source_policy, SourcePolicy::Preserve); assert_eq!( diff --git a/knot2/crates/knot-migrate/tests/migrate.rs b/knot2/crates/knot-migrate/tests/migrate.rs index 7a5cba991..d5a493215 100644 --- a/knot2/crates/knot-migrate/tests/migrate.rs +++ b/knot2/crates/knot-migrate/tests/migrate.rs @@ -652,6 +652,7 @@ fn rendered_config_loads() { let config = emit::render_config(&ConfigValues { hostname: KnotHostname::new("knot.oyster.cafe").unwrap(), admins: vec![acc("bailey")], + contact_email: "ops@oyster.cafe".to_string(), scan_path: scan_path.clone(), ssh_host_key_file: dir.path().join("ssh_host_key"), sealed_key_file: dir.path().join("sealed-keys"), @@ -664,6 +665,7 @@ fn rendered_config_loads() { assert!(config.contains("admins = [\"did:plc:bailey\"]")); assert!(config.contains("admission = \"closed\"")); assert!(config.contains("object_format = \"sha1\"")); + assert!(config.contains("email = \"ops@oyster.cafe\"")); let path = dir.path().join("config.toml"); std::fs::write(&path, &config).unwrap(); knot_config::load(Some(&path)).unwrap(); @@ -1675,6 +1677,8 @@ fn run_migrate(fx: &Fixture, host_key: &Path, extra: &[&str]) -> std::process::O "knot.oyster.cafe", "--plc-url", "https://plc.directory", + "--contact-email", + "ops@oyster.cafe", "--target", fx.target.to_str().unwrap(), ]) diff --git a/knot2/crates/knot-pack/src/objects.rs b/knot2/crates/knot-pack/src/objects.rs index e00915c51..0dd965dc5 100644 --- a/knot2/crates/knot-pack/src/objects.rs +++ b/knot2/crates/knot-pack/src/objects.rs @@ -332,8 +332,9 @@ pub(crate) fn open_pack_file( let signature_len = std::fs::File::open(path)?.read(&mut signature).unwrap_or(0); match (signature_len, signature.as_slice()) { (0, _) => Err(PackError::Pack("packfile is empty".to_string())), - (_, b"PACK") => gix_pack::data::File::at(path, kind) - .map_err(|error| PackError::Pack(error.to_string())), + (_, b"PACK") => { + gix_pack::data::File::at(path, kind).map_err(|error| PackError::Pack(error.to_string())) + } _ => Err(PackError::Pack( "packfile is missing its PACK signature".to_string(), )), diff --git a/knot2/crates/knot-pack/tests/common/mod.rs b/knot2/crates/knot-pack/tests/common/mod.rs index 30e27dbb7..9707818e5 100644 --- a/knot2/crates/knot-pack/tests/common/mod.rs +++ b/knot2/crates/knot-pack/tests/common/mod.rs @@ -227,7 +227,6 @@ pub fn object_set(dir: &Path) -> BTreeSet { .collect() } - pub fn serve_dids() -> Arc { Arc::new(|target: &RepoTarget| match target { RepoTarget::Did(did) => RepoLookup::Hosted(did.clone()), diff --git a/knot2/crates/knot-receive/src/lib.rs b/knot2/crates/knot-receive/src/lib.rs index 8b93f5d38..cecfd9c2b 100644 --- a/knot2/crates/knot-receive/src/lib.rs +++ b/knot2/crates/knot-receive/src/lib.rs @@ -285,10 +285,8 @@ fn skip_escape(chars: &mut std::iter::Peekable>) { Some('[') => { chars.find(|c| ('@'..='~').contains(c)); } - Some(']') => { - if chars.find(|c| matches!(c, '\u{7}' | '\u{1b}')) == Some('\u{1b}') { - chars.next_if_eq(&'\\'); - } + Some(']') if chars.find(|c| matches!(c, '\u{7}' | '\u{1b}')) == Some('\u{1b}') => { + chars.next_if_eq(&'\\'); } _ => {} } diff --git a/knot2/crates/knot-record/src/artifact.rs b/knot2/crates/knot-record/src/artifact.rs index d4775ba27..f2385db79 100644 --- a/knot2/crates/knot-record/src/artifact.rs +++ b/knot2/crates/knot-record/src/artifact.rs @@ -49,7 +49,9 @@ impl<'de> Deserialize<'de> for ArtifactBlob { let mime = match AttachmentMime::new(&blob.mime) { Ok(mime) => mime, Err(MimeRefused::Empty) => { - return Err(serde::de::Error::custom("An artifact's `mimeType` is missing.")); + return Err(serde::de::Error::custom( + "An artifact's `mimeType` is missing.", + )); } Err(MimeRefused::Html(mime)) => { return Err(serde::de::Error::custom(format!( diff --git a/knot2/crates/knot-record/src/chain.rs b/knot2/crates/knot-record/src/chain.rs index 0638e3f65..3f72f64a9 100644 --- a/knot2/crates/knot-record/src/chain.rs +++ b/knot2/crates/knot-record/src/chain.rs @@ -882,11 +882,11 @@ fn take_extra_blocks(plan: &mut Plan<'_, S>, draft: &mut ChunkDraft) { }) .map(|(cid, _)| *cid) .collect(); - draft - .blocks - .extend(taken.into_iter().filter_map(|cid| { - plan.extra_blocks.remove(&cid).map(|bytes| (cid, bytes)) - })); + draft.blocks.extend( + taken + .into_iter() + .filter_map(|cid| plan.extra_blocks.remove(&cid).map(|bytes| (cid, bytes))), + ); } async fn emit_chunk( @@ -1244,9 +1244,7 @@ mod tests { Self { due: leaves .iter() - .filter_map(|(key, cid)| { - Some((key.clone(), Due::Keep(RecordCid::new(*cid)?))) - }) + .filter_map(|(key, cid)| Some((key.clone(), Due::Keep(RecordCid::new(*cid)?)))) .collect(), extras, } @@ -1286,8 +1284,14 @@ mod tests { .collect(); let blocks_at_tip = tip(&repo).unwrap().unwrap().blocks; futures::stream::iter([ - (body(50).cid().cid(), Bytes::from(body(50).as_bytes().to_vec())), - (body(51).cid().cid(), Bytes::from(body(51).as_bytes().to_vec())), + ( + body(50).cid().cid(), + Bytes::from(body(50).as_bytes().to_vec()), + ), + ( + body(51).cid().cid(), + Bytes::from(body(51).as_bytes().to_vec()), + ), ]) .for_each(|(cid, bytes)| { let repo = &repo; @@ -1295,7 +1299,10 @@ mod tests { async move { assert!(served.contains(&cid), "the chunk blocks name {cid}"); assert_eq!( - blocks::GitBlocks::at(repo, blocks_at_tip).get(&cid).await.unwrap(), + blocks::GitBlocks::at(repo, blocks_at_tip) + .get(&cid) + .await + .unwrap(), Some(bytes), "the block tree holds the historical body by its own cid" ); @@ -1324,7 +1331,11 @@ mod tests { ) .await .unwrap(); - assert_eq!(chunks.len(), 1, "one commit is written for the extra blocks"); + assert_eq!( + chunks.len(), + 1, + "one commit is written for the extra blocks" + ); assert!( chunks[0].ops.is_empty(), "the commit has no record ops, only the extra blocks" diff --git a/knot2/crates/knot-record/src/reaction.rs b/knot2/crates/knot-record/src/reaction.rs index a642d1d9e..6786b3544 100644 --- a/knot2/crates/knot-record/src/reaction.rs +++ b/knot2/crates/knot-record/src/reaction.rs @@ -24,7 +24,9 @@ pub struct ReactionKind { impl ReactionKind { pub fn new(value: &str) -> Result { if value.len() > MAX_REACTION_BYTES { - return Err(format!("reaction {value:?} is longer than {MAX_REACTION_BYTES} bytes")); + return Err(format!( + "reaction {value:?} is longer than {MAX_REACTION_BYTES} bytes" + )); } if value.graphemes(true).count() > 1 { return Err(format!("reaction {value:?} is longer than one grapheme")); diff --git a/knot2/crates/knot-record/src/ticket.rs b/knot2/crates/knot-record/src/ticket.rs index 06f78f9e0..62e57a47b 100644 --- a/knot2/crates/knot-record/src/ticket.rs +++ b/knot2/crates/knot-record/src/ticket.rs @@ -2,9 +2,7 @@ use std::fmt; use std::str::FromStr; use jacquard_common::deps::codegen::unicode_segmentation::UnicodeSegmentation; -use knot_types::{ - AccountDid, BranchName, RecordBody, RecordBodyError, RefName, UnixSeconds, -}; +use knot_types::{AccountDid, BranchName, RecordBody, RecordBodyError, RefName, UnixSeconds}; use serde::{Deserialize, Serialize, Serializer}; use crate::comment::{MarkdownEncoding, markdown_encoding}; @@ -176,6 +174,7 @@ pub struct TicketRecord { } impl TicketRecord { + #[allow(clippy::too_many_arguments)] pub fn new( prose: Prose, state: TicketState, diff --git a/knot2/crates/knot-secrets/src/lib.rs b/knot2/crates/knot-secrets/src/lib.rs index 788b52b76..9eb64568b 100644 --- a/knot2/crates/knot-secrets/src/lib.rs +++ b/knot2/crates/knot-secrets/src/lib.rs @@ -149,7 +149,6 @@ impl KeyMaterial { #[serde(transparent)] struct EncodedSecret(String); - #[derive(Serialize, Deserialize)] struct VaultFile { version: u32, diff --git a/knot2/crates/knot-ssh/tests/ssh_push.rs b/knot2/crates/knot-ssh/tests/ssh_push.rs index e43192f31..e57af9395 100644 --- a/knot2/crates/knot-ssh/tests/ssh_push.rs +++ b/knot2/crates/knot-ssh/tests/ssh_push.rs @@ -764,20 +764,12 @@ async fn addressing_variants_land_for_rkey_and_tid_repos() { let tid_target = tid_repo_did(); let rkey_variants = [ format!("ssh://git@127.0.0.1:{port}/{}/{}", owner(), repo_name()), - format!( - "ssh://git@127.0.0.1:{port}/{}/{}.git", - owner(), - repo_name() - ), + format!("ssh://git@127.0.0.1:{port}/{}/{}.git", owner(), repo_name()), format!("ssh://git@127.0.0.1:{port}/nel.pet/{}", repo_name()), format!("ssh://git@127.0.0.1:{port}/{}", repo_did()), ]; let tid_variants = [ - format!( - "ssh://git@127.0.0.1:{port}/{}/{}", - owner(), - tid_repo_name() - ), + format!("ssh://git@127.0.0.1:{port}/{}/{}", owner(), tid_repo_name()), format!( "ssh://git@127.0.0.1:{port}/{}/{}.git", owner(), diff --git a/knot2/crates/knot-types/src/lib.rs b/knot2/crates/knot-types/src/lib.rs index 4cda71ea1..139e488b3 100644 --- a/knot2/crates/knot-types/src/lib.rs +++ b/knot2/crates/knot-types/src/lib.rs @@ -6,12 +6,12 @@ pub use changes::{ChangedFiles, ChangedFilesBudget, Listing}; mod ids; pub use ids::{ - AccountDid, ActorId, AppviewEndpoint, AuthorName, BranchName, ChangeId, ClonePath, - CobId, DidRkey, Email, HttpStatus, KnotHostname, KnotId, KnotServiceUrl, LabelKey, - LanguageBytes, LanguageName, ObjectCount, ObjectFormat, OfferedKey, Oid, - OriginUrl, OwnerDid, OwnerRef, ParseError, PushOption, PushOptions, RecordCollection, - RecordRkey, RefName, RefTransition, RepoDid, RepoName, RepoPath, RepoRkey, ServiceDid, TagName, - TypeName, UnixMicros, UnixSeconds, UriRkey, + AccountDid, ActorId, AppviewEndpoint, AuthorName, BranchName, ChangeId, ClonePath, CobId, + DidRkey, Email, HttpStatus, KnotHostname, KnotId, KnotServiceUrl, LabelKey, LanguageBytes, + LanguageName, ObjectCount, ObjectFormat, OfferedKey, Oid, OriginUrl, OwnerDid, OwnerRef, + ParseError, PushOption, PushOptions, RecordCollection, RecordRkey, RefName, RefTransition, + RepoDid, RepoName, RepoPath, RepoRkey, ServiceDid, TagName, TypeName, UnixMicros, UnixSeconds, + UriRkey, }; mod record; diff --git a/knot2/crates/knot-workflow/src/lib.rs b/knot2/crates/knot-workflow/src/lib.rs index 192ea8209..0f9485d80 100644 --- a/knot2/crates/knot-workflow/src/lib.rs +++ b/knot2/crates/knot-workflow/src/lib.rs @@ -247,7 +247,6 @@ fn constraint_matches( } } - fn glob_set(patterns: &[String]) -> Result { patterns .iter() diff --git a/knot2/crates/knot-xrpc/src/atproto.rs b/knot2/crates/knot-xrpc/src/atproto.rs index 64935b350..e9a17bae3 100644 --- a/knot2/crates/knot-xrpc/src/atproto.rs +++ b/knot2/crates/knot-xrpc/src/atproto.rs @@ -264,16 +264,14 @@ pub(crate) async fn sync_get_repo( ) -> Result { let found = resolve(&state, ¶ms.did)?; let response_limit = state.byte_limits.response; - serve_car( - found, - response_limit, - move |found, tip| match ¶ms.since { + serve_car(found, response_limit, move |found, tip| { + match ¶ms.since { Some(since) => { frames_since_blocks(found, tip, since.tid().clone(), response_limit.get()) } None => whole_tree_blocks(found, tip), - }, - ) + } + }) .await } @@ -572,13 +570,7 @@ pub(crate) async fn sync_list_repos( .ok() }) .collect(); - crate::reads::json( - ListReposOut { - repos, - cursor, - }, - response_limit, - ) + crate::reads::json(ListReposOut { repos, cursor }, response_limit) }) .await } diff --git a/knot2/crates/knot-xrpc/src/blocklist.rs b/knot2/crates/knot-xrpc/src/blocklist.rs index ea45922b9..0fa7b3d0a 100644 --- a/knot2/crates/knot-xrpc/src/blocklist.rs +++ b/knot2/crates/knot-xrpc/src/blocklist.rs @@ -8,10 +8,10 @@ use knot_record::grant::GrantCollection; use knot_runtime::{Clock, HttpTransport}; use knot_types::{AccountDid, UnixSeconds}; +use crate::XrpcState; use crate::cob::Authorized; use crate::error::XrpcError; use crate::members::{commit_grant_list, require_admin}; -use crate::XrpcState; pub(crate) async fn block_add( state: &Arc>, diff --git a/knot2/crates/knot-xrpc/src/foster.rs b/knot2/crates/knot-xrpc/src/foster.rs index 5580549af..ae847a86c 100644 --- a/knot2/crates/knot-xrpc/src/foster.rs +++ b/knot2/crates/knot-xrpc/src/foster.rs @@ -21,14 +21,12 @@ use knot_capability::Capability; use knot_cob::{ ChangePayload, CobHome, CobStore, Emits, HistoryModel, KnotAuthored, Materialized, Record, }; -use knot_cobs::{ - Registry, RepoRef, RepoRegistryCob, Retired, deregister_repo, retire_repo, -}; +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, HttpRequest, PublicKeyBytes, Signature, Signer}; +use knot_runtime::{Clock, HttpRequest, HttpTransport, PublicKeyBytes, Signature, Signer}; use knot_types::crypto::{KeyCodec, PublicKey as CryptoKey}; use knot_types::{ AccountDid, ActorId, CobId, ContributionPolicy, MstKey, Nsid, OwnerDid, RecordAddress, @@ -246,15 +244,16 @@ impl CheckedRecord { return Err(SeedAdmission::Stamp(record.created_at)); } if record.record["$type"].as_str() != Some(record.collection.as_str()) { - return Err(SeedAdmission::TypeMismatch(record.collection.as_str().to_owned())); + return Err(SeedAdmission::TypeMismatch( + record.collection.as_str().to_owned(), + )); } if contains_nul(&record.record) { return Err(SeedAdmission::Nul); } if record.collection.as_str() == knot_record::ticket::TICKET_COLLECTION && record.record["title"]["$type"].as_str() == Some(knot_record::ticket::INLINE_MARKUP) - && record - .record["title"]["text"] + && record.record["title"]["text"] .as_str() .is_some_and(|text| text.trim().is_empty()) { @@ -265,7 +264,7 @@ impl CheckedRecord { None => { return Err(SeedAdmission::Unsettleable( record.collection.as_str().to_owned(), - )) + )); } Some(Err(why)) => return Err(SeedAdmission::Undecodable(why)), Some(Ok(())) => {} @@ -479,8 +478,14 @@ async fn seed_repo( { // an interrupted seed can register the repo without ever opening its manifest Err(error) if error.is_not_found() => { - crate::social::open_manifest(state, actor, existing.clone(), manifest, state.now()) - .await? + crate::social::open_manifest( + state, + actor, + existing.clone(), + manifest, + state.now(), + ) + .await? } other => other?, } @@ -624,7 +629,7 @@ async fn origin_knot_url( XrpcError::invalid_request(error.to_string()) } other => other.into(), - }) + }); } }; if !crate::handover::url_is_trusted(&url) { @@ -645,14 +650,19 @@ async fn origin_key( origin: &Url, ) -> Result { let multibase = knot_identity( - state, origin, "origin knot", &Capability::SocialWrites, + 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")); + return Err(XrpcError::bad_gateway( + "the origin knot's #atproto key isn't secp256k1", + )); } Ok(PublicKeyBytes::from_bytes(key.bytes.as_ref().to_vec())) } @@ -686,9 +696,12 @@ async fn spot_check_record( .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().and_then(RecordCid::parse).ok_or_else(|| { - XrpcError::bad_gateway("the origin knot served getRecord without a cid it can name") - })?; + let cid = served["cid"] + .as_str() + .and_then(RecordCid::parse) + .ok_or_else(|| { + XrpcError::bad_gateway("the origin knot served getRecord without a cid it can name") + })?; if cid != version.produces { return Err(XrpcError::bad_gateway(format!( "the origin knot serves {} at {cid} while the shadow recorded {}", @@ -847,8 +860,7 @@ async fn commit_seeded( .map(|record| { let collection = record.collection.clone(); let rkey = record.rkey.clone(); - let outcome = - CheckedRecord::new(record, now); + let outcome = CheckedRecord::new(record, now); (collection, rkey, outcome) }) .collect(); @@ -858,7 +870,11 @@ async fn commit_seeded( async fn seed_checked( state: &Arc>, repo_did: &RepoDid, - checked: Vec<(RecordCollection, RecordRkey, Result)>, + checked: Vec<( + RecordCollection, + RecordRkey, + Result, + )>, deletes: Vec, ) -> Result, XrpcError> { let signer = state.secrets.signer(&state.knot_did)?; @@ -898,13 +914,7 @@ async fn seed_checked( }) .collect::>(); results.extend(erase_records( - &state, - &git, - &store, - &home, - &repo, - &signer, - deletes, + &state, &git, &store, &home, &repo, &signer, deletes, )?); Ok(results) }) @@ -1012,7 +1022,6 @@ pub(crate) fn history_of(type_name: &knot_types::TypeName) -> Option, @@ -1105,9 +1114,10 @@ fn erase_one( address: &RecordAddress, now: UnixSeconds, ) -> Result<(), XrpcError> { - let object = objects.get(address).copied().ok_or_else(|| { - XrpcError::not_found("No such seeded record at that address") - })?; + let object = objects + .get(address) + .copied() + .ok_or_else(|| XrpcError::not_found("No such seeded record at that address"))?; let from = leaves .get(&address.mst_key()) .copied() @@ -1117,7 +1127,10 @@ fn erase_one( home, object, &Materialized { - change: SeedChange::Erase(SeedErasure { from, erased_at: now }), + change: SeedChange::Erase(SeedErasure { + from, + erased_at: now, + }), record: Record::Tombstone, }, signer, @@ -1165,7 +1178,9 @@ async fn remint_shadow( .await? .filter(|record| record.fostered) else { - return Err(XrpcError::not_found("That repo isn't fostered on this knot")); + return Err(XrpcError::not_found( + "That repo isn't fostered on this knot", + )); }; if record.retired_at.is_some() { return Err(XrpcError::conflict( @@ -1251,7 +1266,11 @@ fn seeded_snapshot(store: &CobStore) -> Result, XrpcError> { .map(|version| { let body = crate::social::tree_body(store, &seed_type, version.object, version.produces)? - .ok_or_else(|| XrpcError::internal("one record in this seeded recordset hasn't got any body data"))?; + .ok_or_else(|| { + XrpcError::internal( + "one record in this seeded recordset hasn't got any body data", + ) + })?; Ok(CheckedRecord { address: version.address, author: version.seeded.author, diff --git a/knot2/crates/knot-xrpc/src/handover.rs b/knot2/crates/knot-xrpc/src/handover.rs index a5136f047..24b4c68ab 100644 --- a/knot2/crates/knot-xrpc/src/handover.rs +++ b/knot2/crates/knot-xrpc/src/handover.rs @@ -93,9 +93,10 @@ impl Drop for HandoverLease { } pub(crate) fn pending(state: &XrpcState, repo: &RepoDid) -> bool { - state.layout.repo_path(repo).is_ok_and(|path| { - path.join(SIDECAR).exists() && !path.join(SETTLED).exists() - }) + state + .layout + .repo_path(repo) + .is_ok_and(|path| path.join(SIDECAR).exists() && !path.join(SETTLED).exists()) } fn mark_settled(repo_path: &Path) -> std::io::Result<()> { @@ -396,8 +397,7 @@ async fn handover( let missing: BTreeSet = wanted .iter() .filter(|(type_name, object, tip)| { - let name = - cob_ref_name(type_name, *object).expect("a served cob ref name parses"); + let name = cob_ref_name(type_name, *object).expect("a served cob ref name parses"); local.get(name.as_str()) != Some(tip) }) .map(|(_, _, tip)| *tip) @@ -489,7 +489,9 @@ fn seeded_ref_delete( object: CobId, ) -> Result, XrpcError> { let name = cob_ref_name(seed_type, object).expect("a served cob ref name parses"); - Ok(git.find_ref(&name)?.map(|old| RefUpdate::Delete { name, old })) + Ok(git + .find_ref(&name)? + .map(|old| RefUpdate::Delete { name, old })) } enum Shadow { @@ -533,10 +535,7 @@ fn apply_cob_refs( .filter_map(|(type_name, object, tip)| { let name = cob_ref_name(type_name, *object).expect("a served cob ref name parses"); match placed.get(name.as_str()) { - None => Some(Ok(RefUpdate::Create { - name, - new: *tip, - })), + None => Some(Ok(RefUpdate::Create { name, new: *tip })), Some(&old) if old == *tip => None, Some(&old) => Some(git.merge_base(old, *tip).map_err(XrpcError::from).and_then( |base| match base { @@ -565,27 +564,35 @@ fn confirm_local_leaves( converted: &[crate::social::Converted], ) -> Result<(), XrpcError> { let leaves = futures::executor::block_on(chain::leaves(git))?; - 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(|| { - XrpcError::internal(format!( - "the rebuilt chain doesn't serve {} at {} the way the foster knot's \ + 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(|| { + XrpcError::internal(format!( + "the rebuilt chain doesn't serve {} at {} the way the foster knot's \ last change recorded it", - version.address.mst_key().as_str(), - version.produces.cid() - )) - }) - }) + version.address.mst_key().as_str(), + version.produces.cid() + )) + }) + }, + ) } 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())) - }) + described["capabilities"] + .as_array() + .is_some_and(|capabilities| { + capabilities + .iter() + .any(|token| token.as_str() == Some(capability.token())) + }) } pub(crate) async fn knot_identity( @@ -612,14 +619,18 @@ pub(crate) async fn knot_identity( ))); } 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(); + 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) } @@ -637,7 +648,11 @@ async fn foster_key( Some(actor) => Ok(actor), None => { let multibase = knot_identity( - state, &foster.0, "foster knot", &Capability::Foster, "doesn't take fostered repos", + state, + &foster.0, + "foster knot", + &Capability::Foster, + "doesn't take fostered repos", ) .await?; let actor = ActorId::new(multibase).map_err(|_| { @@ -715,7 +730,10 @@ async fn confirm_foster_serving( if cursor.is_none() || empty { break; } - if cursor.as_deref().is_some_and(|at| !asked.insert(at.to_owned())) { + if cursor + .as_deref() + .is_some_and(|at| !asked.insert(at.to_owned())) + { return Err(XrpcError::bad_gateway(format!( "the foster knot keeps returning the same page of {collection} records, so \ its shadow can't be confirmed" @@ -749,9 +767,13 @@ fn collect_blob_cids(value: &Ipld, found: &mut BTreeSet) -> Result<(), { found.insert(BlobCid::new(*link).ok_or_else(|| link.to_string())?); } - fields.values().try_for_each(|field| collect_blob_cids(field, found))?; + fields + .values() + .try_for_each(|field| collect_blob_cids(field, found))?; } - Ipld::List(items) => items.iter().try_for_each(|item| collect_blob_cids(item, found))?, + Ipld::List(items) => items + .iter() + .try_for_each(|item| collect_blob_cids(item, found))?, _ => {} } Ok(()) @@ -768,11 +790,13 @@ async fn transfer_blobs( if matches!(converted.settled, knot_cobs::Settled::Superseded) { continue; } - let body: Ipld = serde_ipld_dagcbor::from_slice(converted.body.as_bytes()).map_err( - |error| XrpcError::internal(format!("a settled record body wouldn't decode: {error}")), - )?; - collect_blob_cids(&body, &mut wanted) - .map_err(|cid| XrpcError::bad_gateway(format!("a blob ref the store can't take: {cid}")))?; + let body: Ipld = + serde_ipld_dagcbor::from_slice(converted.body.as_bytes()).map_err(|error| { + XrpcError::internal(format!("a settled record body wouldn't decode: {error}")) + })?; + collect_blob_cids(&body, &mut wanted).map_err(|cid| { + XrpcError::bad_gateway(format!("a blob ref the store can't take: {cid}")) + })?; } if wanted.is_empty() { return Ok(()); @@ -793,19 +817,15 @@ async fn transfer_blobs( let bytes = fetch_blob(state, &url, &cid).await?; let claimed = ClaimedSize::new(bytes.len() as u64); let permit = lfs.handle.admission.admit(claimed).map_err(|error| { - XrpcError::internal(format!("the blob store can't take the transferred bytes: {error}")) + XrpcError::internal(format!( + "the blob store can't take the transferred bytes: {error}" + )) })?; let stored = Arc::clone(&lfs.handle.store); let target = repo.clone(); let owned = oid.clone(); run_blocking(move || { - let outcome = stored.put( - Pool::Attachments, - &target, - &owned, - claimed, - &mut &bytes[..], - ); + let outcome = stored.put(Pool::Attachments, &target, &owned, claimed, &mut &bytes[..]); drop(permit); outcome.map_err(|error| XrpcError::internal(error.to_string())) }) @@ -962,10 +982,7 @@ async fn knot_doc_is_ours( }) } -async fn await_knot_doc( - state: &XrpcState, - repo: &RepoDid, -) { +async fn await_knot_doc(state: &XrpcState, repo: &RepoDid) { for _ in 0..KNOT_DOC_POLLS { if knot_doc_is_ours(state, repo).await { return; @@ -983,11 +1000,14 @@ pub(crate) async fn get_json( url: Url, who: &str, ) -> Result { - let response = http.execute(HttpRequest::get(url)).await.map_err(|error| { - XrpcError::bad_gateway(format!("the {who} is unreachable: {error}")) - })?; + let response = http + .execute(HttpRequest::get(url)) + .await + .map_err(|error| XrpcError::bad_gateway(format!("the {who} is unreachable: {error}")))?; if response.status == StatusCode::TOO_MANY_REQUESTS { - return Err(XrpcError::throttled(format!("the {who} is rate-limiting the handover"))); + return Err(XrpcError::throttled(format!( + "the {who} is rate-limiting the handover" + ))); } if !response.status.is_success() { return Err(XrpcError::bad_gateway(format!( @@ -1206,7 +1226,13 @@ async fn drain_batch( foster_down = drain.foster_down, "the foster knot keeps answering with gateway errors, so the remaining handovers wait for the next boot" ); - return (folded.unfinished, Drain { parked: true, ..drain }); + return ( + folded.unfinished, + Drain { + parked: true, + ..drain + }, + ); }; tracing::error!( attempts = drain.attempts, diff --git a/knot2/crates/knot-xrpc/src/keep_sweep.rs b/knot2/crates/knot-xrpc/src/keep_sweep.rs index a69d14a7c..4a17011d0 100644 --- a/knot2/crates/knot-xrpc/src/keep_sweep.rs +++ b/knot2/crates/knot-xrpc/src/keep_sweep.rs @@ -96,20 +96,30 @@ pub(crate) async fn sweep_repo( "the keep sweep pinned a patch head" ); } - Err(error) => { + Err(PinError::Retry(error)) => { failed += 1; - tracing::warn!( - repo = %repo, - patch = %patch.address.rkey().as_str(), - head = %patch.head, - %error, - "the keep sweep couldn't pin a patch head, so it won't try that head again" - ); - remember_failure(state, repo, &patch) - .inspect_err(|error| { - tracing::error!(repo = %repo, %error, "the keep sweep couldn't record a failed head") - }) - .ok(); + tracing::warn!( + repo = %repo, + patch = %patch.address.rkey().as_str(), + head = %patch.head, + %error, + "the keep sweep couldn't pin a patch head, so it'll try again on the next boot" + ); + } + Err(PinError::Gone(error)) => { + failed += 1; + tracing::warn!( + repo = %repo, + patch = %patch.address.rkey().as_str(), + head = %patch.head, + %error, + "the keep sweep can never pin this head, so it won't try again" + ); + remember_failure(state, repo, &patch) + .inspect_err(|error| { + tracing::error!(repo = %repo, %error, "the keep sweep couldn't record a failed head") + }) + .ok(); } } } @@ -168,24 +178,33 @@ async fn unpinned( .await } +enum PinError { + Gone(XrpcError), + Retry(XrpcError), +} + async fn pin( state: &Arc>, repo: &RepoDid, patch: &Unpinned, -) -> Result { +) -> Result { let head = patch.head; let local = { let state = Arc::clone(state); let repo = repo.clone(); - run_blocking(move || Ok(state.layout.open(&repo)?.contains(head))).await? + run_blocking(move || Ok(state.layout.open(&repo)?.contains(head))) + .await + .map_err(PinError::Retry)? }; if !local { let Some(fork) = &patch.fork else { - return Err(XrpcError::not_found(format!( + return Err(PinError::Gone(XrpcError::not_found(format!( "the head {head} isn't in this repo and the patch has no fork to fetch it from" - ))); + )))); }; - crate::forks::fetch_commit(state, repo, fork, head).await?; + crate::forks::fetch_commit(state, repo, fork, head) + .await + .map_err(PinError::Retry)?; } let state = Arc::clone(state); let repo = repo.clone(); @@ -198,6 +217,7 @@ async fn pin( Ok(!local) }) .await + .map_err(PinError::Retry) } fn remember_failure( diff --git a/knot2/crates/knot-xrpc/src/lfs.rs b/knot2/crates/knot-xrpc/src/lfs.rs index 10447c96a..628d54bdf 100644 --- a/knot2/crates/knot-xrpc/src/lfs.rs +++ b/knot2/crates/knot-xrpc/src/lfs.rs @@ -281,27 +281,23 @@ async fn serve_batch( } let base = &state.knot_service_url; let objects: Result, Box> = match request.operation { - BatchOperation::Upload => match crate::receive::authorized_pusher( - state, - peer, - headers, - &repo, - PUSH_DENIED, - ) - .await - .map_err(|error| Box::new(challenge(error))) - { - Ok(_actor) => match probe_all(lfs, &repo, &request.objects).await { - Ok(stored) => request - .objects - .iter() - .zip(stored) - .map(|(object, present)| upload_action(base, path_prefix, object, present)) - .collect(), - Err(response) => Err(response), - }, - Err(response) => return *response, - }, + BatchOperation::Upload => { + match crate::receive::authorized_pusher(state, peer, headers, &repo, PUSH_DENIED) + .await + .map_err(|error| Box::new(challenge(error))) + { + Ok(_actor) => match probe_all(lfs, &repo, &request.objects).await { + Ok(stored) => request + .objects + .iter() + .zip(stored) + .map(|(object, present)| upload_action(base, path_prefix, object, present)) + .collect(), + Err(response) => Err(response), + }, + Err(response) => return *response, + } + } BatchOperation::Download => match probe_all(lfs, &repo, &request.objects).await { Ok(stored) => request .objects @@ -612,15 +608,10 @@ async fn serve_object_upload( let Ok(oid) = LfsOid::new(oid_raw.as_str()) else { return fail(StatusCode::NOT_FOUND, "object not found"); }; - let denied = crate::receive::authorized_pusher( - state, - peer, - request.headers(), - &repo, - PUSH_DENIED, - ) - .await - .map_err(|error| Box::new(challenge(error))); + let denied = + crate::receive::authorized_pusher(state, peer, request.headers(), &repo, PUSH_DENIED) + .await + .map_err(|error| Box::new(challenge(error))); if let Err(response) = denied { return *response; } diff --git a/knot2/crates/knot-xrpc/src/lib.rs b/knot2/crates/knot-xrpc/src/lib.rs index 818702b8b..dc6358510 100644 --- a/knot2/crates/knot-xrpc/src/lib.rs +++ b/knot2/crates/knot-xrpc/src/lib.rs @@ -84,8 +84,8 @@ use knot_resource::Slots; use knot_runtime::{Clock, Entropy, HttpTransport}; use knot_secrets::SealedStore; use knot_types::{ - AccountDid, AdmissionPolicy, AppviewEndpoint, ClonePath, ContributionPolicy, - KnotHostname, KnotId, KnotServiceUrl, Nsid, OwnerDid, OwnerRef, RepoDid, UnixSeconds, + AccountDid, AdmissionPolicy, AppviewEndpoint, ClonePath, ContributionPolicy, KnotHostname, + KnotId, KnotServiceUrl, Nsid, OwnerDid, OwnerRef, RepoDid, UnixSeconds, }; use base64::Engine; diff --git a/knot2/crates/knot-xrpc/src/members.rs b/knot2/crates/knot-xrpc/src/members.rs index 0a3307b29..436031bff 100644 --- a/knot2/crates/knot-xrpc/src/members.rs +++ b/knot2/crates/knot-xrpc/src/members.rs @@ -75,11 +75,13 @@ pub(crate) async fn commit_members( .await } +type SyncFn = fn(&Arc>, &Repo) -> Result; + pub(crate) async fn commit_grant_list( state: &Arc>, change: Authorized, refresh: fn(&Index) -> Result<(), IndexError>, - sync: fn(&Arc>, &Repo) -> Result, + sync: SyncFn, collection: GrantCollection, now: UnixSeconds, ) -> Result diff --git a/knot2/crates/knot-xrpc/src/plc.rs b/knot2/crates/knot-xrpc/src/plc.rs index cd17e04a6..70549da37 100644 --- a/knot2/crates/knot-xrpc/src/plc.rs +++ b/knot2/crates/knot-xrpc/src/plc.rs @@ -136,14 +136,18 @@ pub fn record_deletion( did: &RepoDid, deleted_at: UnixMicros, ) -> Result<(), XrpcError> { - edit_list(meta, &deleted_ref(), |accounts: &mut Vec| { - accounts.retain(|account| account.did != did.as_str()); - accounts.push(DeletedAccount { - did: did.as_str().to_owned(), - deleted_at_micros: deleted_at.get(), - }); - true - }) + edit_list( + meta, + &deleted_ref(), + |accounts: &mut Vec| { + accounts.retain(|account| account.did != did.as_str()); + accounts.push(DeletedAccount { + did: did.as_str().to_owned(), + deleted_at_micros: deleted_at.get(), + }); + true + }, + ) } pub fn deleted_dids(meta: &Repo) -> Result, XrpcError> { @@ -226,19 +230,18 @@ impl SweepMemory { } fn schedule(&mut self, did: &RepoDid, now_micros: UnixMicros) -> u32 { - let waiter = self - .waiting - .entry(did.clone()) - .or_insert(Waiter { - attempts: 0, - due_at_micros: UnixMicros::new(0), - }); + let waiter = self.waiting.entry(did.clone()).or_insert(Waiter { + attempts: 0, + due_at_micros: UnixMicros::new(0), + }); waiter.attempts = waiter.attempts.saturating_add(1); - waiter.due_at_micros = UnixMicros::new(now_micros.get().saturating_add( - BACKOFF_BASE_MICROS - .saturating_mul(1u64 << waiter.attempts.saturating_sub(1).min(7)) - .min(BACKOFF_CAP_MICROS), - )); + waiter.due_at_micros = UnixMicros::new( + now_micros.get().saturating_add( + BACKOFF_BASE_MICROS + .saturating_mul(1u64 << waiter.attempts.saturating_sub(1).min(7)) + .min(BACKOFF_CAP_MICROS), + ), + ); let attempts = waiter.attempts; if attempts >= MAX_ATTEMPTS { tracing::error!( diff --git a/knot2/crates/knot-xrpc/src/reads.rs b/knot2/crates/knot-xrpc/src/reads.rs index 9b25e90f8..074355d87 100644 --- a/knot2/crates/knot-xrpc/src/reads.rs +++ b/knot2/crates/knot-xrpc/src/reads.rs @@ -108,7 +108,9 @@ fn names_reserved(refspec: &str) -> bool { } pub(crate) fn warming_for(projection: &str) -> XrpcError { - XrpcError::warming(format!("{projection} projection is still warming, please retry shortly")) + XrpcError::warming(format!( + "{projection} projection is still warming, please retry shortly" + )) } pub(crate) fn warming() -> XrpcError { @@ -1494,11 +1496,7 @@ fn pinned_modified(modified_secs: i64) -> std::time::SystemTime { std::time::UNIX_EPOCH + std::time::Duration::from_secs(modified_secs.max(0) as u64) } -pub(crate) fn reconcile_if_range( - request: &mut Request, - etag: &str, - modified_secs: Option, -) { +pub(crate) fn reconcile_if_range(request: &mut Request, etag: &str, modified_secs: Option) { let Some(value) = request .headers() .get(header::IF_RANGE) @@ -1508,10 +1506,14 @@ pub(crate) fn reconcile_if_range( else { return; }; - let resumes = match (value.starts_with('"') || value.starts_with("W/"), modified_secs) { + let resumes = match ( + value.starts_with('"') || value.starts_with("W/"), + modified_secs, + ) { (true, _) => value.starts_with('"') && value == etag, - (false, Some(secs)) => httpdate::parse_http_date(&value) - .is_ok_and(|client| client == pinned_modified(secs)), + (false, Some(secs)) => { + httpdate::parse_http_date(&value).is_ok_and(|client| client == pinned_modified(secs)) + } (false, None) => false, }; let headers = request.headers_mut(); diff --git a/knot2/crates/knot-xrpc/src/social.rs b/knot2/crates/knot-xrpc/src/social.rs index adb1653e1..cd52811a5 100644 --- a/knot2/crates/knot-xrpc/src/social.rs +++ b/knot2/crates/knot-xrpc/src/social.rs @@ -14,8 +14,8 @@ use knot_cob::{ }; use knot_cobs::{ Artifact, Comment, Creation, Editing, Erasure, LabelDef, LabelDefChange, Manifest, - ManifestChange, Opening, Patch, Reaction, SocialBody, SocialChange, SocialCob, - SocialObject, Ticket, TicketChange, + ManifestChange, Opening, Patch, Reaction, SocialBody, SocialChange, SocialCob, SocialObject, + Ticket, TicketChange, }; use knot_git::{CommitRange, Repo}; use knot_index::{Resolved, SocialEntry}; @@ -44,8 +44,7 @@ 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}; @@ -77,8 +76,7 @@ pub(crate) const LIST_RECORD_VERSIONS_ROUTE: &str = "/xrpc/org.tangled.repo.list const NOT_STATE_AUTHORITY: &str = "State changes belong to the ticket's author or a maintainer"; const NOT_ASSIGN_AUTHORITY: &str = "Assignees belong to a maintainer"; -const STILL_MOVING: &str = - "This repository is still moving to this knot, so it can't take new issues or pulls yet. Please try again in a few minutes."; +const STILL_MOVING: &str = "This repository is still moving to this knot, so it can't take new issues or pulls yet. Please try again in a few minutes."; pub(crate) const COLLABORATE_REMEDY: &str = "Please ask the repo owner to add you as a collaborator, or to open the contribution policy to members or anyone, thanks."; @@ -377,6 +375,13 @@ impl MarkupInput { let unusable = |what: &'static str| { move |error| XrpcError::invalid_request(format!("{whose} {what} is unusable: {error}")) }; + let bare = + self.text.trim().is_empty() && self.blobs.as_ref().is_none_or(|blobs| blobs.is_empty()); + if bare { + return Err(XrpcError::invalid_request(format!( + "{whose} text is unusable: empty, with no attachments" + ))); + } let text = MarkdownBody::new(self.text).map_err(unusable("text"))?; Ok(CommentMarkup::new(text, self.blobs)) } @@ -1057,7 +1062,9 @@ pub(crate) fn convert_seeded( .collect::, XrpcError>>()?; let body = tree_body(&store, &seed_type, version.object, version.produces)? .ok_or_else(|| XrpcError::internal("a seeded record has no body"))?; - let taken = spent.entry(version.address.collection().clone()).or_default(); + let taken = spent + .entry(version.address.collection().clone()) + .or_default(); let number = match seeded_number .filter(|number| !taken.get(number).is_some_and(|at| at != &version.address)) { @@ -1141,7 +1148,9 @@ fn seeded_parent( body: &RecordBody, ) -> Result, XrpcError> { match address.collection() { - collection if *collection == comment_collection() || *collection == reaction_collection() => { + collection + if *collection == comment_collection() || *collection == reaction_collection() => + { let subject = knot_cobs::stored_subject(body).map_err(|error| { XrpcError::bad_gateway(format!( "the foster knot serves {} whose subject won't decode: {error}", @@ -1154,7 +1163,6 @@ fn seeded_parent( } } - pub(crate) fn reconcile( state: &Arc>, git: &Repo, @@ -1196,9 +1204,14 @@ pub(crate) fn reconcile( .filter(|collection| !collections.contains(collection)) .collect(); collections.extend(missing); + let erased: Vec = entries + .iter() + .filter(|entry| entry.state.erased()) + .map(|entry| entry.address.mst_key()) + .collect(); let unheld: Vec<_> = fostered .into_iter() - .filter(|(key, _, _)| !records.due.contains_key(key)) + .filter(|(key, _, _)| !records.due.contains_key(key) && !erased.contains(key)) .collect(); records = unheld .into_iter() @@ -1225,14 +1238,17 @@ pub(crate) fn reconcile( }) } +type FosteredDue = (MstKey, RecordCollection, chain::Due); +type FosteredState = (Vec, BTreeMap); +type FosteredGraph = (Option, BTreeMap); + fn fostered_state( store: &CobStore, git: &Repo, leaves: &BTreeMap, -) -> Result<(Vec<(MstKey, RecordCollection, chain::Due)>, BTreeMap), XrpcError> { +) -> Result { let seed_type = crate::foster::seed_type_name(); - let held = chain::tip(git)? - .map(|tip| knot_record::blocks::GitBlocks::at(git, tip.blocks)); + let held = chain::tip(git)?.map(|tip| knot_record::blocks::GitBlocks::at(git, tip.blocks)); let graphs = crate::foster::seeded_versions(store)? .into_iter() .map(|version| fostered_graph(store, &seed_type, held.as_ref(), leaves, version)) @@ -1253,16 +1269,15 @@ fn fostered_graph( held: Option<&knot_record::blocks::GitBlocks>, leaves: &BTreeMap, version: crate::foster::SeededVersion, -) -> Result< - ( - Option<(MstKey, RecordCollection, chain::Due)>, - BTreeMap, - ), - XrpcError, -> { +) -> Result { let key = version.address.mst_key(); let bodies: Vec = store - .changes_of(seed_type, version.object, knot_cob::HistoryModel::Convergent, None)? + .changes_of( + seed_type, + version.object, + knot_cob::HistoryModel::Convergent, + None, + )? .changes .into_iter() .filter_map(|change| change.record.into_body()) @@ -1609,6 +1624,7 @@ where .await } +#[allow(clippy::too_many_arguments)] async fn open_keyed( state: &Arc>, session: Session, @@ -1698,6 +1714,7 @@ fn vet_ticket_scope( Ok(()) } +#[allow(clippy::too_many_arguments)] fn ticket_body_of( prose: Prose, state: TicketState, @@ -1928,8 +1945,7 @@ async fn artifact_open( repo: RepoDid, record: serde_json::Value, ) -> Result { - let (oid, name, artifact) = - typed::(record, &artifact_collection())?.parse()?; + let (oid, name, artifact) = typed::(record, &artifact_collection())?.parse()?; let session = crate::blobs::session_admitting_artifact(state, headers, method, repo, &artifact).await?; let record = @@ -2523,14 +2539,12 @@ async fn label_def_open( record: serde_json::Value, ) -> Result { let collection = label_def_collection(); - let slug = rkey - .and_then(|rkey| rkey.as_label_key()) - .ok_or_else(|| { - XrpcError::invalid_request( - "a label def's record key is the slug the owner chooses; send that slug \ + let slug = rkey.and_then(|rkey| rkey.as_label_key()).ok_or_else(|| { + XrpcError::invalid_request( + "a label def's record key is the slug the owner chooses; send that slug \ as the rkey", - ) - })?; + ) + })?; let def = typed::(record, &collection)?.parse(&collection)?; let session = session(state, headers, method, repo).await?; if !Opening::Ownership.decide(session.permission).is_allowed() { diff --git a/knot2/crates/knot-xrpc/src/tests.rs b/knot2/crates/knot-xrpc/src/tests.rs index bc2479192..3833f6cb3 100644 --- a/knot2/crates/knot-xrpc/src/tests.rs +++ b/knot2/crates/knot-xrpc/src/tests.rs @@ -2548,7 +2548,10 @@ async fn assignees_belong_to_a_maintainer_and_round_trip() { let rejected = failed( ticket - .create(&world.stranger, assigned("kelp", json!([world.stranger.did]))) + .create( + &world.stranger, + assigned("kelp", json!([world.stranger.did])), + ) .await, StatusCode::FORBIDDEN, ); @@ -6235,8 +6238,15 @@ async fn a_sweep_as_wide_as_the_candidates_lands_every_genesis_in_one_pass() { .await, ) .status(); - assert_eq!(status, StatusCode::OK, "{name} registers despite the outage"); - match state.index.resolve_repo(&owner, &RepoRkey::new(name).unwrap()) { + assert_eq!( + status, + StatusCode::OK, + "{name} registers despite the outage" + ); + match state + .index + .resolve_repo(&owner, &RepoRkey::new(name).unwrap()) + { Resolved::Ready(Some(did)) => dids.push(did), other => panic!("Repo wasn't registered despite the queue: {other:?}"), } diff --git a/knot2/crates/knot-xrpc/tests/common/mod.rs b/knot2/crates/knot-xrpc/tests/common/mod.rs index 856e92954..54ad12dba 100644 --- a/knot2/crates/knot-xrpc/tests/common/mod.rs +++ b/knot2/crates/knot-xrpc/tests/common/mod.rs @@ -659,7 +659,12 @@ impl World { 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), + &pds_of + .lock() + .unwrap() + .get(&did) + .cloned() + .unwrap_or_else(default_pds), knot.as_deref(), ) } @@ -786,8 +791,12 @@ impl World { &state.knot_service_url, ); let router = match fostering { - true => knot_xrpc::router(Arc::clone(&state)) - .merge(knot_xrpc::foster::router(Arc::clone(&state), false, false, liar)), + true => knot_xrpc::router(Arc::clone(&state)).merge(knot_xrpc::foster::router( + Arc::clone(&state), + false, + false, + liar, + )), false => knot_xrpc::router(Arc::clone(&state)), } .route( diff --git a/knot2/crates/knot-xrpc/tests/handover.rs b/knot2/crates/knot-xrpc/tests/handover.rs index 146896638..495cdeb4a 100644 --- a/knot2/crates/knot-xrpc/tests/handover.rs +++ b/knot2/crates/knot-xrpc/tests/handover.rs @@ -352,7 +352,11 @@ async fn an_upgrading_knot_pulls_its_records_back_and_goes_green() { !local_cob_refs(&upgrade).is_empty(), "the cob refs came over with the objects" ); - assert_eq!(sink.len(), 2, "a transiently rejected ack is retried in the session, which still ends green"); + assert_eq!( + sink.len(), + 2, + "a transiently rejected ack is retried in the session, which still ends green" + ); let ack = &sink.bodies.lock()[0]; assert_eq!(ack["repoDid"], REPO); assert_eq!(ack["records"], 1); @@ -476,7 +480,11 @@ async fn a_repo_the_foster_knot_doesnt_host_settles_without_an_ack() { .exists(), "an real, non-shadowed repo is happily in the right place forever, like never queried for again" ); - assert_eq!(sink.len(), 0, "lack of records returned back to us, so we don't ack"); + assert_eq!( + sink.len(), + 0, + "lack of records returned back to us, so we don't ack" + ); let git = upgrade.layout.open(&repo_did()).unwrap(); assert_eq!( CobStore::new(&git).cobs().unwrap().len(), @@ -664,7 +672,11 @@ async fn empty_seed_still_fosters_manifest() { let (status, body) = post_seed(&foster, serde_json::json!([])).await; 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"); + assert_eq!( + body["results"], + serde_json::json!([]), + "no records, no results" + ); } #[tokio::test] @@ -718,7 +730,11 @@ async fn a_social_write_onto_a_shadow_repo_is_refused() { } async fn ticketed_pair(refuse_first: usize) -> (World, std::net::SocketAddr, AckSink, World) { - handover_pair(serde_json::json!([ticket_record("a feral pearl")]), refuse_first).await + handover_pair( + serde_json::json!([ticket_record("a feral pearl")]), + refuse_first, + ) + .await } fn ticket_path() -> String { @@ -726,7 +742,10 @@ fn ticket_path() -> String { } async fn ticket_cid(world: &World) -> String { - common::get_json(world, &ticket_path()).await["cid"].as_str().unwrap().to_string() + common::get_json(world, &ticket_path()).await["cid"] + .as_str() + .unwrap() + .to_string() } async fn delete_ticket(world: &World) { @@ -781,7 +800,9 @@ async fn deleted_pulled_ticket_stays_deleted_through_reconcile() { 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(); + 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; @@ -793,39 +814,58 @@ async fn deleted_pulled_ticket_stays_deleted_through_reconcile() { async fn delete_during_crashed_handover_doesnt_wedge_replay() { let (_foster, foster_addr, sink, upgrade) = ticketed_pair(3).await; let first = trigger(&upgrade, foster_addr).await; - assert_eq!(first["state"], "running", "the rejected ack leaves it pending"); + 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!( + second["state"], "done", + "the replay converges past the delete: {second}" + ); assert_eq!(sink.len(), 4, "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"); + 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(3).await; let first = trigger(&upgrade, foster_addr).await; - assert_eq!(first["state"], "running", "the rejected ack leaves it pending"); + 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!( + second["state"], "done", + "the replay goes green over the edit: {second}" + ); assert_eq!(sink.len(), 4, "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"); + 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"); + assert_ne!( + fetched["cid"], shadow["cid"], + "the origin serves the edit, not the shadow's version" + ); } #[tokio::test] @@ -836,9 +876,14 @@ async fn finished_handover_drops_seed_refs() { 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"); + 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) { @@ -885,11 +930,20 @@ async fn a_regressed_foster_ref_cant_pull_the_local_ref_backwards() { std::fs::remove_file(settled_path(&upgrade, &repo_did())).unwrap(); let outcome = trigger(&upgrade, foster_addr).await; assert_eq!(outcome["state"], "running", "{outcome}"); - 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"); + 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 { @@ -1233,7 +1287,8 @@ async fn un_acked_shadow_will_mint_n_erase_via_admin() { ) .await; assert_eq!( - status, StatusCode::OK, + status, + StatusCode::OK, "re-mint must succeed: {}", String::from_utf8_lossy(&raw) ); @@ -1263,7 +1318,8 @@ async fn un_acked_shadow_will_mint_n_erase_via_admin() { ) .await; assert_eq!( - status, StatusCode::OK, + status, + StatusCode::OK, "admin-erase must succeed: {}", String::from_utf8_lossy(&raw) ); @@ -1271,7 +1327,8 @@ async fn un_acked_shadow_will_mint_n_erase_via_admin() { assert!(body["results"][0]["error"].is_null(), "{body}"); let (status, _, gone) = common::get(&world, &ticket_path()).await; assert_eq!( - status, StatusCode::BAD_REQUEST, + status, + StatusCode::BAD_REQUEST, "erased record no longer served: {}", String::from_utf8_lossy(&gone) ); @@ -1283,7 +1340,8 @@ async fn un_acked_shadow_will_mint_n_erase_via_admin() { assert_eq!(failed, 0, "boot pass reconciles state"); let (status, _, _) = common::get(&world, &ticket_path()).await; assert_eq!( - status, StatusCode::BAD_REQUEST, + status, + StatusCode::BAD_REQUEST, "restart doesn't resurrect the erased record" ); let bounds = knot_events::ReplayBounds::new( diff --git a/knot2/crates/knot-xrpc/tests/retire.rs b/knot2/crates/knot-xrpc/tests/retire.rs index 79150aac9..ade57f143 100644 --- a/knot2/crates/knot-xrpc/tests/retire.rs +++ b/knot2/crates/knot-xrpc/tests/retire.rs @@ -1,6 +1,6 @@ mod common; -use base64::{engine::general_purpose::URL_SAFE_NO_PAD, Engine as _}; +use base64::{Engine as _, engine::general_purpose::URL_SAFE_NO_PAD}; use common::{World, basic_auth, get_json, serve_router}; use http::StatusCode; use knot_cob::CobStore; @@ -57,7 +57,7 @@ fn json_get(payload: serde_json::Value) -> axum::routing::MethodRouter { }) } -fn knot_social_origin(signer: &K256Signer) -> axum::Router { +fn knot_origin(signer: &K256Signer, capabilities: &[&str]) -> axum::Router { let document = knot_atproto::knot_did_document( &KnotId::new(ORIGIN_DID).unwrap(), &signer.public_key(), @@ -66,18 +66,27 @@ fn knot_social_origin(signer: &K256Signer) -> axum::Router { axum::Router::new() .route( "/xrpc/org.tangled.knot.describeKnot", - json_get(serde_json::json!({"did": ORIGIN_DID, "capabilities": ["knot-social-writes"]})), + json_get(serde_json::json!({"did": ORIGIN_DID, "capabilities": capabilities})), ) .route("/.well-known/did.json", json_get(document)) } +fn knot_social_origin(signer: &K256Signer) -> axum::Router { + knot_origin(signer, &["knot-social-writes"]) +} + async fn serve_origin(world: &World, signer: &K256Signer) -> String { - let path = format!("/xrpc/com.atproto.repo.getRecord?repo={REPO}&collection={COLLECTION}&rkey={RKEY}"); + 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()}))); + 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) } @@ -92,7 +101,12 @@ struct AckSignedView<'a> { } 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 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, @@ -110,7 +124,9 @@ fn retired_at(world: &World) -> Option { 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)) + registry + .record_of(&repo_did()) + .and_then(|record| record.retired_at.map(UnixSeconds::get)) } async fn seeded_origin() -> (World, K256Signer, String) { @@ -128,7 +144,11 @@ async fn valid_ack_retires_shadow() { 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"); + assert_eq!( + retired_at(&foster), + Some(1100), + "retiredAt follows the ack's ackedAt" + ); } #[tokio::test] @@ -137,7 +157,11 @@ async fn retired_shadow_returns_409_to_further_seeds() { 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; + let (status, body) = post_seed( + &foster, + serde_json::json!([ticket_record("a second pearl")]), + ) + .await; assert_eq!(status, StatusCode::CONFLICT, "{body}"); } @@ -147,8 +171,16 @@ async fn forged_ack_rejected() { 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"); + 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] @@ -159,7 +191,11 @@ async fn plain_http_origin_off_loopback_rejected() { 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"); + assert_eq!( + retired_at(&foster), + None, + "no retire over plain http off loopback" + ); } #[tokio::test] @@ -171,7 +207,11 @@ async fn data_url_origin_rejected() { 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"); + assert_eq!( + retired_at(&foster), + None, + "a data url origin doesn't retire the shadow or panic the knot" + ); } #[tokio::test] @@ -180,7 +220,11 @@ async fn pre_epoch_ack_rejected() { 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"); + assert_eq!( + retired_at(&foster), + None, + "a pre-epoch ackedAt can't retire anything" + ); } #[tokio::test] @@ -188,11 +232,23 @@ 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}"); + 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"); + 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] @@ -201,23 +257,63 @@ async fn replay_after_retire_returns_200() { 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}"); + 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"); + 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() { +async fn seed_from_knot_social_origin_is_welcome() { 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}"); + let (status, body) = + post_seed(&foster, serde_json::json!([ticket_record("a feral pearl")])).await; + assert_eq!( + status, + StatusCode::OK, + "the PLC flip can point at the knot-social origin before the seed drain ends, so \ + seeding mustn't mind it: {body}" + ); +} + +#[tokio::test] +async fn ack_from_origin_without_cobs_is_refused() { + let foster = World::fostering_with_grace(FOSTER_SECRET); + seed_foster_record(&foster).await; + let signer = K256Signer::generate(&OsEntropy); + let origin = serve_router(knot_origin(&signer, &[])).await; + foster.publish_plc_knot(REPO, &format!("http://{origin}")); + + let (status, body) = post_ack(&foster, signed_ack(&signer, 1100)).await; + assert_eq!(status, StatusCode::BAD_REQUEST, "{body}"); + assert!( + body["message"] + .as_str() + .is_some_and(|message| message.contains("doesn't serve COBs")), + "the retire check lives at the ack, not the seed: {body}" + ); + assert_eq!( + retired_at(&foster), + None, + "a knot that can't serve the records can't retire the shadow" + ); } #[tokio::test] @@ -225,7 +321,11 @@ 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}"); + 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, &None).await; @@ -234,11 +334,19 @@ async fn gc_takes_retired_repo_past_grace_window() { 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"); + 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"); + assert!( + !tombstones.contains(&repo_did()), + "the repo is alive on the origin, so the retire GC writes no PLC tombstone" + ); } #[derive(Default)] @@ -251,23 +359,21 @@ fn liar_hydrant(calls: &std::sync::Arc) -> axum::Router { let calls = std::sync::Arc::clone(calls); axum::Router::new().route( "/xrpc/org.tangled.foster.adminUnmarkLiarAccount", - axum::routing::post( - move |headers: http::HeaderMap, body: String| { - let calls = std::sync::Arc::clone(&calls); - async move { - let auth = headers - .get(http::header::AUTHORIZATION) - .and_then(|value| value.to_str().ok()) - .map(str::to_owned); - let payload = serde_json::from_str(&body).unwrap_or(serde_json::Value::Null); - calls.seen.lock().unwrap().push((auth, payload)); - match calls.failing.load(std::sync::atomic::Ordering::SeqCst) { - true => (StatusCode::INTERNAL_SERVER_ERROR, String::new()), - false => (StatusCode::OK, r#"{"unmarked":1}"#.to_owned()), - } + axum::routing::post(move |headers: http::HeaderMap, body: String| { + let calls = std::sync::Arc::clone(&calls); + async move { + let auth = headers + .get(http::header::AUTHORIZATION) + .and_then(|value| value.to_str().ok()) + .map(str::to_owned); + let payload = serde_json::from_str(&body).unwrap_or(serde_json::Value::Null); + calls.seen.lock().unwrap().push((auth, payload)); + match calls.failing.load(std::sync::atomic::Ordering::SeqCst) { + true => (StatusCode::INTERNAL_SERVER_ERROR, String::new()), + false => (StatusCode::OK, r#"{"unmarked":1}"#.to_owned()), } - }, - ), + } + }), ) } @@ -329,7 +435,10 @@ async fn ack_releases_the_liar_resolution_and_gc_wipes_the_shadow() { foster.advance(std::time::Duration::from_secs(3_601)); knot_xrpc::foster::gc_tick(&foster.state, &Some(liar)).await; - assert!(!registry_still_serves(&foster), "the unmarked shadow is deregistered"); + assert!( + !registry_still_serves(&foster), + "the unmarked shadow is deregistered" + ); assert!( !foster.layout.repo_path(&repo_did()).unwrap().exists(), "the unmarked shadow leaves disk" @@ -350,7 +459,10 @@ async fn gc_keeps_the_shadow_until_the_liar_takes_the_unmark() { foster.advance(std::time::Duration::from_secs(3_601)); knot_xrpc::foster::gc_tick(&foster.state, &Some(liar.clone())).await; - assert!(registry_still_serves(&foster), "a shadow the liar still points at survives its grace window"); + assert!( + registry_still_serves(&foster), + "a shadow the liar still points at survives its grace window" + ); assert!( foster.layout.repo_path(&repo_did()).unwrap().exists(), "a shadow the liar still points at stays on disk" @@ -360,8 +472,13 @@ async fn gc_keeps_the_shadow_until_the_liar_takes_the_unmark() { "the ack and the sweep both asked the liar hydrant" ); - calls.failing.store(false, std::sync::atomic::Ordering::SeqCst); + calls + .failing + .store(false, std::sync::atomic::Ordering::SeqCst); knot_xrpc::foster::gc_tick(&foster.state, &Some(liar)).await; - assert!(!registry_still_serves(&foster), "the shadow goes once the liar takes the unmark"); + assert!( + !registry_still_serves(&foster), + "the shadow goes once the liar takes the unmark" + ); } diff --git a/knot2/example.toml b/knot2/example.toml index c17146954..2ce34a775 100644 --- a/knot2/example.toml +++ b/knot2/example.toml @@ -498,14 +498,7 @@ # Default value: 4 #foster_boot_width = 4 -# Takes a handover ack even when the repo's DID document declares no -# tangled_knot service endpoint, for the one-time 2026-10-28 upgrade, so -# a slow PLC-ops tail can't stall the foster retire. Such an ack skips -# the origin knot's signature and record checks, so leave this off -# outside that upgrade. -# # Can also be specified via environment variable `KNOT_SOCIAL_FOSTER_ACK_SKIP_KNOT_SERVICE`. -# # Default value: false #foster_ack_skip_knot_service = false diff --git a/knotmirror/crates/knotmirror-index/src/lib.rs b/knotmirror/crates/knotmirror-index/src/lib.rs index 008878a36..6db5f2e5f 100644 --- a/knotmirror/crates/knotmirror-index/src/lib.rs +++ b/knotmirror/crates/knotmirror-index/src/lib.rs @@ -3,10 +3,10 @@ use std::sync::Arc; use std::time::{Duration, Instant}; use bobbin_runtime::Clock; +use jacquard_common::types::did::Did; use knotmirror_db::{Db, DbError, IndexRow}; use knotmirror_git::layout::Layout; use knotmirror_types::{backoff_us, now_us}; -use jacquard_common::types::did::Did; use thiserror::Error; use tokio::sync::{Notify, Semaphore}; use tokio_util::sync::CancellationToken; diff --git a/knotmirror/crates/knotmirror-ingest/src/lib.rs b/knotmirror/crates/knotmirror-ingest/src/lib.rs index 4d9c99c90..7921a1a3d 100644 --- a/knotmirror/crates/knotmirror-ingest/src/lib.rs +++ b/knotmirror/crates/knotmirror-ingest/src/lib.rs @@ -4,10 +4,10 @@ use std::sync::Arc; use std::time::Duration; use bobbin_runtime::{Clock, WsConn, WsMessage, WsTransport}; -use knotmirror_db::{Db, DbError, GitRefs, RefsUpsert}; use jacquard_common::types::collection::Collection as _; use jacquard_common::types::did::Did; use jacquard_common::types::tid::Tid; +use knotmirror_db::{Db, DbError, GitRefs, RefsUpsert}; use lexicons::org_tangled::git::refs::{Refs, RefsRecord}; use serde::Deserialize; use serde_json::value::RawValue; diff --git a/knotmirror/crates/knotmirror-maintenance/src/lib.rs b/knotmirror/crates/knotmirror-maintenance/src/lib.rs index 9548bb160..c253ee4cc 100644 --- a/knotmirror/crates/knotmirror-maintenance/src/lib.rs +++ b/knotmirror/crates/knotmirror-maintenance/src/lib.rs @@ -4,11 +4,11 @@ use std::sync::Arc; use std::time::{Duration, Instant}; use bobbin_runtime::Clock; -use knotmirror_db::Db; -use knotmirror_git::layout::Layout; use jacquard_common::types::did::Did; use knot_maintenance::{RepackStatus, Report}; use knot_types::UnixSeconds; +use knotmirror_db::Db; +use knotmirror_git::layout::Layout; use tokio_util::sync::CancellationToken; use tracing::{error, info, warn}; diff --git a/knotmirror/crates/knotmirror-sync/src/lib.rs b/knotmirror/crates/knotmirror-sync/src/lib.rs index 5f2a87600..8ae300e69 100644 --- a/knotmirror/crates/knotmirror-sync/src/lib.rs +++ b/knotmirror/crates/knotmirror-sync/src/lib.rs @@ -6,10 +6,10 @@ use std::time::{Duration, Instant}; use bobbin_runtime::Clock; use crate::admission::BackfillAdmission; +use jacquard_common::types::did::Did; use knotmirror_db::{Db, DbError}; use knotmirror_git::layout::Layout; use knotmirror_types::{KnotPolicy, backoff_us, now_us}; -use jacquard_common::types::did::Did; use metrics::{gauge, histogram}; use thiserror::Error; use tokio::sync::{Notify, Semaphore}; diff --git a/knotmirror/crates/knotmirror-sync/src/task.rs b/knotmirror/crates/knotmirror-sync/src/task.rs index 5832182d6..61e8c4a67 100644 --- a/knotmirror/crates/knotmirror-sync/src/task.rs +++ b/knotmirror/crates/knotmirror-sync/src/task.rs @@ -1,14 +1,14 @@ use std::path::Path; use std::sync::atomic::AtomicBool; -use knotmirror_db::RepoRow; -use knotmirror_git::layout::Layout; -use knotmirror_types::jacquard_ext::DidDocumentExt as _; -use knotmirror_types::{KnotHost, KnotPolicy}; use gix::refs::FullName; use gix_hash::ObjectId; use jacquard_common::types::did::Did; use jacquard_identity::resolver::IdentityResolver as _; +use knotmirror_db::RepoRow; +use knotmirror_git::layout::Layout; +use knotmirror_types::jacquard_ext::DidDocumentExt as _; +use knotmirror_types::{KnotHost, KnotPolicy}; use thiserror::Error; use tracing::debug; diff --git a/knotmirror/crates/knotmirror-xrpc/src/lib.rs b/knotmirror/crates/knotmirror-xrpc/src/lib.rs index 2e289bc23..2c9e8e06e 100644 --- a/knotmirror/crates/knotmirror-xrpc/src/lib.rs +++ b/knotmirror/crates/knotmirror-xrpc/src/lib.rs @@ -19,10 +19,10 @@ use axum::{ routing::{get, post}, }; use bobbin_runtime::{Clock, ReqwestHttp}; -use knotmirror_git::layout::Layout; use jacquard_axum::service_auth::{self, ServiceAuthError}; use jacquard_common::service_auth::ServiceAuthClaims; use jacquard_identity::JacquardResolver; +use knotmirror_git::layout::Layout; use tangled_axum::atproto::{AtprotoService, ClaimsPolicy, Verified}; use crate::committers::Committers; diff --git a/knotmirror/crates/knotmirror-xrpc/src/routes/git.rs b/knotmirror/crates/knotmirror-xrpc/src/routes/git.rs index 7be89bf8e..4c9fb7318 100644 --- a/knotmirror/crates/knotmirror-xrpc/src/routes/git.rs +++ b/knotmirror/crates/knotmirror-xrpc/src/routes/git.rs @@ -4,7 +4,6 @@ use std::time::{Duration, Instant}; use axum::extract::State; use bobbin_knot_proxy::KnotHost; -use knotmirror_git::repo_ext::{MergeError, RepositoryExt as _}; use gix::ObjectId; use gix::bstr::ByteSlice as _; use gix::revision::plumbing::Spec as RevSpec; @@ -16,6 +15,7 @@ use jacquard_common::service_auth::ServiceAuthClaims; use jacquard_common::types::did::Did; use jacquard_common::types::string::Datetime; use jacquard_identity::resolver::IdentityResolver as _; +use knotmirror_git::repo_ext::{MergeError, RepositoryExt as _}; use lexicons::org_tangled::temp::git::{ delete_branch, list_commits, list_commits_by_ref, sync_fork, }; @@ -1217,8 +1217,8 @@ fn build_pack( #[cfg(test)] mod diff_tests { use super::*; - use knotmirror_git::layout::Layout; use jacquard_common::types::did::Did; + use knotmirror_git::layout::Layout; use std::path::Path; use std::process::Command; diff --git a/knotmirror/crates/knotmirror/src/config.rs b/knotmirror/crates/knotmirror/src/config.rs index 03f3d2288..de243b118 100644 --- a/knotmirror/crates/knotmirror/src/config.rs +++ b/knotmirror/crates/knotmirror/src/config.rs @@ -233,7 +233,10 @@ pub struct MaintenanceConfig { pub interval_secs: u64, /// Repos whose reachable set exceeds this many objects are not repacked. - #[config(env = "KNOTMIRROR_MAINTENANCE_REPACK_MAX_OBJECTS", default = 16_000_000)] + #[config( + env = "KNOTMIRROR_MAINTENANCE_REPACK_MAX_OBJECTS", + default = 16_000_000 + )] pub repack_max_objects: u64, #[config(env = "KNOTMIRROR_MAINTENANCE_REPACK_GEOMETRIC_FACTOR", default = 2)] -- knotmirror 0.0.1