alice.pds.demo.boltless.dev
This repository has no description
did:plc:3fwecdnvtcscjnrx2p4n7a…
1mod coverage;
2mod error;
3mod intern;
4mod projections;
5
6pub use coverage::{Coverage, Resolved};
7pub use error::IndexError;
8pub use knot_types::OfferedKey;
9
10use std::path::PathBuf;
11use std::sync::atomic::{AtomicU64, Ordering};
12use std::time::Duration;
13
14use knot_cob::{ChangePayload, CobStore};
15use knot_cobs::{
16 BlocklistChange, BlocklistCob, CollaboratorsChange, CollaboratorsCob, Grant, MembersChange,
17 MembersCob, RegistryChange, RepoRegistryCob,
18};
19use knot_git::{Layout, Repo};
20use knot_types::{AccountDid, ClonePath, OwnerDid, RepoDid, RepoRkey, UnixSeconds};
21use tokio::sync::watch;
22
23use intern::{Interner, RepoKey};
24use projections::{CollaboratorsProjection, GrantSetProjection, KeyProjection, RegistryProjection};
25
26knot_types::scalar_newtype! {
27 pub struct IndexGeneration(u64);
28}
29
30#[derive(Debug, Clone, Copy, PartialEq, Eq)]
31pub struct IndexCoverage {
32 pub members: Coverage,
33 pub blocklist: Coverage,
34 pub collaborators: Coverage,
35 pub registry: Coverage,
36 pub keys: Coverage,
37}
38
39macro_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
68account_list!(
69 Pushers,
70 KeptAccounts,
71 StalePushers,
72 SuspectPushers,
73 UnreadMembers,
74 StaleMembers,
75);
76
77#[derive(Debug, Clone, Copy, PartialEq, Eq)]
78pub enum HostedCoverage {
79 Whole,
80 Partial { unread: usize },
81}
82
83impl HostedCoverage {
84 const fn over(unread: usize) -> Self {
85 match unread {
86 0 => Self::Whole,
87 unread => Self::Partial { unread },
88 }
89 }
90}
91
92struct Granted {
93 subjects: Vec<AccountDid>,
94 hosted: HostedCoverage,
95}
96
97enum Folded {
98 Grants(Vec<AccountDid>),
99 Pending,
100 Unreadable,
101}
102
103impl 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)]
113pub struct SweepFloor(Duration);
114
115impl 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)]
128pub struct KeyTtl(Duration);
129
130impl 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)]
157pub struct KeyLease {
158 read_at: UnixSeconds,
159 expires_at: UnixSeconds,
160}
161
162impl 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)]
174pub struct KeyReprieve {
175 retry: Duration,
176 budget: Duration,
177}
178
179impl 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
228pub(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
236const fn after(now: UnixSeconds, span: Duration) -> UnixSeconds {
237 now.saturating_add_secs(whole_secs(span))
238}
239
240#[derive(Debug, Clone, Copy, PartialEq, Eq)]
241pub enum KeyRecord {
242 Stored,
243 Unheld,
244 Saturated,
245}
246
247#[derive(Debug, Clone, Copy, PartialEq, Eq)]
248pub enum KeyReprieved {
249 Extended,
250 Pending,
251 Exhausted,
252}
253
254#[derive(Debug, Clone, Copy, PartialEq, Eq)]
255pub struct KeyBudget(usize);
256
257impl 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
273pub struct MemberWork {
274 pub unread: UnreadMembers,
275 pub due: StaleMembers,
276 pub kept: KeptAccounts,
277}
278
279pub 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)]
291enum Recheck {
292 Renewals,
293 Everything,
294}
295
296fn sorted(mut subjects: Vec<AccountDid>) -> Vec<AccountDid> {
297 subjects.sort();
298 subjects.dedup();
299 subjects
300}
301
302pub struct Index {
303 meta_path: PathBuf,
304 layout: Layout,
305 interner: Interner,
306 members: GrantSetProjection<MembersCob>,
307 blocklist: GrantSetProjection<BlocklistCob>,
308 collaborators: CollaboratorsProjection,
309 registry: RegistryProjection,
310 keys: KeyProjection,
311 unreadable: scc::HashSet<RepoKey>,
312 generation: AtomicU64,
313 generations: watch::Sender<IndexGeneration>,
314}
315
316impl Index {
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 {
326 Self {
327 meta_path: meta_path.into(),
328 layout,
329 interner: Interner::new(),
330 members: GrantSetProjection::new(),
331 blocklist: GrantSetProjection::new(),
332 collaborators: CollaboratorsProjection::new(),
333 registry: RegistryProjection::new(),
334 keys: KeyProjection::new(budget),
335 unreadable: scc::HashSet::new(),
336 generation: AtomicU64::new(0),
337 generations: watch::Sender::new(IndexGeneration::new(0)),
338 }
339 }
340
341 pub fn generation(&self) -> IndexGeneration {
342 IndexGeneration(self.generation.load(Ordering::Acquire))
343 }
344
345 pub fn generations(&self) -> watch::Receiver<IndexGeneration> {
346 self.generations.subscribe()
347 }
348
349 fn bump_generation(&self) {
350 self.generation.fetch_add(1, Ordering::Release);
351 self.generations.send_replace(self.generation());
352 }
353
354 pub fn rebuild(&self) -> Result<(), IndexError> {
355 self.refresh_members()?;
356 self.refresh_blocklist()?;
357 self.refresh_registry()?;
358 Ok(())
359 }
360
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))
382 }
383
384 pub fn refresh_members(&self) -> Result<(), IndexError> {
385 let meta = Repo::open(&self.meta_path)?;
386 let store = CobStore::new(&meta);
387 match store.list::<MembersCob>()?.as_slice() {
388 [] => self.members.reset(),
389 [object] => self.members.refresh(&self.interner, &store, *object)?,
390 many => {
391 return Err(IndexError::Ambiguous {
392 type_name: MembersChange::type_name(),
393 count: many.len(),
394 });
395 }
396 }
397 self.bump_generation();
398 Ok(())
399 }
400
401 pub fn refresh_blocklist(&self) -> Result<(), IndexError> {
402 let meta = Repo::open(&self.meta_path)?;
403 let store = CobStore::new(&meta);
404 match store.list::<BlocklistCob>()?.as_slice() {
405 [] => self.blocklist.reset(),
406 [object] => self.blocklist.refresh(&self.interner, &store, *object)?,
407 many => {
408 return Err(IndexError::Ambiguous {
409 type_name: BlocklistChange::type_name(),
410 count: many.len(),
411 });
412 }
413 }
414 self.bump_generation();
415 Ok(())
416 }
417
418 pub fn refresh_registry(&self) -> Result<(), IndexError> {
419 let meta = Repo::open(&self.meta_path)?;
420 let store = CobStore::new(&meta);
421 let evacuated = match store.list::<RepoRegistryCob>()?.as_slice() {
422 [] => self.registry.reset(&self.interner),
423 [object] => self.registry.refresh(&self.interner, &store, *object)?,
424 many => {
425 return Err(IndexError::Ambiguous {
426 type_name: RegistryChange::type_name(),
427 count: many.len(),
428 });
429 }
430 };
431 evacuated.iter().for_each(|repo| {
432 if let Some(key) = self.interner.repo(repo) {
433 self.collaborators.drop_repo(key);
434 self.unreadable.remove_sync(&key);
435 }
436 });
437 self.bump_generation();
438 Ok(())
439 }
440
441 pub fn ensure_collaborators(&self, repo: &RepoDid) -> Result<(), IndexError> {
442 if self.collaborators.is_folded(&self.interner, repo) {
443 return Ok(());
444 }
445 self.refresh_collaborators(repo)
446 }
447
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> {
460 let git = self.layout.open(repo)?;
461 let store = CobStore::new(&git);
462 match store.list::<CollaboratorsCob>()?.as_slice() {
463 [] => self.collaborators.mark_repo_empty(repo_key),
464 [object] => {
465 self.collaborators
466 .refresh_repo(&self.interner, &store, repo_key, *object)?
467 }
468 many => {
469 return Err(IndexError::Ambiguous {
470 type_name: CollaboratorsChange::type_name(),
471 count: many.len(),
472 });
473 }
474 }
475 self.bump_generation();
476 Ok(())
477 }
478
479 pub fn is_member(&self, did: &AccountDid) -> Resolved<bool> {
480 self.members.contains(&self.interner, did)
481 }
482
483 pub fn member_entries(&self) -> Resolved<Vec<Grant>> {
484 self.members.entries(&self.interner)
485 }
486
487 pub fn is_blocked(&self, did: &AccountDid) -> Resolved<bool> {
488 self.blocklist.contains(&self.interner, did)
489 }
490
491 pub fn blocked_entries(&self) -> Resolved<Vec<Grant>> {
492 self.blocklist.entries(&self.interner)
493 }
494
495 pub fn is_collaborator(&self, repo: &RepoDid, did: &AccountDid) -> Resolved<bool> {
496 self.collaborators.contains(&self.interner, repo, did)
497 }
498
499 pub fn collaborator_entries(&self, repo: &RepoDid) -> Resolved<Vec<Grant>> {
500 self.collaborators.entries(&self.interner, repo)
501 }
502
503 pub fn collaborators_of(&self, repo: &RepoDid) -> Resolved<Vec<AccountDid>> {
504 self.collaborator_entries(repo)
505 .map(|entries| entries.into_iter().map(|grant| grant.subject).collect())
506 }
507
508 pub fn resolve_repo(&self, owner: &OwnerDid, rkey: &RepoRkey) -> Resolved<Option<RepoDid>> {
509 self.registry.resolve(&self.interner, owner, rkey)
510 }
511
512 pub fn resolve_clone_path(
513 &self,
514 owner: &OwnerDid,
515 path: &ClonePath,
516 ) -> Resolved<Option<RepoDid>> {
517 self.registry
518 .resolve_clone_path(&self.interner, owner, path)
519 }
520
521 pub fn owner_of(&self, repo: &RepoDid) -> Resolved<Option<OwnerDid>> {
522 self.registry.owner_of(&self.interner, repo)
523 }
524
525 pub fn rkey_of(&self, repo: &RepoDid) -> Resolved<Option<RepoRkey>> {
526 self.registry.rkey_of(&self.interner, repo)
527 }
528
529 pub fn hosted_repos(&self) -> Vec<RepoDid> {
530 self.registry.hosted_repos(&self.interner)
531 }
532
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()))
587 }
588
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()
603 }
604
605 pub fn coverage(&self) -> IndexCoverage {
606 IndexCoverage {
607 members: self.members.coverage(),
608 blocklist: self.blocklist.coverage(),
609 collaborators: self.collaborators.coverage(),
610 registry: self.registry.coverage(),
611 keys: self.keys.coverage(self.generation()),
612 }
613 }
614}
615
616pub struct KeySet<'a>(&'a Index);
617
618impl 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}