knot2/index: watch account's published keys with lease & budget · alice.pds.demo.boltless.dev/core@060d673

+2 -1

Cargo.lock

··· 4539 4539 name = "knot-index" 4540 4540 version = "2.0.0" 4541 4541 dependencies = [ 4542 - "knot-cache", 4543 4542 "knot-cob", 4544 4543 "knot-cobs", 4545 4544 "knot-git", ··· 4551 4550 "serde", 4552 4551 "tempfile", 4553 4552 "thiserror 2.0.18", 4553 + "tokio", 4554 + "tracing", 4554 4555 ] 4555 4556 4556 4557 [[package]]

+2 -1

knot2/crates/knot-index/Cargo.toml

View file

··· 10 10 knot-git = { workspace = true } 11 11 knot-cob = { workspace = true } 12 12 knot-cobs = { workspace = true } 13 - knot-cache = { workspace = true } 14 13 scc = { workspace = true } 15 14 lasso = { workspace = true } 16 15 thiserror = { workspace = true } 16 + tokio = { workspace = true } 17 + tracing = { workspace = true } 17 18 18 19 [dev-dependencies] 19 20 knot-runtime = { workspace = true }

+505 -13

knot2/crates/knot-index/src/lib.rs

View file

··· 9 9 10 10 use std::path::PathBuf; 11 11 use std::sync::atomic::{AtomicU64, Ordering}; 12 + use std::time::Duration; 12 13 13 14 use knot_cob::{ChangePayload, CobStore}; 14 15 use knot_cobs::{ ··· 16 17 MembersCob, RegistryChange, RepoRegistryCob, 17 18 }; 18 19 use knot_git::{Layout, Repo}; 19 - use knot_types::{AccountDid, ClonePath, OwnerDid, RepoDid, RepoRkey}; 20 + use knot_types::{AccountDid, ClonePath, OwnerDid, RepoDid, RepoRkey, UnixSeconds}; 21 + use tokio::sync::watch; 20 22 21 - use intern::Interner; 23 + use intern::{Interner, RepoKey}; 22 24 use projections::{CollaboratorsProjection, GrantSetProjection, KeyProjection, RegistryProjection}; 23 25 24 26 knot_types::scalar_newtype! { ··· 34 36 pub keys: Coverage, 35 37 } 36 38 39 + macro_rules! account_list { 40 + ($($name:ident),+ $(,)?) => {$( 41 + #[derive(Debug, Clone, PartialEq, Eq)] 42 + pub struct $name(Vec<AccountDid>); 43 + 44 + impl $name { 45 + pub fn new(accounts: Vec<AccountDid>) -> Self { 46 + Self(accounts) 47 + } 48 + 49 + pub fn len(&self) -> usize { 50 + self.0.len() 51 + } 52 + 53 + pub fn is_empty(&self) -> bool { 54 + self.0.is_empty() 55 + } 56 + 57 + pub fn as_slice(&self) -> &[AccountDid] { 58 + &self.0 59 + } 60 + 61 + pub fn into_vec(self) -> Vec<AccountDid> { 62 + self.0 63 + } 64 + } 65 + )+}; 66 + } 67 + 68 + account_list!( 69 + Pushers, 70 + KeptAccounts, 71 + StalePushers, 72 + SuspectPushers, 73 + UnreadMembers, 74 + StaleMembers, 75 + ); 76 + 77 + #[derive(Debug, Clone, Copy, PartialEq, Eq)] 78 + pub enum HostedCoverage { 79 + Whole, 80 + Partial { unread: usize }, 81 + } 82 + 83 + impl HostedCoverage { 84 + const fn over(unread: usize) -> Self { 85 + match unread { 86 + 0 => Self::Whole, 87 + unread => Self::Partial { unread }, 88 + } 89 + } 90 + } 91 + 92 + struct Granted { 93 + subjects: Vec<AccountDid>, 94 + hosted: HostedCoverage, 95 + } 96 + 97 + enum Folded { 98 + Grants(Vec<AccountDid>), 99 + Pending, 100 + Unreadable, 101 + } 102 + 103 + impl Folded { 104 + fn into_grants(self) -> Option<Vec<AccountDid>> { 105 + match self { 106 + Folded::Grants(subjects) => Some(subjects), 107 + Folded::Pending | Folded::Unreadable => None, 108 + } 109 + } 110 + } 111 + 112 + #[derive(Debug, Clone, Copy, PartialEq, Eq)] 113 + pub struct SweepFloor(Duration); 114 + 115 + impl SweepFloor { 116 + pub const DEFAULT: Self = Self::from_secs(60); 117 + 118 + pub const fn from_secs(secs: u64) -> Self { 119 + Self(Duration::from_secs(secs)) 120 + } 121 + 122 + pub const fn get(self) -> Duration { 123 + self.0 124 + } 125 + } 126 + 127 + #[derive(Debug, Clone, Copy, PartialEq, Eq)] 128 + pub struct KeyTtl(Duration); 129 + 130 + impl KeyTtl { 131 + pub const DEFAULT: Self = Self::from_secs(3_600); 132 + 133 + pub const fn from_secs(secs: u64) -> Self { 134 + Self(Duration::from_secs(secs)) 135 + } 136 + 137 + pub const fn get(self) -> Duration { 138 + self.0 139 + } 140 + 141 + pub const fn longest(self, other: Self) -> Self { 142 + match self.0.as_secs() >= other.0.as_secs() { 143 + true => self, 144 + false => other, 145 + } 146 + } 147 + 148 + pub const fn lease_from(self, now: UnixSeconds) -> KeyLease { 149 + KeyLease { 150 + read_at: now, 151 + expires_at: after(now, self.0), 152 + } 153 + } 154 + } 155 + 156 + #[derive(Debug, Clone, Copy, PartialEq, Eq)] 157 + pub struct KeyLease { 158 + read_at: UnixSeconds, 159 + expires_at: UnixSeconds, 160 + } 161 + 162 + impl KeyLease { 163 + pub(crate) const fn is_live(self, now: UnixSeconds) -> bool { 164 + self.expires_at.get() > now.get() 165 + } 166 + 167 + pub(crate) const fn renewal_due(self, now: UnixSeconds) -> bool { 168 + let held = self.expires_at.get().saturating_sub(self.read_at.get()); 169 + now.get() >= self.read_at.saturating_add_secs(held / 2).get() 170 + } 171 + } 172 + 173 + #[derive(Debug, Clone, Copy, PartialEq, Eq)] 174 + pub struct KeyReprieve { 175 + retry: Duration, 176 + budget: Duration, 177 + } 178 + 179 + impl KeyReprieve { 180 + pub const DEFAULT: Self = Self::from_secs(300, 21_600); 181 + 182 + pub const fn from_secs(retry: u64, budget: u64) -> Self { 183 + Self { 184 + retry: Duration::from_secs(retry), 185 + budget: Duration::from_secs(budget), 186 + } 187 + } 188 + 189 + pub const fn budgeted_for(self, ttl: KeyTtl) -> Self { 190 + match self.budget.as_secs() >= ttl.0.as_secs() { 191 + true => self, 192 + false => Self { 193 + retry: self.retry, 194 + budget: ttl.0, 195 + }, 196 + } 197 + } 198 + 199 + pub(crate) const fn first_failure(self, now: UnixSeconds) -> KeyLease { 200 + KeyLease { 201 + read_at: now, 202 + expires_at: after(now, self.retry), 203 + } 204 + } 205 + 206 + pub(crate) const fn extend(self, lease: KeyLease, now: UnixSeconds) -> Option<KeyLease> { 207 + let horizon = after(lease.read_at, self.budget); 208 + match horizon.get() > now.get() { 209 + false => None, 210 + true => { 211 + let retry = after(now, self.retry); 212 + let granted = match retry.get() < horizon.get() { 213 + true => retry, 214 + false => horizon, 215 + }; 216 + Some(KeyLease { 217 + read_at: lease.read_at, 218 + expires_at: match lease.expires_at.get() > granted.get() { 219 + true => lease.expires_at, 220 + false => granted, 221 + }, 222 + }) 223 + } 224 + } 225 + } 226 + } 227 + 228 + pub(crate) const fn whole_secs(span: Duration) -> i64 { 229 + let secs = span.as_secs(); 230 + match secs > i64::MAX as u64 { 231 + true => i64::MAX, 232 + false => secs as i64, 233 + } 234 + } 235 + 236 + const fn after(now: UnixSeconds, span: Duration) -> UnixSeconds { 237 + now.saturating_add_secs(whole_secs(span)) 238 + } 239 + 240 + #[derive(Debug, Clone, Copy, PartialEq, Eq)] 241 + pub enum KeyRecord { 242 + Stored, 243 + Unheld, 244 + Saturated, 245 + } 246 + 247 + #[derive(Debug, Clone, Copy, PartialEq, Eq)] 248 + pub enum KeyReprieved { 249 + Extended, 250 + Pending, 251 + Exhausted, 252 + } 253 + 254 + #[derive(Debug, Clone, Copy, PartialEq, Eq)] 255 + pub struct KeyBudget(usize); 256 + 257 + impl KeyBudget { 258 + pub const DEFAULT: Self = Self::from_mib(64); 259 + 260 + pub const fn from_mib(mib: usize) -> Self { 261 + Self(mib * 1024 * 1024) 262 + } 263 + 264 + pub const fn from_bytes(bytes: usize) -> Self { 265 + Self(bytes) 266 + } 267 + 268 + pub const fn get(self) -> usize { 269 + self.0 270 + } 271 + } 272 + 273 + pub struct MemberWork { 274 + pub unread: UnreadMembers, 275 + pub due: StaleMembers, 276 + pub kept: KeptAccounts, 277 + } 278 + 279 + pub struct KeyWork { 280 + pub generation: IndexGeneration, 281 + pub tracked: usize, 282 + pub hosted: HostedCoverage, 283 + pub pushers: Pushers, 284 + pub due: StalePushers, 285 + pub suspected: SuspectPushers, 286 + pub complete: bool, 287 + pub members: Resolved<MemberWork>, 288 + } 289 + 290 + #[derive(Debug, Clone, Copy, PartialEq, Eq)] 291 + enum Recheck { 292 + Renewals, 293 + Everything, 294 + } 295 + 296 + fn sorted(mut subjects: Vec<AccountDid>) -> Vec<AccountDid> { 297 + subjects.sort(); 298 + subjects.dedup(); 299 + subjects 300 + } 301 + 37 302 pub struct Index { 38 303 meta_path: PathBuf, 39 304 layout: Layout, ··· 43 308 collaborators: CollaboratorsProjection, 44 309 registry: RegistryProjection, 45 310 keys: KeyProjection, 311 + unreadable: scc::HashSet<RepoKey>, 46 312 generation: AtomicU64, 313 + generations: watch::Sender<IndexGeneration>, 47 314 } 48 315 49 316 impl Index { 50 317 pub fn new(meta_path: impl Into<PathBuf>, layout: Layout) -> Self { 318 + Self::with_key_budget(meta_path, layout, KeyBudget::DEFAULT) 319 + } 320 + 321 + pub fn with_key_budget( 322 + meta_path: impl Into<PathBuf>, 323 + layout: Layout, 324 + budget: KeyBudget, 325 + ) -> Self { 51 326 Self { 52 327 meta_path: meta_path.into(), 53 328 layout, ··· 56 331 blocklist: GrantSetProjection::new(), 57 332 collaborators: CollaboratorsProjection::new(), 58 333 registry: RegistryProjection::new(), 59 - keys: KeyProjection::new(), 334 + keys: KeyProjection::new(budget), 335 + unreadable: scc::HashSet::new(), 60 336 generation: AtomicU64::new(0), 337 + generations: watch::Sender::new(IndexGeneration::new(0)), 61 338 } 62 339 } 63 340 ··· 65 342 IndexGeneration(self.generation.load(Ordering::Acquire)) 66 343 } 67 344 345 + pub fn generations(&self) -> watch::Receiver<IndexGeneration> { 346 + self.generations.subscribe() 347 + } 348 + 68 349 fn bump_generation(&self) { 69 350 self.generation.fetch_add(1, Ordering::Release); 351 + self.generations.send_replace(self.generation()); 70 352 } 71 353 72 354 pub fn rebuild(&self) -> Result<(), IndexError> { ··· 76 358 Ok(()) 77 359 } 78 360 79 - pub fn warm_collaborators(&self) { 80 - self.hosted_repos().iter().for_each(|repo| { 81 - let _ = self.ensure_collaborators(repo); 82 - }); 361 + pub fn warm_collaborators(&self) -> usize { 362 + self.hosted_repos() 363 + .iter() 364 + .filter(|repo| match self.ensure_collaborators(repo) { 365 + Ok(()) => false, 366 + Err(error) => { 367 + tracing::warn!( 368 + repo = repo.as_str(), 369 + %error, 370 + "collaborators unread, the knot can't open the repo" 371 + ); 372 + true 373 + } 374 + }) 375 + .count() 376 + } 377 + 378 + fn unreadable_repo(&self, repo: &RepoDid) -> bool { 379 + self.interner 380 + .repo(repo) 381 + .is_some_and(|repo| self.unreadable.contains_sync(&repo)) 83 382 } 84 383 85 384 pub fn refresh_members(&self) -> Result<(), IndexError> { ··· 132 431 evacuated.iter().for_each(|repo| { 133 432 if let Some(key) = self.interner.repo(repo) { 134 433 self.collaborators.drop_repo(key); 434 + self.unreadable.remove_sync(&key); 135 435 } 136 436 }); 137 437 self.bump_generation(); ··· 146 446 } 147 447 148 448 pub fn refresh_collaborators(&self, repo: &RepoDid) -> Result<(), IndexError> { 449 + let repo_key = self.interner.intern_repo(repo); 450 + self.fold_collaborators(repo, repo_key) 451 + .inspect(|()| { 452 + self.unreadable.remove_sync(&repo_key); 453 + }) 454 + .inspect_err(|_| { 455 + let _ = self.unreadable.insert_sync(repo_key); 456 + }) 457 + } 458 + 459 + fn fold_collaborators(&self, repo: &RepoDid, repo_key: RepoKey) -> Result<(), IndexError> { 149 460 let git = self.layout.open(repo)?; 150 461 let store = CobStore::new(&git); 151 - let repo_key = self.interner.intern_repo(repo); 152 462 match store.list::<CollaboratorsCob>()?.as_slice() { 153 463 [] => self.collaborators.mark_repo_empty(repo_key), 154 464 [object] => { ··· 220 530 self.registry.hosted_repos(&self.interner) 221 531 } 222 532 223 - pub fn owner_of_key(&self, key: &OfferedKey) -> Resolved<Option<AccountDid>> { 224 - self.keys.owner(&self.interner, key) 533 + pub fn owner_of_key(&self, key: &OfferedKey, now: UnixSeconds) -> Resolved<Option<AccountDid>> { 534 + self.keys.owner(&self.interner, key, now) 535 + } 536 + 537 + pub fn keys(&self) -> KeySet<'_> { 538 + KeySet(self) 539 + } 540 + 541 + fn push_grants(&self) -> Resolved<Granted> { 542 + match self.registry.coverage() { 543 + Coverage::Warming => Resolved::Warming, 544 + Coverage::Ready => { 545 + let folded: Vec<Folded> = self 546 + .hosted_repos() 547 + .iter() 548 + .map( 549 + |repo| match (self.owner_of(repo), self.collaborators_of(repo)) { 550 + (Resolved::Ready(owner), Resolved::Ready(collaborators)) => { 551 + Folded::Grants( 552 + owner 553 + .map(AccountDid::from) 554 + .into_iter() 555 + .chain(collaborators) 556 + .collect(), 557 + ) 558 + } 559 + _ if self.unreadable_repo(repo) => Folded::Unreadable, 560 + _ => Folded::Pending, 561 + }, 562 + ) 563 + .collect(); 564 + let hosted = HostedCoverage::over( 565 + folded 566 + .iter() 567 + .filter(|repo| matches!(repo, Folded::Pending)) 568 + .count(), 569 + ); 570 + Resolved::Ready(Granted { 571 + subjects: sorted( 572 + folded 573 + .into_iter() 574 + .filter_map(Folded::into_grants) 575 + .flatten() 576 + .collect(), 577 + ), 578 + hosted, 579 + }) 580 + } 581 + } 582 + } 583 + 584 + fn member_grants(&self) -> Resolved<Vec<AccountDid>> { 585 + self.member_entries() 586 + .map(|entries| sorted(entries.into_iter().map(|grant| grant.subject).collect())) 225 587 } 226 588 227 - pub fn cache_key(&self, key: OfferedKey, did: &AccountDid) { 228 - self.keys.cache(&self.interner, key, did); 589 + fn due_among( 590 + &self, 591 + subjects: &[AccountDid], 592 + now: UnixSeconds, 593 + against: Recheck, 594 + ) -> Vec<AccountDid> { 595 + subjects 596 + .iter() 597 + .filter(|did| match against { 598 + Recheck::Everything => true, 599 + Recheck::Renewals => self.keys.renewal_due(&self.interner, did, now), 600 + }) 601 + .cloned() 602 + .collect() 229 603 } 230 604 231 605 pub fn coverage(&self) -> IndexCoverage { ··· 234 608 blocklist: self.blocklist.coverage(), 235 609 collaborators: self.collaborators.coverage(), 236 610 registry: self.registry.coverage(), 237 - keys: self.keys.coverage(), 611 + keys: self.keys.coverage(self.generation()), 238 612 } 239 613 } 240 614 } 615 + 616 + pub struct KeySet<'a>(&'a Index); 617 + 618 + impl KeySet<'_> { 619 + pub fn coverage(&self) -> Coverage { 620 + self.0.keys.coverage(self.0.generation()) 621 + } 622 + 623 + pub fn mark_ready(&self, generation: IndexGeneration) { 624 + self.0.keys.mark_ready(generation); 625 + } 626 + 627 + pub fn mark_warming(&self) { 628 + self.0.keys.mark_warming(); 629 + } 630 + 631 + pub fn record(&self, did: &AccountDid, keys: Vec<OfferedKey>, lease: KeyLease) -> KeyRecord { 632 + self.0.keys.record(&self.0.interner, did, keys, lease) 633 + } 634 + 635 + pub fn reprieve( 636 + &self, 637 + did: &AccountDid, 638 + now: UnixSeconds, 639 + grace: KeyReprieve, 640 + exhausted: KeyLease, 641 + ) -> KeyReprieved { 642 + self.0 643 + .keys 644 + .reprieve(&self.0.interner, did, now, grace, exhausted) 645 + } 646 + 647 + pub fn retain(&self, kept: &KeptAccounts) { 648 + self.0.keys.retain(&self.0.interner, kept.as_slice()); 649 + } 650 + 651 + pub fn is_fresh(&self, did: &AccountDid, now: UnixSeconds) -> bool { 652 + self.0.keys.is_fresh(&self.0.interner, did, now) 653 + } 654 + 655 + pub fn publisher_among( 656 + &self, 657 + candidates: &[AccountDid], 658 + key: &OfferedKey, 659 + now: UnixSeconds, 660 + ) -> Option<AccountDid> { 661 + self.0 662 + .keys 663 + .publisher_among(&self.0.interner, candidates, key, now) 664 + } 665 + 666 + pub fn note_miss(&self) { 667 + self.0.keys.suspect_now(); 668 + } 669 + 670 + pub fn any_unheld(&self) -> bool { 671 + self.0.keys.any_unheld() 672 + } 673 + 674 + pub fn work(&self, now: UnixSeconds, floor: SweepFloor) -> Resolved<KeyWork> { 675 + let index = self.0; 676 + let generation = index.generation(); 677 + let Resolved::Ready(granted) = index.push_grants() else { 678 + return Resolved::Warming; 679 + }; 680 + let pushers = granted.subjects; 681 + let against = match index.keys.take_suspicion(now, floor) { 682 + true => Recheck::Everything, 683 + false => Recheck::Renewals, 684 + }; 685 + let members = index.member_grants().map(|members| { 686 + let outside: Vec<AccountDid> = members 687 + .into_iter() 688 + .filter(|did| pushers.binary_search(did).is_err()) 689 + .collect(); 690 + let (unread, due) = index 691 + .due_among(&outside, now, against) 692 + .into_iter() 693 + .partition(|did| !index.keys.on_file(&index.interner, did)); 694 + MemberWork { 695 + unread: UnreadMembers(unread), 696 + due: StaleMembers(due), 697 + kept: KeptAccounts(pushers.iter().cloned().chain(outside).collect()), 698 + } 699 + }); 700 + let tracked = match &members { 701 + Resolved::Ready(work) => work.kept.len(), 702 + Resolved::Warming => pushers.len(), 703 + }; 704 + let due = index.due_among(&pushers, now, Recheck::Renewals); 705 + let suspected = match against { 706 + Recheck::Renewals => Vec::new(), 707 + Recheck::Everything => pushers 708 + .iter() 709 + .filter(|did| !index.keys.renewal_due(&index.interner, did, now)) 710 + .cloned() 711 + .collect(), 712 + }; 713 + let complete = granted.hosted == HostedCoverage::Whole 714 + && index.keys.all_live(&index.interner, &pushers, now); 715 + Resolved::Ready(KeyWork { 716 + generation, 717 + tracked, 718 + hosted: granted.hosted, 719 + pushers: Pushers(pushers), 720 + due: StalePushers(due), 721 + suspected: SuspectPushers(suspected), 722 + complete, 723 + members, 724 + }) 725 + } 726 + 727 + pub fn all_live(&self, pushers: &Pushers, now: UnixSeconds) -> bool { 728 + self.0 729 + .keys 730 + .all_live(&self.0.interner, pushers.as_slice(), now) 731 + } 732 + }

+418 -14

knot2/crates/knot-index/src/projections.rs

View file

··· 1 - use std::collections::{BTreeMap, BTreeSet}; 1 + use std::collections::{BTreeMap, BTreeSet, HashSet}; 2 2 use std::hash::{Hash, Hasher}; 3 3 use std::marker::PhantomData; 4 - use std::sync::Mutex; 4 + use std::sync::atomic::{AtomicBool, AtomicI64, AtomicU64, AtomicUsize, Ordering}; 5 + use std::sync::{Arc, Mutex}; 5 6 6 - use knot_cache::{Cache, EntryCount, Lru}; 7 7 use knot_cob::{Change, ChangeId, ChangePayload, Checkpoint, CobId, CobStore, Evaluate}; 8 8 use knot_cobs::{ 9 9 CollaboratorsChange, CollaboratorsCob, Grant, GrantChange, Registration, Registry, ··· 14 14 use crate::coverage::{Coverage, CoverageCell, Resolved}; 15 15 use crate::error::IndexError; 16 16 use crate::intern::{AccountKey, Interner, NameKey, OwnerKey, RepoKey, RkeyKey}; 17 - 18 - const KEY_CACHE_CAPACITY: usize = 16_384; 17 + use crate::{ 18 + IndexGeneration, KeyBudget, KeyLease, KeyRecord, KeyReprieve, KeyReprieved, SweepFloor, 19 + }; 19 20 20 21 #[derive(Debug, Clone, Copy)] 21 22 struct Provenance { ··· 818 819 } 819 820 } 820 821 822 + enum Reading { 823 + Published(Vec<Arc<OfferedKey>>), 824 + Unheld, 825 + Unread, 826 + } 827 + 828 + impl Reading { 829 + fn keys(&self) -> &[Arc<OfferedKey>] { 830 + match self { 831 + Reading::Published(keys) => keys, 832 + Reading::Unheld | Reading::Unread => &[], 833 + } 834 + } 835 + 836 + fn is_published(&self) -> bool { 837 + matches!(self, Reading::Published(_)) 838 + } 839 + } 840 + 841 + struct Held { 842 + reading: Reading, 843 + lease: KeyLease, 844 + } 845 + 846 + const PER_KEY_OVERHEAD: usize = 192; 847 + 848 + const PER_ACCOUNT_OVERHEAD: usize = 128; 849 + 850 + impl Held { 851 + fn bytes(&self) -> usize { 852 + PER_ACCOUNT_OVERHEAD 853 + + self 854 + .reading 855 + .keys() 856 + .iter() 857 + .map(|key| key.as_bytes().len() + PER_KEY_OVERHEAD) 858 + .sum::<usize>() 859 + } 860 + 861 + fn answers(&self, now: UnixSeconds) -> bool { 862 + self.reading.is_published() && self.lease.is_live(now) 863 + } 864 + 865 + fn settled(&self, now: UnixSeconds) -> bool { 866 + !matches!(self.reading, Reading::Unread) && self.lease.is_live(now) 867 + } 868 + 869 + fn renewal_due(&self, now: UnixSeconds) -> bool { 870 + match self.reading { 871 + Reading::Published(_) => self.lease.renewal_due(now), 872 + Reading::Unheld | Reading::Unread => !self.lease.is_live(now), 873 + } 874 + } 875 + 876 + fn publishes(&self, key: &OfferedKey, now: UnixSeconds) -> bool { 877 + self.answers(now) 878 + && self 879 + .reading 880 + .keys() 881 + .iter() 882 + .any(|published| published.as_ref() == key) 883 + } 884 + 885 + fn is_unheld(&self) -> bool { 886 + matches!(self.reading, Reading::Unheld) 887 + } 888 + } 889 + 890 + #[derive(Default)] 891 + struct Publishers(Vec<AccountKey>); 892 + 893 + impl Publishers { 894 + fn add(&mut self, account: AccountKey) { 895 + if let Err(at) = self.0.binary_search(&account) { 896 + self.0.insert(at, account); 897 + } 898 + } 899 + 900 + fn remove(&mut self, account: AccountKey) -> bool { 901 + if let Ok(at) = self.0.binary_search(&account) { 902 + self.0.remove(at); 903 + } 904 + self.0.is_empty() 905 + } 906 + 907 + fn to_vec(&self) -> Vec<AccountKey> { 908 + self.0.clone() 909 + } 910 + } 911 + 912 + const NEVER: u64 = u64::MAX; 913 + 821 914 pub(crate) struct KeyProjection { 822 - cache: Lru<OfferedKey, AccountKey>, 915 + owners: scc::HashMap<Arc<OfferedKey>, Publishers>, 916 + held: scc::HashMap<AccountKey, Held>, 917 + budget: KeyBudget, 918 + tracked_bytes: AtomicUsize, 919 + unheld: AtomicUsize, 920 + suspect: AtomicBool, 921 + swept_at: AtomicI64, 922 + ready_at: AtomicU64, 823 923 } 824 924 825 925 impl KeyProjection { 826 - pub(crate) fn new() -> Self { 926 + pub(crate) fn new(budget: KeyBudget) -> Self { 827 927 Self { 828 - cache: Lru::by_count(EntryCount::new(KEY_CACHE_CAPACITY as u64)), 928 + owners: scc::HashMap::new(), 929 + held: scc::HashMap::new(), 930 + budget, 931 + tracked_bytes: AtomicUsize::new(0), 932 + unheld: AtomicUsize::new(0), 933 + suspect: AtomicBool::new(false), 934 + swept_at: AtomicI64::new(i64::MIN), 935 + ready_at: AtomicU64::new(NEVER), 936 + } 937 + } 938 + 939 + pub(crate) fn coverage(&self, generation: IndexGeneration) -> Coverage { 940 + match self.ready_at.load(Ordering::Acquire) == generation.get() { 941 + true => Coverage::Ready, 942 + false => Coverage::Warming, 943 + } 944 + } 945 + 946 + pub(crate) fn suspect_now(&self) { 947 + self.suspect.store(true, Ordering::Release); 948 + } 949 + 950 + pub(crate) fn take_suspicion(&self, now: UnixSeconds, floor: SweepFloor) -> bool { 951 + let held_until = |swept: i64| swept.saturating_add(crate::whole_secs(floor.get())); 952 + match self.suspect.load(Ordering::Acquire) { 953 + false => false, 954 + true => { 955 + self.swept_at 956 + .fetch_update(Ordering::AcqRel, Ordering::Acquire, |swept| { 957 + (now.get() >= held_until(swept)).then_some(now.get()) 958 + }) 959 + .is_ok() 960 + && self.suspect.swap(false, Ordering::AcqRel) 961 + } 829 962 } 830 963 } 831 964 832 - pub(crate) fn coverage(&self) -> Coverage { 833 - Coverage::Ready 965 + pub(crate) fn mark_ready(&self, generation: IndexGeneration) { 966 + self.ready_at.store(generation.get(), Ordering::Release); 834 967 } 835 968 836 - pub(crate) fn cache(&self, interner: &Interner, key: OfferedKey, did: &AccountDid) { 837 - self.cache.insert(key, interner.intern_account(did)); 969 + pub(crate) fn mark_warming(&self) { 970 + self.ready_at.store(NEVER, Ordering::Release); 838 971 } 839 972 840 973 pub(crate) fn owner( 841 974 &self, 842 975 interner: &Interner, 843 976 key: &OfferedKey, 977 + now: UnixSeconds, 844 978 ) -> Resolved<Option<AccountDid>> { 979 + let publishers = self 980 + .owners 981 + .read_sync(key, |_, publishers| publishers.to_vec()) 982 + .unwrap_or_default(); 845 983 Resolved::Ready( 846 - self.cache 847 - .get(key) 984 + publishers 985 + .into_iter() 986 + .find(|account| { 987 + self.held_by(*account, |held| held.publishes(key, now)) == Some(true) 988 + }) 848 989 .map(|account| interner.resolve_account(account)), 849 990 ) 991 + } 992 + 993 + pub(crate) fn any_unheld(&self) -> bool { 994 + self.unheld.load(Ordering::Acquire) > 0 995 + } 996 + 997 + fn track_unheld(&self, lost: bool, gained: bool) { 998 + match (lost, gained) { 999 + (false, true) => { 1000 + self.unheld.fetch_add(1, Ordering::AcqRel); 1001 + } 1002 + (true, false) => { 1003 + self.unheld.fetch_sub(1, Ordering::AcqRel); 1004 + } 1005 + (true, true) | (false, false) => {} 1006 + } 1007 + } 1008 + 1009 + pub(crate) fn record( 1010 + &self, 1011 + interner: &Interner, 1012 + did: &AccountDid, 1013 + keys: Vec<OfferedKey>, 1014 + lease: KeyLease, 1015 + ) -> KeyRecord { 1016 + let account = interner.intern_account(did); 1017 + let incoming = Held { 1018 + reading: Reading::Published(keys.into_iter().map(Arc::new).collect()), 1019 + lease, 1020 + }; 1021 + let entry = self.held.entry_sync(account); 1022 + let held = match &entry { 1023 + scc::hash_map::Entry::Occupied(slot) => slot.get().bytes(), 1024 + scc::hash_map::Entry::Vacant(_) => 0, 1025 + }; 1026 + if self.reserve_bytes(incoming.bytes() as isize - held as isize) { 1027 + self.settle(entry, account, incoming); 1028 + return KeyRecord::Stored; 1029 + } 1030 + let unheld = Held { 1031 + reading: Reading::Unheld, 1032 + lease, 1033 + }; 1034 + if !self.reserve_bytes(unheld.bytes() as isize - held as isize) { 1035 + return KeyRecord::Saturated; 1036 + } 1037 + self.settle(entry, account, unheld); 1038 + KeyRecord::Unheld 1039 + } 1040 + 1041 + fn settle( 1042 + &self, 1043 + entry: scc::hash_map::Entry<'_, AccountKey, Held>, 1044 + account: AccountKey, 1045 + incoming: Held, 1046 + ) { 1047 + let published: HashSet<Arc<OfferedKey>> = incoming.reading.keys().iter().cloned().collect(); 1048 + let gained = incoming.is_unheld(); 1049 + match entry { 1050 + scc::hash_map::Entry::Occupied(mut slot) => { 1051 + let replaced = slot.insert(incoming); 1052 + self.track_unheld(replaced.is_unheld(), gained); 1053 + self.rewire(account, &published, replaced.reading.keys()); 1054 + } 1055 + scc::hash_map::Entry::Vacant(slot) => { 1056 + let _locked = slot.insert_entry(incoming); 1057 + self.track_unheld(false, gained); 1058 + self.rewire(account, &published, &[]); 1059 + } 1060 + } 1061 + } 1062 + 1063 + fn rewire( 1064 + &self, 1065 + account: AccountKey, 1066 + published: &HashSet<Arc<OfferedKey>>, 1067 + replaced: &[Arc<OfferedKey>], 1068 + ) { 1069 + published 1070 + .iter() 1071 + .for_each(|key| self.claim(Arc::clone(key), account)); 1072 + replaced 1073 + .iter() 1074 + .filter(|key| !published.contains(*key)) 1075 + .for_each(|key| self.disown(key, account)); 1076 + } 1077 + 1078 + pub(crate) fn reprieve( 1079 + &self, 1080 + interner: &Interner, 1081 + did: &AccountDid, 1082 + now: UnixSeconds, 1083 + grace: KeyReprieve, 1084 + exhausted: KeyLease, 1085 + ) -> KeyReprieved { 1086 + let account = interner.intern_account(did); 1087 + let mut slot = match self.held.entry_sync(account) { 1088 + scc::hash_map::Entry::Vacant(slot) => { 1089 + let pending = Held { 1090 + reading: Reading::Unread, 1091 + lease: grace.first_failure(now), 1092 + }; 1093 + if self.reserve_bytes(pending.bytes() as isize) { 1094 + let _locked = slot.insert_entry(pending); 1095 + } 1096 + return KeyReprieved::Pending; 1097 + } 1098 + scc::hash_map::Entry::Occupied(slot) => slot, 1099 + }; 1100 + match grace.extend(slot.get().lease, now) { 1101 + Some(lease) => { 1102 + let carried = slot.get().reading.is_published(); 1103 + slot.get_mut().lease = lease; 1104 + match carried { 1105 + true => KeyReprieved::Extended, 1106 + false => KeyReprieved::Pending, 1107 + } 1108 + } 1109 + None => { 1110 + let given_up = Held { 1111 + reading: Reading::Published(Vec::new()), 1112 + lease: exhausted, 1113 + }; 1114 + let _ = self.reserve_bytes(given_up.bytes() as isize - slot.get().bytes() as isize); 1115 + self.track_unheld(slot.get().is_unheld(), false); 1116 + self.disown_all(&slot, account); 1117 + *slot.get_mut() = given_up; 1118 + KeyReprieved::Exhausted 1119 + } 1120 + } 1121 + } 1122 + 1123 + pub(crate) fn on_file(&self, interner: &Interner, did: &AccountDid) -> bool { 1124 + interner 1125 + .account(did) 1126 + .is_some_and(|account| self.held.contains_sync(&account)) 1127 + } 1128 + 1129 + fn held_by(&self, account: AccountKey, ready: impl Fn(&Held) -> bool) -> Option<bool> { 1130 + self.held.read_sync(&account, |_, held| ready(held)) 1131 + } 1132 + 1133 + fn held_for( 1134 + &self, 1135 + interner: &Interner, 1136 + did: &AccountDid, 1137 + ready: impl Fn(&Held) -> bool, 1138 + ) -> Option<bool> { 1139 + interner 1140 + .account(did) 1141 + .and_then(|account| self.held_by(account, ready)) 1142 + } 1143 + 1144 + pub(crate) fn is_fresh(&self, interner: &Interner, did: &AccountDid, now: UnixSeconds) -> bool { 1145 + self.held_for(interner, did, |held| held.answers(now)) 1146 + .unwrap_or(false) 1147 + } 1148 + 1149 + pub(crate) fn renewal_due( 1150 + &self, 1151 + interner: &Interner, 1152 + did: &AccountDid, 1153 + now: UnixSeconds, 1154 + ) -> bool { 1155 + self.held_for(interner, did, |held| held.renewal_due(now)) 1156 + .unwrap_or(true) 1157 + } 1158 + 1159 + pub(crate) fn all_live( 1160 + &self, 1161 + interner: &Interner, 1162 + subjects: &[AccountDid], 1163 + now: UnixSeconds, 1164 + ) -> bool { 1165 + subjects.iter().all(|did| { 1166 + self.held_for(interner, did, |held| held.settled(now)) 1167 + .unwrap_or(false) 1168 + }) 1169 + } 1170 + 1171 + pub(crate) fn publisher_among( 1172 + &self, 1173 + interner: &Interner, 1174 + candidates: &[AccountDid], 1175 + key: &OfferedKey, 1176 + now: UnixSeconds, 1177 + ) -> Option<AccountDid> { 1178 + candidates 1179 + .iter() 1180 + .find(|did| { 1181 + self.held_for(interner, did, |held| held.publishes(key, now)) 1182 + .unwrap_or(false) 1183 + }) 1184 + .cloned() 1185 + } 1186 + 1187 + pub(crate) fn retain(&self, interner: &Interner, kept: &[AccountDid]) { 1188 + let keep: BTreeSet<AccountKey> = kept 1189 + .iter() 1190 + .filter_map(|did| interner.account(did)) 1191 + .collect(); 1192 + let mut released = Vec::new(); 1193 + self.held.iter_sync(|account, _| { 1194 + if !keep.contains(account) { 1195 + released.push(*account); 1196 + } 1197 + true 1198 + }); 1199 + released 1200 + .into_iter() 1201 + .for_each(|account| self.release(account)); 1202 + } 1203 + 1204 + fn release(&self, account: AccountKey) { 1205 + if let scc::hash_map::Entry::Occupied(slot) = self.held.entry_sync(account) { 1206 + self.evict(&slot, account); 1207 + let _ = slot.remove(); 1208 + } 1209 + } 1210 + 1211 + fn evict( 1212 + &self, 1213 + slot: &scc::hash_map::OccupiedEntry<'_, AccountKey, Held>, 1214 + account: AccountKey, 1215 + ) { 1216 + self.track_unheld(slot.get().is_unheld(), false); 1217 + self.disown_all(slot, account); 1218 + let _ = self.reserve_bytes(-(slot.get().bytes() as isize)); 1219 + } 1220 + 1221 + fn disown_all( 1222 + &self, 1223 + slot: &scc::hash_map::OccupiedEntry<'_, AccountKey, Held>, 1224 + account: AccountKey, 1225 + ) { 1226 + slot.get() 1227 + .reading 1228 + .keys() 1229 + .iter() 1230 + .for_each(|key| self.disown(key, account)); 1231 + } 1232 + 1233 + fn reserve_bytes(&self, growth: isize) -> bool { 1234 + self.tracked_bytes 1235 + .fetch_update(Ordering::AcqRel, Ordering::Acquire, |tracked| { 1236 + let next = tracked.saturating_add_signed(growth); 1237 + (growth <= 0 || next <= self.budget.get()).then_some(next) 1238 + }) 1239 + .is_ok() 1240 + } 1241 + 1242 + fn claim(&self, key: Arc<OfferedKey>, account: AccountKey) { 1243 + self.owners 1244 + .entry_sync(key) 1245 + .or_default() 1246 + .get_mut() 1247 + .add(account); 1248 + } 1249 + 1250 + fn disown(&self, key: &OfferedKey, account: AccountKey) { 1251 + let _ = self 1252 + .owners 1253 + .remove_if_sync(key, |publishers| publishers.remove(account)); 850 1254 } 851 1255 }

+3 -3

knot2/crates/knot-index/tests/lifecycle.rs

View file

··· 53 53 blocklist: Coverage::Ready, 54 54 collaborators: Coverage::Ready, 55 55 registry: Coverage::Ready, 56 - keys: Coverage::Ready, 56 + keys: Coverage::Warming, 57 57 } 58 58 ); 59 59 } ··· 88 88 assert_eq!(index.coverage().members, Coverage::Warming); 89 89 90 90 assert_eq!( 91 - index.owner_of_key(&OfferedKey::from_bytes(vec![1, 2, 3])), 91 + index.owner_of_key(&OfferedKey::from_bytes(vec![1, 2, 3]), at(0)), 92 92 Resolved::Ready(None), 93 - "key cache is operational from boot, never warming" 93 + "a key lookup answers from the first request, even while the key set is warming" 94 94 ); 95 95 } 96 96

