diff --git a/knot2/crates/knot-index/Cargo.toml b/knot2/crates/knot-index/Cargo.toml index 58871831d..10f6e57b5 100644 --- a/knot2/crates/knot-index/Cargo.toml +++ b/knot2/crates/knot-index/Cargo.toml @@ -12,6 +12,7 @@ test-support = [] knot-types = { workspace = true } knot-git = { workspace = true } knot-cob = { workspace = true } +knot-resource = { workspace = true } knot-cobs = { workspace = true } scc = { workspace = true } lasso = { workspace = true } @@ -24,3 +25,4 @@ knot-runtime = { workspace = true } tempfile = { workspace = true } proptest = { workspace = true } serde = { workspace = true } +serde_ipld_dagcbor = { workspace = true } diff --git a/knot2/crates/knot-index/src/consent.rs b/knot2/crates/knot-index/src/consent.rs index aa95ac926..ce9a509b9 100644 --- a/knot2/crates/knot-index/src/consent.rs +++ b/knot2/crates/knot-index/src/consent.rs @@ -646,7 +646,7 @@ mod tests { assert_eq!( checks(&index, &leaving, consent, 30), Consented::Stale, - "the readings map can't outgrow the roster, so a dropped subject resolves again" + "dropped subject resolves again, keeping readings map within roll's size" ); } } diff --git a/knot2/crates/knot-index/src/intern.rs b/knot2/crates/knot-index/src/intern.rs index 3b74f8019..547157c20 100644 --- a/knot2/crates/knot-index/src/intern.rs +++ b/knot2/crates/knot-index/src/intern.rs @@ -1,6 +1,6 @@ use std::sync::Arc; -use knot_types::{AccountDid, OwnerDid, RepoDid, RepoName, RepoRkey}; +use knot_types::{AccountDid, OwnerDid, RecordCollection, RecordRkey, RepoDid, RepoName, RepoRkey}; use lasso::{Spur, ThreadedRodeo}; #[derive(Debug, Clone, Default)] @@ -51,4 +51,6 @@ interned! { OwnerKey of OwnerDid { intern_owner, owner, resolve_owner, "owner DID" } RkeyKey of RepoRkey { intern_rkey, rkey, resolve_rkey, "rkey" } NameKey of RepoName { intern_name, name } + CollectionKey of RecordCollection { intern_collection, collection, resolve_collection, "record collection" } + RecordKey of RecordRkey { intern_record_rkey, record_rkey, resolve_record_rkey, "record key" } } diff --git a/knot2/crates/knot-index/src/lib.rs b/knot2/crates/knot-index/src/lib.rs index c9cf122b3..d8dc09584 100644 --- a/knot2/crates/knot-index/src/lib.rs +++ b/knot2/crates/knot-index/src/lib.rs @@ -3,6 +3,7 @@ mod coverage; mod error; mod intern; mod projections; +mod social; pub use consent::{ Acceptance, AcceptanceGrace, AcceptanceTtl, Acceptances, Answer, CollaborationConsent, @@ -11,19 +12,24 @@ pub use consent::{ pub use coverage::{Coverage, Resolved}; pub use error::IndexError; pub use knot_types::OfferedKey; +pub use social::SocialResidency; use std::path::PathBuf; use std::sync::atomic::{AtomicU64, Ordering}; use std::time::Duration; -use knot_cob::{ChangePayload, CobStore}; +use knot_cob::{ChangePayload, CobId, CobStore}; use knot_cobs::{ BlocklistChange, BlocklistCob, CollaboratorsChange, CollaboratorsCob, MembersChange, MembersCob, RegistryChange, RepoRef, RepoRegistryCob, }; -pub use knot_cobs::{EffectiveSince, Entry, Offer, Standing}; +pub use knot_cobs::{EffectiveSince, Entry, EntryState, Offer}; use knot_git::{Layout, Repo}; -use knot_types::{AccountDid, ClonePath, OwnerDid, RepoDid, RepoRkey, UnixSeconds}; +use knot_resource::StructureBytes; +use knot_types::{ + AccountDid, ClonePath, OwnerDid, RecordAddress, RecordCid, RecordCollection, RepoDid, + RepoPolicy, RepoRkey, SubjectNumber, TypeName, UnixSeconds, +}; use tokio::sync::watch; use consent::ConsentProjection; @@ -125,9 +131,9 @@ impl Folded { } #[derive(Debug, Clone, Copy, PartialEq, Eq)] -pub struct SweepFloor(Duration); +pub struct SweepInterval(Duration); -impl SweepFloor { +impl SweepInterval { pub const DEFAULT: Self = Self::from_secs(60); pub const fn from_secs(secs: u64) -> Self { @@ -322,6 +328,7 @@ pub struct Index { blocklist: GrantSetProjection, collaborators: CollaboratorsProjection, registry: RegistryProjection, + social: social::SocialProjection, keys: KeyProjection, consents: ConsentProjection, freshness: Freshness, @@ -348,6 +355,7 @@ impl Index { blocklist: GrantSetProjection::new(), collaborators: CollaboratorsProjection::new(), registry: RegistryProjection::new(), + social: social::SocialProjection::new(SocialResidency::DEFAULT), keys: KeyProjection::new(budget), consents: ConsentProjection::new(), freshness: Freshness::DEFAULT, @@ -362,6 +370,11 @@ impl Index { self } + pub fn with_social_residency(mut self, residency: SocialResidency) -> Self { + self.social = social::SocialProjection::new(residency); + self + } + pub fn generation(&self) -> IndexGeneration { IndexGeneration(self.generation.load(Ordering::Acquire)) } @@ -444,6 +457,11 @@ impl Index { } pub fn refresh_registry(&self) -> Result<(), IndexError> { + self.registry_refresh() + .inspect_err(|_| self.registry.warm()) + } + + fn registry_refresh(&self) -> Result<(), IndexError> { let meta = Repo::open(&self.meta_path)?; let store = CobStore::new(&meta); let evacuated = match store.list::()?.as_slice() { @@ -459,6 +477,7 @@ impl Index { evacuated.iter().for_each(|repo| { if let Some(key) = self.interner.repo(repo) { self.collaborators.drop_repo(key); + self.social.drop_repo(key); self.consents.forget_repo(key); self.unreadable.remove_sync(&key); } @@ -509,13 +528,107 @@ impl Index { Ok(()) } + pub fn ensure_social(&self, repo: &RepoDid) -> Result<(), IndexError> { + if self.social.is_folded(&self.interner, repo) { + return Ok(()); + } + self.refresh_social(repo) + } + + pub fn refresh_social(&self, repo: &RepoDid) -> Result<(), IndexError> { + let git = self.layout.open(repo)?; + let store = CobStore::new(&git); + let repo_key = self.interner.intern_repo(repo); + self.social + .refresh_repo(&self.interner, &store, repo_key) + .inspect(|()| self.bump_generation()) + } + + pub fn refresh_social_object( + &self, + repo: &RepoDid, + type_name: &TypeName, + object: CobId, + ) -> Result<(), IndexError> { + let git = self.layout.open(repo)?; + let store = CobStore::new(&git); + let repo_key = self.interner.intern_repo(repo); + self.social + .refresh_object(&self.interner, &store, repo_key, type_name, object) + .inspect(|()| self.bump_generation()) + } + + pub fn social_object_at( + &self, + repo: &RepoDid, + address: &RecordAddress, + ) -> Resolved> { + self.social.object_at(&self.interner, repo, address) + } + + pub fn social_address_of( + &self, + repo: &RepoDid, + object: CobId, + ) -> Resolved> { + self.social.address_of(&self.interner, repo, object) + } + + pub fn social_of_kind( + &self, + repo: &RepoDid, + collection: &RecordCollection, + ) -> Resolved> { + self.social.of_kind(&self.interner, repo, collection) + } + + pub fn social_children_of( + &self, + repo: &RepoDid, + parent: &RecordAddress, + ) -> Resolved> { + self.social.children_of(&self.interner, repo, parent) + } + + pub fn social_latest_child( + &self, + repo: &RepoDid, + parent: &RecordAddress, + collection: &RecordCollection, + ) -> Resolved> { + self.social + .latest_child(&self.interner, repo, parent, collection) + } + + pub fn social_author_of(&self, repo: &RepoDid, object: CobId) -> Resolved> { + self.social.author_of(&self.interner, repo, object) + } + + pub fn social_version_of(&self, repo: &RepoDid, object: CobId) -> Resolved> { + self.social.version_of(&self.interner, repo, object) + } + + pub fn next_subject_number( + &self, + repo: &RepoDid, + collection: &RecordCollection, + ) -> Resolved> { + self.social + .highest_number(&self.interner, repo, collection) + .map(social::next_number) + } + + pub fn social_structure(&self, repo: &RepoDid) -> Resolved { + self.social.structure(&self.interner, repo) + } + pub fn effective_member(&self, did: &AccountDid) -> Resolved { - self.member_standing(did) - .map(|standing| standing.is_some_and(Standing::is_effective)) + self.member_state(did) + .map(|state| state.is_some_and(EntryState::is_effective)) } - pub fn member_standing(&self, did: &AccountDid) -> Resolved> { - self.members.slot(&self.interner, did, Provenance::standing) + pub fn member_state(&self, did: &AccountDid) -> Resolved> { + self.members.slot(&self.interner, did, Provenance::state) } pub fn member_entries(&self) -> Resolved> { @@ -524,20 +637,20 @@ impl Index { pub fn member_entries_where( &self, - keep: impl Fn(Standing) -> bool, + keep: impl Fn(EntryState) -> bool, ) -> Resolved> { self.members.entries_where(&self.interner, keep) } pub fn effective_members(&self) -> Resolved> { self.members - .subjects_where(&self.interner, Standing::is_effective) + .subjects_where(&self.interner, EntryState::is_effective) } pub fn is_blocked(&self, did: &AccountDid) -> Resolved { self.blocklist - .slot(&self.interner, did, Provenance::standing) - .map(|standing| standing.is_some()) + .slot(&self.interner, did, Provenance::state) + .map(|state| state.is_some()) } pub fn blocked(&self) -> Resolved> { @@ -545,17 +658,17 @@ impl Index { } pub fn effective_collaborator(&self, repo: &RepoDid, did: &AccountDid) -> Resolved { - self.collaborator_standing(repo, did) - .map(|standing| standing.is_some_and(Standing::is_effective)) + self.collaborator_state(repo, did) + .map(|state| state.is_some_and(EntryState::is_effective)) } - pub fn collaborator_standing( + pub fn collaborator_state( &self, repo: &RepoDid, did: &AccountDid, - ) -> Resolved> { + ) -> Resolved> { self.collaborators - .slot(&self.interner, repo, did, Provenance::standing) + .slot(&self.interner, repo, did, Provenance::state) } pub fn collaborator_entries(&self, repo: &RepoDid) -> Resolved> { @@ -565,20 +678,20 @@ impl Index { pub fn collaborator_entries_where( &self, repo: &RepoDid, - keep: impl Fn(Standing) -> bool, + keep: impl Fn(EntryState) -> bool, ) -> Resolved> { self.collaborators.entries_where(&self.interner, repo, keep) } pub fn effective_collaborators(&self, repo: &RepoDid) -> Resolved> { self.collaborators - .subjects_where(&self.interner, repo, Standing::is_effective) + .subjects_where(&self.interner, repo, EntryState::is_effective) } pub fn invited_collaborators(&self, repo: &RepoDid) -> Resolved> { self.collaborators - .subjects_where(&self.interner, repo, |standing| { - matches!(standing, Standing::Invited) + .subjects_where(&self.interner, repo, |state| { + matches!(state, EntryState::Invited) }) } @@ -611,6 +724,12 @@ impl Index { self.registry.rkey_of(&self.interner, repo) } + pub fn contribution_policy(&self, repo: &RepoDid) -> Resolved> { + self.registry + .policy_of(&self.interner, repo) + .map(|policy| policy.map(|policy| RepoPolicy::new(repo.clone(), policy))) + } + pub fn ownership_of(&self, repo: &RepoDid) -> Resolved> { self.registry.ownership(&self.interner, repo) } @@ -765,14 +884,14 @@ impl KeySet<'_> { self.0.keys.any_unheld() } - pub fn work(&self, now: UnixSeconds, floor: SweepFloor) -> Resolved { + pub fn work(&self, now: UnixSeconds, interval: SweepInterval) -> Resolved { let index = self.0; let generation = index.generation(); let Resolved::Ready(granted) = index.push_grants() else { return Resolved::Warming; }; let pushers = granted.subjects; - let against = match index.keys.take_suspicion(now, floor) { + let against = match index.keys.take_suspicion(now, interval) { true => Recheck::Everything, false => Recheck::Renewals, }; diff --git a/knot2/crates/knot-index/src/projections.rs b/knot2/crates/knot-index/src/projections.rs index 488fb1d1c..fb8a1c77e 100644 --- a/knot2/crates/knot-index/src/projections.rs +++ b/knot2/crates/knot-index/src/projections.rs @@ -6,17 +6,19 @@ use std::sync::{Arc, Mutex}; use knot_cob::{Change, ChangeId, ChangePayload, Checkpoint, CobId, CobStore, Evaluate}; use knot_cobs::{ - CollaboratorsChange, CollaboratorsCob, Effect, Entry, Grant, GrantChange, Offer, Registration, - Registry, RegistryChange, Rename, RepoRef, RepoRegistryCob, Roster, Standing, + CollaboratorsChange, CollaboratorsCob, Effect, Entry, EntryState, Grant, GrantChange, Offer, + Policy, Registration, Registry, RegistryChange, Rename, RepoRef, RepoRegistryCob, Roll, +}; +use knot_types::{ + AccountDid, ClonePath, ContributionPolicy, OfferedKey, OwnerDid, RepoDid, RepoRkey, UnixSeconds, }; -use knot_types::{AccountDid, ClonePath, OfferedKey, OwnerDid, RepoDid, RepoRkey, UnixSeconds}; use crate::consent::Consent; use crate::coverage::{Coverage, CoverageCell, Resolved}; use crate::error::IndexError; use crate::intern::{AccountKey, Interner, NameKey, OwnerKey, RepoKey, RkeyKey}; use crate::{ - IndexGeneration, KeyBudget, KeyLease, KeyRecord, KeyReprieve, KeyReprieved, SweepFloor, + IndexGeneration, KeyBudget, KeyLease, KeyRecord, KeyReprieve, KeyReprieved, SweepInterval, }; #[derive(Debug, Clone, Copy)] @@ -51,8 +53,8 @@ impl Provenance { self } - pub(crate) fn standing(self) -> Standing { - Standing::of(self.offer, self.created_at, self.verified_at) + pub(crate) fn state(self) -> EntryState { + EntryState::of(self.offer, self.created_at, self.verified_at) } pub(crate) fn consent(self) -> Consent { @@ -159,7 +161,7 @@ where fn rows_where( &self, interner: &Interner, - keep: impl Fn(Standing) -> bool, + keep: impl Fn(EntryState) -> bool, row: impl Fn(AccountDid, Provenance) -> T, ) -> Resolved> { match self.coverage.get() { @@ -167,7 +169,7 @@ where Coverage::Ready => { let mut out = BTreeMap::new(); self.membership.iter_sync(|&subject, slot| { - if keep(slot.standing()) { + if keep(slot.state()) { let did = interner.resolve_account(subject); out.insert(did.clone(), row(did, *slot)); } @@ -181,7 +183,7 @@ where pub(crate) fn entries_where( &self, interner: &Interner, - keep: impl Fn(Standing) -> bool, + keep: impl Fn(EntryState) -> bool, ) -> Resolved> { self.rows_where(interner, keep, |did, slot| (did, slot.entry(interner))) } @@ -189,7 +191,7 @@ where pub(crate) fn subjects_where( &self, interner: &Interner, - keep: impl Fn(Standing) -> bool, + keep: impl Fn(EntryState) -> bool, ) -> Resolved> { self.rows_where(interner, keep, |did, _| did) } @@ -201,7 +203,7 @@ where object: CobId, ) -> Result<(), IndexError> where - Cob: Checkpoint + Evaluate, + Cob: Checkpoint + Evaluate, { let mut tip = self .tip @@ -209,10 +211,10 @@ where .unwrap_or_else(|poisoned| poisoned.into_inner()); match *tip { None => { - let (roster, seeded) = store + let (roll, seeded) = store .materialize::(object) .inspect_err(|_| self.coverage.set(Coverage::Warming))?; - self.seed(interner, &roster); + self.seed(interner, &roll); *tip = Some(seeded); } Some(prev) => { @@ -229,9 +231,9 @@ where Ok(()) } - fn seed(&self, interner: &Interner, roster: &Roster) { + fn seed(&self, interner: &Interner, roll: &Roll) { self.membership.clear_sync(); - roster.entries().for_each(|(subject, entry)| { + roll.entries().for_each(|(subject, entry)| { let _ = self.membership.insert_sync( interner.intern_account(subject), Provenance::seeded(interner, entry), @@ -275,7 +277,7 @@ where } } -struct RepoRoster { +struct RepoRoll { tip: Option, entries: BTreeMap, } @@ -283,24 +285,23 @@ struct RepoRoster { const REPO_LOCK_STRIPES: usize = 256; pub(crate) struct CollaboratorsProjection { - rosters: scc::HashMap, + rolls: scc::HashMap, locks: [Mutex<()>; REPO_LOCK_STRIPES], } impl CollaboratorsProjection { pub(crate) fn new() -> Self { Self { - rosters: scc::HashMap::new(), + rolls: scc::HashMap::new(), locks: std::array::from_fn(|_| Mutex::new(())), } } pub(crate) fn any_invited(&self) -> bool { - !self.rosters.iter_sync(|_, roster| { - roster - .entries + !self.rolls.iter_sync(|_, roll| { + roll.entries .values() - .all(|slot| !matches!(slot.standing(), Standing::Invited)) + .all(|slot| !matches!(slot.state(), EntryState::Invited)) }) } @@ -317,7 +318,7 @@ impl CollaboratorsProjection { pub(crate) fn is_folded(&self, interner: &Interner, repo: &RepoDid) -> bool { interner .repo(repo) - .is_some_and(|repo| self.rosters.contains_sync(&repo)) + .is_some_and(|repo| self.rolls.contains_sync(&repo)) } pub(crate) fn slot( @@ -330,10 +331,10 @@ impl CollaboratorsProjection { let Some(repo) = interner.repo(repo) else { return Resolved::Warming; }; - match self.rosters.read_sync(&repo, |_, roster| { + match self.rolls.read_sync(&repo, |_, roll| { interner .account(did) - .and_then(|account| roster.entries.get(&account)) + .and_then(|account| roll.entries.get(&account)) .map(|slot| read(*slot)) }) { Some(value) => Resolved::Ready(value), @@ -342,10 +343,9 @@ impl CollaboratorsProjection { } pub(crate) fn conditional(&self, repo: RepoKey, subject: AccountKey) -> bool { - self.rosters - .read_sync(&repo, |_, roster| { - roster - .entries + self.rolls + .read_sync(&repo, |_, roll| { + roll.entries .get(&subject) .is_some_and(|slot| slot.consent().is_conditional()) }) @@ -356,17 +356,16 @@ impl CollaboratorsProjection { &self, interner: &Interner, repo: &RepoDid, - keep: impl Fn(Standing) -> bool, + keep: impl Fn(EntryState) -> bool, row: impl Fn(AccountDid, Provenance) -> T, ) -> Resolved> { let Some(repo) = interner.repo(repo) else { return Resolved::Warming; }; - match self.rosters.read_sync(&repo, |_, roster| { - roster - .entries + match self.rolls.read_sync(&repo, |_, roll| { + roll.entries .iter() - .filter(|(_, slot)| keep(slot.standing())) + .filter(|(_, slot)| keep(slot.state())) .map(|(&subject, slot)| { let did = interner.resolve_account(subject); (did.clone(), row(did, *slot)) @@ -384,7 +383,7 @@ impl CollaboratorsProjection { &self, interner: &Interner, repo: &RepoDid, - keep: impl Fn(Standing) -> bool, + keep: impl Fn(EntryState) -> bool, ) -> Resolved> { self.rows_where(interner, repo, keep, |did, slot| { (did, slot.entry(interner)) @@ -395,7 +394,7 @@ impl CollaboratorsProjection { &self, interner: &Interner, repo: &RepoDid, - keep: impl Fn(Standing) -> bool, + keep: impl Fn(EntryState) -> bool, ) -> Resolved> { self.rows_where(interner, repo, keep, |did, _| did) } @@ -405,7 +404,7 @@ impl CollaboratorsProjection { let _guard = lock.lock().unwrap_or_else(|poisoned| poisoned.into_inner()); self.install( repo, - RepoRoster { + RepoRoll { tip: None, entries: BTreeMap::new(), }, @@ -432,14 +431,11 @@ impl CollaboratorsProjection { repo: RepoKey, object: CobId, ) -> Result<(), IndexError> { - let prev = self - .rosters - .read_sync(&repo, |_, roster| roster.tip) - .flatten(); + let prev = self.rolls.read_sync(&repo, |_, roll| roll.tip).flatten(); match prev { None => { - let (roster, tip) = store.materialize::(object)?; - let entries = roster + let (roll, tip) = store.materialize::(object)?; + let entries = roll .entries() .map(|(subject, entry)| { ( @@ -450,7 +446,7 @@ impl CollaboratorsProjection { .collect(); self.install( repo, - RepoRoster { + RepoRoll { tip: Some(tip), entries, }, @@ -459,17 +455,17 @@ impl CollaboratorsProjection { Some(prev) => { let delta = store.changes_since::(object, Some(prev))?; let decoded = decode_delta::(&delta.changes)?; - let mut occupied = self.rosters.entry_sync(repo).or_insert_with(|| RepoRoster { + let mut occupied = self.rolls.entry_sync(repo).or_insert_with(|| RepoRoll { tip: None, entries: BTreeMap::new(), }); - let roster = occupied.get_mut(); - roster.entries = decoded + let roll = occupied.get_mut(); + roll.entries = decoded .into_iter() - .fold(std::mem::take(&mut roster.entries), |entries, change| { + .fold(std::mem::take(&mut roll.entries), |entries, change| { apply(interner, entries, change) }); - roster.tip = Some(delta.tip); + roll.tip = Some(delta.tip); } } Ok(()) @@ -482,16 +478,16 @@ impl CollaboratorsProjection { } fn purge_repo(&self, repo: RepoKey) { - let _ = self.rosters.remove_sync(&repo); + let _ = self.rolls.remove_sync(&repo); } - fn install(&self, repo: RepoKey, roster: RepoRoster) { - match self.rosters.entry_sync(repo) { + fn install(&self, repo: RepoKey, roll: RepoRoll) { + match self.rolls.entry_sync(repo) { scc::hash_map::Entry::Occupied(mut occupied) => { - let _ = occupied.insert(roster); + let _ = occupied.insert(roll); } scc::hash_map::Entry::Vacant(vacant) => { - vacant.insert_entry(roster); + vacant.insert_entry(roll); } } } @@ -530,6 +526,7 @@ struct RecordSlot { rkey: RkeyKey, name: NameKey, created_at: UnixSeconds, + policy: ContributionPolicy, } pub(crate) struct RegistryProjection { @@ -555,12 +552,17 @@ impl RegistryProjection { self.coverage.get() } + pub(crate) fn warm(&self) { + self.coverage.set(Coverage::Warming); + } + pub(crate) fn reset(&self, interner: &Interner) -> Vec { let mut tip = self .tip .lock() .unwrap_or_else(|poisoned| poisoned.into_inner()); let evacuated = self.hosted_repos(interner); + self.coverage.set(Coverage::Warming); self.aliases.clear_sync(); self.names.clear_sync(); self.records.clear_sync(); @@ -671,6 +673,20 @@ impl RegistryProjection { ) } + pub(crate) fn policy_of( + &self, + interner: &Interner, + repo: &RepoDid, + ) -> Resolved> { + if self.coverage.get() == Coverage::Warming { + return Resolved::Warming; + } + let Some(target) = interner.repo(repo) else { + return Resolved::Ready(None); + }; + Resolved::Ready(self.records.read_sync(&target, |_, slot| slot.policy)) + } + pub(crate) fn ownership( &self, interner: &Interner, @@ -736,6 +752,7 @@ impl RegistryProjection { } fn seed(&self, interner: &Interner, registry: &Registry) { + self.coverage.set(Coverage::Warming); self.aliases.clear_sync(); self.names.clear_sync(); self.records.clear_sync(); @@ -749,6 +766,7 @@ impl RegistryProjection { interner.intern_rkey(&record.rkey), name, record.created_at, + record.policy, ); self.bind_name(owner, name, record.created_at, repo); }); @@ -795,6 +813,7 @@ impl RegistryProjection { RegistryChange::Deregister(target) => { self.apply_deregister(interner, target, displaced) } + RegistryChange::SetPolicy(policy) => self.apply_policy(interner, policy, displaced), }) } @@ -813,7 +832,14 @@ impl RegistryProjection { displaced.push(repo); } displaced.extend(self.steal_alias(owner, rkey, repo)); - self.upsert_record(repo, owner, rkey, name, registration.created_at); + self.upsert_record( + repo, + owner, + rkey, + name, + registration.created_at, + registration.policy, + ); self.upsert_alias(owner, rkey, repo); self.bind_name(owner, name, registration.created_at, repo); displaced @@ -829,17 +855,17 @@ impl RegistryProjection { let owner = interner.intern_owner(&rename.owner); let rkey = interner.intern_rkey(&rename.rkey); let name = interner.intern_name(&rename.name); - let held = self + let record = self .records .read_sync(&repo, |_, slot| { - (slot.owner == owner).then_some((slot.created_at, slot.name)) + (slot.owner == owner).then_some((slot.created_at, slot.name, slot.policy)) }) .flatten(); - let Some((created_at, previous)) = held else { + let Some((created_at, previous, policy)) = record else { return displaced; }; displaced.extend(self.steal_alias(owner, rkey, repo)); - self.upsert_record(repo, owner, rkey, name, created_at); + self.upsert_record(repo, owner, rkey, name, created_at, policy); self.upsert_alias(owner, rkey, repo); self.bind_name(owner, name, created_at, repo); if previous != name { @@ -848,6 +874,35 @@ impl RegistryProjection { displaced } + fn apply_policy( + &self, + interner: &Interner, + change: Policy, + displaced: Vec, + ) -> Vec { + let repo = interner.intern_repo(&change.repo); + let owner = interner.intern_owner(&change.owner); + let applied = self + .records + .update_sync(&repo, |_, slot| { + let owner_match = slot.owner == owner; + if owner_match { + slot.policy = change.policy; + } + owner_match + }) + .unwrap_or(false); + if !applied { + tracing::warn!( + repo = %change.repo, + owner = %change.owner, + policy = %change.policy, + "contribution policy change names repository registry doesn't have under that owner. Its policy stays put" + ); + } + displaced + } + fn apply_deregister( &self, interner: &Interner, @@ -918,6 +973,7 @@ impl RegistryProjection { rkey: RkeyKey, name: NameKey, created_at: UnixSeconds, + policy: ContributionPolicy, ) { if self .records @@ -926,6 +982,7 @@ impl RegistryProjection { slot.rkey = rkey; slot.name = name; slot.created_at = created_at; + slot.policy = policy; }) .is_none() { @@ -936,6 +993,7 @@ impl RegistryProjection { rkey, name, created_at, + policy, }, ); } @@ -1081,14 +1139,14 @@ impl KeyProjection { self.suspect.store(true, Ordering::Release); } - pub(crate) fn take_suspicion(&self, now: UnixSeconds, floor: SweepFloor) -> bool { - let held_until = |swept: i64| swept.saturating_add(crate::whole_secs(floor.get())); + pub(crate) fn take_suspicion(&self, now: UnixSeconds, interval: SweepInterval) -> bool { + let due_at = |swept: i64| swept.saturating_add(crate::whole_secs(interval.get())); match self.suspect.load(Ordering::Acquire) { false => false, true => { self.swept_at .fetch_update(Ordering::AcqRel, Ordering::Acquire, |swept| { - (now.get() >= held_until(swept)).then_some(now.get()) + (now.get() >= due_at(swept)).then_some(now.get()) }) .is_ok() && self.suspect.swap(false, Ordering::AcqRel) @@ -1233,9 +1291,9 @@ impl KeyProjection { }; match grace.extend(slot.get().lease, now) { Some(lease) => { - let carried = slot.get().reading.is_published(); + let published = slot.get().reading.is_published(); slot.get_mut().lease = lease; - match carried { + match published { true => KeyReprieved::Extended, false => KeyReprieved::Pending, } diff --git a/knot2/crates/knot-index/src/social.rs b/knot2/crates/knot-index/src/social.rs new file mode 100644 index 000000000..c277a20b6 --- /dev/null +++ b/knot2/crates/knot-index/src/social.rs @@ -0,0 +1,533 @@ +use std::collections::{BTreeMap, BTreeSet}; +use std::hash::{Hash, Hasher}; +use std::sync::Mutex; +use std::sync::atomic::{AtomicU64, Ordering as AtomicOrdering}; + +use knot_cob::{ChangePayload, CobId, CobStore, HistoryModel}; +use knot_resource::StructureBytes; +use knot_types::{ + AccountDid, RecordAddress, RecordCid, RecordCollection, RepoDid, SubjectNumber, TypeName, + UnixSeconds, +}; + +use crate::coverage::Resolved; +use crate::error::IndexError; +use crate::intern::{AccountKey, CollectionKey, Interner, RecordKey, RepoKey}; + +const REPO_LOCK_STRIPES: usize = 256; + +knot_types::scalar_newtype! { + pub struct SocialResidency(usize); +} + +impl SocialResidency { + pub const DEFAULT: Self = Self::new(4_096); +} + +#[derive(Debug, Clone, Copy, PartialEq, Eq, PartialOrd, Ord)] +struct Slot { + collection: CollectionKey, + rkey: RecordKey, +} + +impl Slot { + fn of(interner: &Interner, address: &RecordAddress) -> Self { + Self { + collection: interner.intern_collection(address.collection()), + rkey: interner.intern_record_rkey(address.rkey()), + } + } + + fn known(interner: &Interner, address: &RecordAddress) -> Option { + Some(Self { + collection: interner.collection(address.collection())?, + rkey: interner.record_rkey(address.rkey())?, + }) + } + + fn address(self, interner: &Interner) -> RecordAddress { + RecordAddress::new( + interner.resolve_collection(self.collection), + interner.resolve_record_rkey(self.rkey), + ) + } +} + +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +struct Objected { + slot: Slot, + author: AccountKey, + number: SubjectNumber, + parent: Option, + created_at: UnixSeconds, + version: Option, +} + +#[derive(Debug, Clone, Copy, PartialEq, Eq, PartialOrd, Ord)] +struct Ordered(UnixSeconds, CobId); + +#[derive(Debug)] +struct RepoSocial { + folded_at: u64, + charged: BTreeMap, + objects: BTreeMap, + claiming: BTreeMap>, + addressed: BTreeMap, + contested: BTreeSet, + of_kind: BTreeMap>, + children: BTreeMap>, + numbered: BTreeMap>, + structure: StructureBytes, +} + +impl RepoSocial { + fn new(folded_at: u64) -> Self { + Self { + folded_at, + charged: BTreeMap::new(), + objects: BTreeMap::new(), + claiming: BTreeMap::new(), + addressed: BTreeMap::new(), + contested: BTreeSet::new(), + of_kind: BTreeMap::new(), + children: BTreeMap::new(), + numbered: BTreeMap::new(), + structure: StructureBytes::new(0), + } + } + + fn install(&mut self, object: CobId, folded: Folded) { + let left = self.retract(object); + self.charged.insert(object, folded.bytes); + self.structure = + StructureBytes::new(self.structure.get().saturating_add(folded.bytes.get())); + let arrived = folded.objected.map(|objected| { + self.claiming + .entry(objected.slot) + .or_default() + .insert(Ordered(objected.created_at, object)); + self.objects.insert(object, objected); + objected.slot + }); + left.into_iter() + .chain(arrived.filter(|slot| Some(*slot) != left)) + .for_each(|slot| self.settle(slot)); + } + + fn retract(&mut self, object: CobId) -> Option { + let charged = self.charged.remove(&object).map_or(0, StructureBytes::get); + self.structure = StructureBytes::new(self.structure.get().saturating_sub(charged)); + self.contested.remove(&object); + self.objects.remove(&object).map(|objected| { + self.unlist(object, &objected); + Self::without( + &mut self.claiming, + &objected.slot, + &Ordered(objected.created_at, object), + ); + objected.slot + }) + } + + fn settle(&mut self, slot: Slot) { + let Some(Ordered(_, winner)) = self + .claiming + .get(&slot) + .and_then(|claiming| claiming.first().copied()) + else { + self.addressed.remove(&slot); + return; + }; + self.addressed.insert(slot, winner); + let claimants: Vec = self + .claiming + .get(&slot) + .into_iter() + .flatten() + .map(|Ordered(_, claimant)| *claimant) + .collect(); + claimants.into_iter().for_each(|claimant| { + if claimant == winner { + self.contested.remove(&claimant); + } else if self.contested.insert(claimant) { + tracing::warn!( + object = %claimant, + "when two collaborative objects claim one record address, \ + later one doesn't answer address lookups" + ); + } + self.relist(claimant); + }); + } + + fn relist(&mut self, object: CobId) { + let Some(objected) = self.objects.get(&object).copied() else { + return; + }; + self.unlist(object, &objected); + if !self.contested.contains(&object) { + self.list(object, &objected); + } + } + + fn list(&mut self, object: CobId, objected: &Objected) { + let ordered = Ordered(objected.created_at, object); + self.of_kind + .entry(objected.slot.collection) + .or_default() + .insert(ordered); + self.numbered + .entry(objected.slot.collection) + .or_default() + .insert((objected.number, object)); + objected.parent.iter().for_each(|parent| { + self.children.entry(*parent).or_default().insert(ordered); + }); + } + + fn unlist(&mut self, object: CobId, objected: &Objected) { + let ordered = Ordered(objected.created_at, object); + Self::without(&mut self.of_kind, &objected.slot.collection, &ordered); + Self::without( + &mut self.numbered, + &objected.slot.collection, + &(objected.number, object), + ); + objected + .parent + .iter() + .for_each(|parent| Self::without(&mut self.children, parent, &ordered)); + } + + fn without(index: &mut BTreeMap>, key: &K, value: &V) { + let emptied = index.get_mut(key).is_some_and(|set| { + set.remove(value); + set.is_empty() + }); + if emptied { + index.remove(key); + } + } + + fn serving(&self, object: CobId) -> Option<&Objected> { + self.objects + .get(&object) + .filter(|_| !self.contested.contains(&object)) + } +} + +struct Folded { + bytes: StructureBytes, + objected: Option, +} + +pub(crate) struct SocialProjection { + repos: scc::HashMap, + locks: [Mutex<()>; REPO_LOCK_STRIPES], + residency: SocialResidency, + folds: AtomicU64, +} + +impl SocialProjection { + pub(crate) fn new(residency: SocialResidency) -> Self { + Self { + repos: scc::HashMap::new(), + locks: std::array::from_fn(|_| Mutex::new(())), + residency, + folds: AtomicU64::new(0), + } + } + + fn repo_lock(&self, repo: RepoKey) -> &Mutex<()> { + let mut hasher = std::collections::hash_map::DefaultHasher::new(); + repo.hash(&mut hasher); + &self.locks[(hasher.finish() % REPO_LOCK_STRIPES as u64) as usize] + } + + fn next_fold(&self) -> u64 { + self.folds.fetch_add(1, AtomicOrdering::Relaxed) + } + + pub(crate) fn is_folded(&self, interner: &Interner, repo: &RepoDid) -> bool { + interner + .repo(repo) + .is_some_and(|repo| self.repos.contains_sync(&repo)) + } + + pub(crate) fn drop_repo(&self, repo: RepoKey) { + let lock = self.repo_lock(repo); + let _guard = lock.lock().unwrap_or_else(|poisoned| poisoned.into_inner()); + self.repos.remove_sync(&repo); + } + + pub(crate) fn refresh_repo( + &self, + interner: &Interner, + store: &CobStore, + repo: RepoKey, + ) -> Result<(), IndexError> { + let lock = self.repo_lock(repo); + let _guard = lock.lock().unwrap_or_else(|poisoned| poisoned.into_inner()); + let social = self.fold_repo(interner, store).inspect_err(|_| { + self.repos.remove_sync(&repo); + })?; + self.publish(repo, social); + Ok(()) + } + + pub(crate) fn refresh_object( + &self, + interner: &Interner, + store: &CobStore, + repo: RepoKey, + type_name: &TypeName, + object: CobId, + ) -> Result<(), IndexError> { + let lock = self.repo_lock(repo); + let _guard = lock.lock().unwrap_or_else(|poisoned| poisoned.into_inner()); + if !is_social(type_name) { + return Ok(()); + } + let folded = fold_object(interner, store, type_name, object)?; + let stored = self + .repos + .update_sync(&repo, |_, social| social.install(object, folded)) + .is_some(); + if stored { + return Ok(()); + } + let social = self.fold_repo(interner, store)?; + self.publish(repo, social); + Ok(()) + } + + fn publish(&self, repo: RepoKey, social: RepoSocial) { + self.evict_past_residency(repo); + match self.repos.entry_sync(repo) { + scc::hash_map::Entry::Occupied(mut occupied) => *occupied.get_mut() = social, + scc::hash_map::Entry::Vacant(vacant) => { + vacant.insert_entry(social); + } + } + } + + fn evict_past_residency(&self, keeping: RepoKey) { + if self.repos.len() < self.residency.get() || self.repos.contains_sync(&keeping) { + return; + } + let mut coldest: Option<(u64, RepoKey)> = None; + self.repos.iter_sync(|repo, social| { + if coldest.is_none_or(|(at, _)| social.folded_at < at) { + coldest = Some((social.folded_at, *repo)); + } + true + }); + coldest.into_iter().for_each(|(_, repo)| { + self.repos.remove_sync(&repo); + }); + } + + fn fold_repo(&self, interner: &Interner, store: &CobStore) -> Result { + let folded = store + .cobs()? + .into_iter() + .filter(|(type_name, _)| is_social(type_name)) + .map(|(type_name, object)| { + fold_object(interner, store, &type_name, object).map(|folded| (object, folded)) + }) + .collect::, IndexError>>()?; + Ok(folded.into_iter().fold( + RepoSocial::new(self.next_fold()), + |mut social, (object, folded)| { + social.install(object, folded); + social + }, + )) + } + + fn read( + &self, + interner: &Interner, + repo: &RepoDid, + read: impl FnOnce(&RepoSocial) -> T, + ) -> Resolved { + let Some(repo) = interner.repo(repo) else { + return Resolved::Warming; + }; + match self.repos.read_sync(&repo, |_, social| read(social)) { + Some(value) => Resolved::Ready(value), + None => Resolved::Warming, + } + } + + pub(crate) fn object_at( + &self, + interner: &Interner, + repo: &RepoDid, + address: &RecordAddress, + ) -> Resolved> { + let slot = Slot::known(interner, address); + self.read(interner, repo, |social| { + slot.and_then(|slot| social.addressed.get(&slot).copied()) + }) + } + + pub(crate) fn address_of( + &self, + interner: &Interner, + repo: &RepoDid, + object: CobId, + ) -> Resolved> { + self.read(interner, repo, |social| { + social + .serving(object) + .map(|objected| objected.slot.address(interner)) + }) + } + + pub(crate) fn of_kind( + &self, + interner: &Interner, + repo: &RepoDid, + collection: &RecordCollection, + ) -> Resolved> { + let key = interner.collection(collection); + self.read(interner, repo, |social| { + key.and_then(|key| social.of_kind.get(&key)) + .map(|ordered| ordered.iter().map(|Ordered(_, object)| *object).collect()) + .unwrap_or_default() + }) + } + + pub(crate) fn children_of( + &self, + interner: &Interner, + repo: &RepoDid, + parent: &RecordAddress, + ) -> Resolved> { + let slot = Slot::known(interner, parent); + self.read(interner, repo, |social| { + slot.and_then(|slot| social.children.get(&slot)) + .map(|ordered| ordered.iter().map(|Ordered(_, object)| *object).collect()) + .unwrap_or_default() + }) + } + + pub(crate) fn latest_child( + &self, + interner: &Interner, + repo: &RepoDid, + parent: &RecordAddress, + collection: &RecordCollection, + ) -> Resolved> { + let slot = Slot::known(interner, parent); + let kind = interner.collection(collection); + self.read(interner, repo, |social| { + let (Some(slot), Some(kind)) = (slot, kind) else { + return None; + }; + social + .children + .get(&slot)? + .iter() + .rev() + .filter_map(|Ordered(_, object)| social.objects.get(object)) + .find(|objected| objected.slot.collection == kind && objected.version.is_some()) + .map(|objected| objected.slot.address(interner)) + }) + } + + pub(crate) fn highest_number( + &self, + interner: &Interner, + repo: &RepoDid, + collection: &RecordCollection, + ) -> Resolved> { + let key = interner.collection(collection); + self.read(interner, repo, |social| { + key.and_then(|key| social.numbered.get(&key)) + .and_then(|numbers| numbers.last().map(|(number, _)| *number)) + }) + } + + pub(crate) fn author_of( + &self, + interner: &Interner, + repo: &RepoDid, + object: CobId, + ) -> Resolved> { + self.read(interner, repo, |social| { + social + .objects + .get(&object) + .map(|objected| interner.resolve_account(objected.author)) + }) + } + + pub(crate) fn version_of( + &self, + interner: &Interner, + repo: &RepoDid, + object: CobId, + ) -> Resolved> { + self.read(interner, repo, |social| { + social + .objects + .get(&object) + .and_then(|objected| objected.version) + }) + } + + pub(crate) fn structure( + &self, + interner: &Interner, + repo: &RepoDid, + ) -> Resolved { + self.read(interner, repo, |social| social.structure) + } +} + +fn is_social(type_name: &TypeName) -> bool { + *type_name != knot_cobs::CollaboratorsChange::type_name() +} + +fn fold_object( + interner: &Interner, + store: &CobStore, + type_name: &TypeName, + object: CobId, +) -> Result { + let delta = store.changes_of(type_name, object, HistoryModel::Convergent, None)?; + let bytes = StructureBytes::new(delta.changes.iter().fold(0u64, |total, change| { + let bytes_of = change.payload().len() + change.record.bytes().map_or(0, <[u8]>::len); + total.saturating_add(bytes_of as u64) + })); + let state = + delta + .changes + .iter() + .try_fold(knot_cobs::SocialState::Unopened, |state, change| { + knot_cobs::decode_social(change.payload()) + .map(|wire| state.fold(&wire)) + .map_err(|error| IndexError::Decode { + change: change.id, + type_name: type_name.clone(), + reason: error.to_string(), + }) + })?; + Ok(Folded { + bytes, + objected: state.object().map(|object| Objected { + slot: Slot::of(interner, object.address()), + author: interner.intern_account(object.author()), + number: object.number(), + parent: object.parent().map(|parent| Slot::of(interner, parent)), + created_at: object.created_at(), + version: object.version(), + }), + }) +} + +pub(crate) fn next_number(highest: Option) -> Option { + highest.map_or(Some(SubjectNumber::FIRST), SubjectNumber::after) +}