+1 -31

knot2/crates/knot-index/tests/projections.rs

View file

··· 4 4 use knot_cob::{ChangePayload, CobHome, CobId, CobStore}; 5 5 use knot_cobs::{CollaboratorsChange, MembersChange, RegistryChange, Removal, Rename, RepoRef}; 6 6 use knot_git::{RefUpdate, Repo}; 7 - use knot_index::{Coverage, IndexError, OfferedKey, Resolved}; 7 + use knot_index::{Coverage, IndexError, Resolved}; 8 8 use knot_types::{ClonePath, RefName, RepoName}; 9 9 use serde::{Deserialize, Serialize}; 10 10 ··· 535 535 index.is_collaborator(&repo, &acc("lyna")), 536 536 Resolved::Ready(true), 537 537 "deregister and re-register within single delta leaves repo hosted, so its collaborators survive" 538 - ); 539 - } 540 - 541 - #[test] 542 - fn key_cache_evicts_least_recently_used() { 543 - const CAP: u32 = 16_384; 544 - let world = World::new(); 545 - let index = world.index(); 546 - let key = |i: u32| OfferedKey::from_bytes(i.to_le_bytes().to_vec()); 547 - 548 - (0..CAP).for_each(|i| index.cache_key(key(i), &acc("nel"))); 549 - assert_eq!( 550 - index.owner_of_key(&key(0)), 551 - Resolved::Ready(Some(acc("nel"))) 552 - ); 553 - index.cache_key(key(CAP), &acc("nel")); 554 - 555 - assert_eq!( 556 - index.owner_of_key(&key(1)), 557 - Resolved::Ready(None), 558 - "least-recently-used key is evicted" 559 - ); 560 - assert_eq!( 561 - index.owner_of_key(&key(0)), 562 - Resolved::Ready(Some(acc("nel"))), 563 - "recently-used key survives despite being inserted first" 564 - ); 565 - assert_eq!( 566 - index.owner_of_key(&key(CAP)), 567 - Resolved::Ready(Some(acc("nel"))) 568 538 ); 569 539 } 570 540

+2 -3

knot2/crates/knot-ssh/src/exec.rs

View file

··· 805 805 _ => Vec::new(), 806 806 }; 807 807 let candidates: Vec<AccountDid> = owner.into_iter().chain(collaborators).collect(); 808 - if let Resolved::Ready(Some(cached)) = state.index.owner_of_key(key) 808 + let now = state.atproto.now().seconds(); 809 + if let Resolved::Ready(Some(cached)) = state.index.owner_of_key(key, now) 809 810 && candidates.contains(&cached) 810 811 { 811 812 return Some(cached); ··· 813 814 let _permit = state.slots.resolve.acquire().await; 814 815 let matches = futures::stream::iter(candidates).filter_map(|did| async move { 815 816 let keys = state.atproto.resolve_pubkeys(&did).await.ok()?; 816 - keys.iter() 817 - .for_each(|resolved| state.index.cache_key(resolved.clone(), &did)); 818 817 keys.iter().any(|resolved| resolved == key).then_some(did) 819 818 }); 820 819 futures::pin_mut!(matches);

+3 -2

knot2/crates/knot-ssh/tests/ssh_push.rs

View file

··· 1034 1034 .unwrap() 1035 1035 .to_bytes() 1036 1036 .unwrap(); 1037 - fx.index.cache_key( 1038 - knot_types::OfferedKey::from_bytes(blob), 1037 + fx.index.keys().record( 1039 1038 &AccountDid::new("did:plc:whelk").unwrap(), 1039 + vec![knot_types::OfferedKey::from_bytes(blob)], 1040 + knot_index::KeyTtl::from_secs(u32::MAX.into()).lease_from(knot_types::UnixSeconds::new(0)), 1040 1041 ); 1041 1042 let (ok, out) = push( 1042 1043 &fx.work,

+12 -1

knot2/crates/knot-types/src/ids.rs

View file

··· 434 434 } 435 435 } 436 436 437 - #[derive(Clone)] 437 + #[derive(Clone, PartialEq, Eq)] 438 438 pub enum OwnerRef { 439 439 Did(OwnerDid), 440 440 Handle(Handle), ··· 761 761 762 762 pub const fn next(self) -> Self { 763 763 Self(self.0.saturating_add(1)) 764 + } 765 + 766 + pub const fn seconds(self) -> UnixSeconds { 767 + UnixSeconds((self.0 / 1_000_000) as i64) 764 768 } 765 769 } 766 770 ··· 1348 1352 ); 1349 1353 assert_eq!(base.get(), 1_000); 1350 1354 assert_eq!(base.to_string(), "1000"); 1355 + } 1356 + 1357 + #[test] 1358 + fn micros_truncate_to_the_second_they_fall_in() { 1359 + assert_eq!(UnixMicros::new(1_999_999).seconds(), UnixSeconds::new(1)); 1360 + assert_eq!(UnixMicros::new(2_000_000).seconds(), UnixSeconds::new(2)); 1361 + assert_eq!(UnixMicros::new(0).seconds(), UnixSeconds::new(0)); 1351 1362 } 1352 1363 1353 1364 #[test]

Configure Feed

Configure Feed