alice.pds.demo.boltless.dev
This repository has no description
34 kB
1129 lines
did:plc:3fwecdnvtcscjnrx2p4n7a…
1use std::collections::{BTreeSet, HashMap, VecDeque};
2use std::net::IpAddr;
3use std::sync::{Arc, Mutex};
4
5use serde::ser::SerializeMap;
6use serde::{Serialize, Serializer};
7use serde_json::Value;
8use tokio::sync::{OwnedSemaphorePermit, Semaphore, watch};
9
10use knot_runtime::{Clock, UnixMicros};
11use knot_types::{
12 AccountDid, ChangedFiles, Email, LanguageBytes, LanguageName, ObjectFormat, Oid, OwnerDid,
13 PushOptions, RefName, RefTransition, RepoDid, RepoPath, Tid,
14};
15
16#[derive(Debug, Clone, Copy, PartialEq, Eq, PartialOrd, Ord, Serialize)]
17#[serde(transparent)]
18pub struct EventCursor(i64);
19
20impl EventCursor {
21 pub const START: Self = Self(0);
22
23 pub fn new(nanos: i64) -> Self {
24 Self(nanos)
25 }
26
27 pub fn get(self) -> i64 {
28 self.0
29 }
30
31 fn from_unix_micros(micros: UnixMicros) -> Self {
32 Self((micros.get() as i64).saturating_mul(1_000))
33 }
34}
35
36pub trait Publish: Serialize {
37 const NSID: &'static str;
38}
39
40#[derive(Debug, Clone, Serialize)]
41pub struct Event {
42 pub rkey: Tid,
43 pub nsid: &'static str,
44 #[serde(rename = "event")]
45 pub payload: Value,
46 pub created: EventCursor,
47}
48
49// `sh.tangled.git.refUpdate` requires ref, oldSha and newSha,
50// so a record about no ref at all sends "" for the three rather than `null`.
51#[derive(Debug, Clone)]
52enum RefChange {
53 Absent,
54 Applied {
55 ref_name: RefName,
56 old_sha: Oid,
57 new_sha: Oid,
58 },
59}
60
61impl Serialize for RefChange {
62 fn serialize<S: Serializer>(&self, serializer: S) -> Result<S::Ok, S::Error> {
63 let mut map = serializer.serialize_map(Some(3))?;
64 match self {
65 Self::Absent => {
66 map.serialize_entry("ref", "")?;
67 map.serialize_entry("oldSha", "")?;
68 map.serialize_entry("newSha", "")?;
69 }
70 Self::Applied {
71 ref_name,
72 old_sha,
73 new_sha,
74 } => {
75 map.serialize_entry("ref", ref_name.as_str())?;
76 map.serialize_entry("oldSha", &old_sha.to_hex())?;
77 map.serialize_entry("newSha", &new_sha.to_hex())?;
78 }
79 }
80 map.end()
81 }
82}
83
84#[derive(Debug, Clone, Serialize)]
85pub struct GitRefUpdate {
86 #[serde(rename = "$type")]
87 record_type: &'static str,
88 #[serde(rename = "changedFiles", skip_serializing_if = "Vec::is_empty")]
89 changed_files: Vec<RepoPath>,
90 #[serde(rename = "committerDid")]
91 committer_did: AccountDid,
92 meta: Option<RefUpdateMeta>,
93 #[serde(rename = "ownerDid", skip_serializing_if = "Option::is_none")]
94 owner_did: Option<OwnerDid>,
95 #[serde(rename = "pushOptions", skip_serializing_if = "PushOptions::is_empty")]
96 push_options: PushOptions,
97 #[serde(flatten)]
98 change: RefChange,
99 repo: RepoDid,
100}
101
102impl GitRefUpdate {
103 pub fn new(repo: RepoDid, owner: Option<OwnerDid>, committer: AccountDid) -> Self {
104 Self {
105 record_type: Self::NSID,
106 changed_files: Vec::new(),
107 committer_did: committer,
108 meta: None,
109 owner_did: owner,
110 push_options: PushOptions::default(),
111 change: RefChange::Absent,
112 repo,
113 }
114 }
115
116 pub fn on_ref(
117 mut self,
118 ref_name: RefName,
119 transition: RefTransition,
120 format: ObjectFormat,
121 ) -> Self {
122 self.change = RefChange::Applied {
123 ref_name,
124 old_sha: transition.old_oid().unwrap_or_else(|| format.null_oid()),
125 new_sha: transition.new_oid().unwrap_or_else(|| format.null_oid()),
126 };
127 self
128 }
129
130 pub fn with_changed_files(mut self, changed: ChangedFiles) -> Self {
131 self.changed_files = changed.into_paths();
132 self
133 }
134
135 pub fn with_push_options(mut self, options: &PushOptions) -> Self {
136 self.push_options = options.clone();
137 self
138 }
139
140 pub fn with_meta(mut self, meta: RefUpdateMeta) -> Self {
141 self.meta = Some(meta);
142 self
143 }
144}
145
146impl Publish for GitRefUpdate {
147 const NSID: &'static str = "sh.tangled.git.refUpdate";
148}
149
150#[derive(Debug, Clone, Serialize)]
151pub struct RefUpdateMeta {
152 #[serde(rename = "isDefaultRef")]
153 is_default_ref: bool,
154 #[serde(rename = "commitCount")]
155 commit_count: CommitCountBreakdown,
156 #[serde(rename = "langBreakdown", skip_serializing_if = "Option::is_none")]
157 lang_breakdown: Option<LangBreakdown>,
158}
159
160impl RefUpdateMeta {
161 pub fn new(
162 is_default_ref: bool,
163 by_email: Vec<EmailCommitCount>,
164 languages: Vec<LanguageSize>,
165 ) -> Self {
166 Self {
167 is_default_ref,
168 commit_count: CommitCountBreakdown {
169 by_email: (!by_email.is_empty()).then_some(by_email),
170 },
171 lang_breakdown: (!languages.is_empty()).then_some(LangBreakdown {
172 inputs: Some(languages),
173 }),
174 }
175 }
176}
177
178#[derive(Debug, Clone, Serialize)]
179struct CommitCountBreakdown {
180 #[serde(rename = "byEmail", skip_serializing_if = "Option::is_none")]
181 by_email: Option<Vec<EmailCommitCount>>,
182}
183
184#[derive(Debug, Default, Clone, Copy, PartialEq, Eq, PartialOrd, Ord, Serialize)]
185#[serde(transparent)]
186pub struct CommitCount(u64);
187
188impl CommitCount {
189 pub const fn new(count: u64) -> Self {
190 Self(count)
191 }
192
193 pub const fn get(self) -> u64 {
194 self.0
195 }
196
197 pub const fn succ(self) -> Self {
198 Self(self.0.saturating_add(1))
199 }
200}
201
202#[derive(Debug, Clone, Serialize)]
203pub struct EmailCommitCount {
204 email: Email,
205 count: CommitCount,
206}
207
208impl EmailCommitCount {
209 pub fn new(email: Email, count: CommitCount) -> Self {
210 Self { email, count }
211 }
212}
213
214#[derive(Debug, Clone, Serialize)]
215struct LangBreakdown {
216 #[serde(skip_serializing_if = "Option::is_none")]
217 inputs: Option<Vec<LanguageSize>>,
218}
219
220#[derive(Debug, Clone, Serialize)]
221pub struct LanguageSize {
222 lang: LanguageName,
223 size: LanguageBytes,
224}
225
226impl LanguageSize {
227 pub fn new(lang: LanguageName, size: LanguageBytes) -> Self {
228 Self { lang, size }
229 }
230}
231
232#[derive(Debug, Clone, Copy, Serialize)]
233#[serde(rename_all = "lowercase")]
234enum AclOp {
235 Add,
236 Remove,
237}
238
239#[derive(Debug, Clone, Serialize)]
240pub struct KnotMemberUpdate {
241 op: AclOp,
242 subject: AccountDid,
243}
244
245impl KnotMemberUpdate {
246 pub fn added(subject: AccountDid) -> Self {
247 Self {
248 op: AclOp::Add,
249 subject,
250 }
251 }
252
253 pub fn removed(subject: AccountDid) -> Self {
254 Self {
255 op: AclOp::Remove,
256 subject,
257 }
258 }
259}
260
261impl Publish for KnotMemberUpdate {
262 const NSID: &'static str = "sh.tangled.knot.memberUpdate";
263}
264
265#[derive(Debug, Clone, Serialize)]
266pub struct RepoCollaboratorUpdate {
267 op: AclOp,
268 subject: AccountDid,
269 repo: RepoDid,
270}
271
272impl RepoCollaboratorUpdate {
273 pub fn added(subject: AccountDid, repo: RepoDid) -> Self {
274 Self {
275 op: AclOp::Add,
276 subject,
277 repo,
278 }
279 }
280
281 pub fn removed(subject: AccountDid, repo: RepoDid) -> Self {
282 Self {
283 op: AclOp::Remove,
284 subject,
285 repo,
286 }
287 }
288}
289
290impl Publish for RepoCollaboratorUpdate {
291 const NSID: &'static str = "sh.tangled.repo.collaboratorUpdate";
292}
293
294#[derive(Debug, Clone, Copy, PartialEq, Eq)]
295pub struct ReplayEvents(std::num::NonZeroUsize);
296
297impl ReplayEvents {
298 pub fn new(value: usize) -> Option<Self> {
299 std::num::NonZeroUsize::new(value).map(Self)
300 }
301
302 pub fn get(self) -> usize {
303 self.0.get()
304 }
305}
306
307#[derive(Debug, Clone, Copy, PartialEq, Eq)]
308pub struct ReplayBytes(std::num::NonZeroUsize);
309
310impl ReplayBytes {
311 pub fn new(value: usize) -> Option<Self> {
312 std::num::NonZeroUsize::new(value).map(Self)
313 }
314
315 pub fn get(self) -> usize {
316 self.0.get()
317 }
318}
319
320#[derive(Debug, Clone, Copy, PartialEq, Eq)]
321pub struct ReplayBounds {
322 events: ReplayEvents,
323 bytes: ReplayBytes,
324}
325
326impl ReplayBounds {
327 pub fn new(events: ReplayEvents, bytes: ReplayBytes) -> Self {
328 Self { events, bytes }
329 }
330}
331
332#[derive(Debug, Clone, Copy, PartialEq, Eq)]
333pub enum BatchEnd {
334 CaughtUp,
335 Bounded,
336}
337
338pub struct Replayed {
339 pub events: Vec<Arc<Event>>,
340 pub end: BatchEnd,
341}
342
343struct Entry {
344 event: Arc<Event>,
345 bytes: usize,
346}
347
348struct Ring {
349 entries: VecDeque<Entry>,
350 bytes: usize,
351 last_micros: UnixMicros,
352 pending: BTreeSet<EventCursor>,
353}
354
355struct Inner {
356 bounds: ReplayBounds,
357 ring: Mutex<Ring>,
358 head: watch::Sender<EventCursor>,
359}
360
361impl Inner {
362 fn lock(&self) -> std::sync::MutexGuard<'_, Ring> {
363 self.ring
364 .lock()
365 .unwrap_or_else(|poisoned| poisoned.into_inner())
366 }
367
368 // leave that guiness be, boy! it needs to
369 fn settle(&self, ring: std::sync::MutexGuard<'_, Ring>) {
370 let head = stable_head(&ring);
371 drop(ring);
372 self.head.send_if_modified(|current| {
373 let changed = *current != head;
374 *current = head;
375 changed
376 });
377 }
378}
379
380fn stable_head(ring: &Ring) -> EventCursor {
381 let stable = match ring.pending.iter().next().copied() {
382 Some(horizon) => ring
383 .entries
384 .partition_point(|entry| entry.event.created < horizon),
385 None => ring.entries.len(),
386 };
387 stable
388 .checked_sub(1)
389 .and_then(|index| ring.entries.get(index))
390 .map(|entry| entry.event.created)
391 .unwrap_or(EventCursor::START)
392}
393
394fn value_bytes(value: &Value) -> usize {
395 let node = std::mem::size_of::<Value>();
396 match value {
397 Value::Null | Value::Bool(_) | Value::Number(_) => node,
398 Value::String(text) => node + text.len(),
399 Value::Array(items) => node + items.iter().map(value_bytes).sum::<usize>(),
400 Value::Object(fields) => {
401 node + fields
402 .iter()
403 .map(|(key, field)| key.len() + value_bytes(field))
404 .sum::<usize>()
405 }
406 }
407}
408
409fn insert_sorted(ring: &mut Ring, event: Event, bounds: ReplayBounds) {
410 let bytes = std::mem::size_of::<Event>() + value_bytes(&event.payload);
411 let position = ring
412 .entries
413 .partition_point(|existing| existing.event.created < event.created);
414 ring.entries.insert(
415 position,
416 Entry {
417 event: Arc::new(event),
418 bytes,
419 },
420 );
421 ring.bytes += bytes;
422 evict_oldest(ring, bounds);
423}
424
425fn evict_oldest(ring: &mut Ring, bounds: ReplayBounds) {
426 let over = ring.entries.len() > bounds.events.get()
427 || (ring.bytes > bounds.bytes.get() && ring.entries.len() > 1);
428 if let Some(evicted) = over.then(|| ring.entries.pop_front()).flatten() {
429 ring.bytes -= evicted.bytes;
430 evict_oldest(ring, bounds);
431 }
432}
433
434pub struct EventLog<C> {
435 clock: C,
436 inner: Arc<Inner>,
437}
438
439impl<C: Clock> EventLog<C> {
440 pub fn new(clock: C, bounds: ReplayBounds) -> Self {
441 Self {
442 clock,
443 inner: Arc::new(Inner {
444 bounds,
445 ring: Mutex::new(Ring {
446 entries: VecDeque::new(),
447 bytes: 0,
448 last_micros: UnixMicros::new(0),
449 pending: BTreeSet::new(),
450 }),
451 head: watch::Sender::new(EventCursor::START),
452 }),
453 }
454 }
455
456 fn next_cursor(&self, ring: &mut Ring) -> (UnixMicros, EventCursor) {
457 let micros = self.clock.now_unix_micros().max(ring.last_micros.next());
458 ring.last_micros = micros;
459 (micros, EventCursor::from_unix_micros(micros))
460 }
461
462 pub fn publish<P: Publish>(&self, payload: &P) -> EventCursor {
463 let payload = serde_json::to_value(payload).expect("event payload serializes to JSON");
464 let mut ring = self.inner.lock();
465 let (micros, created) = self.next_cursor(&mut ring);
466 insert_sorted(
467 &mut ring,
468 Event {
469 rkey: Tid::from_time(micros.get(), 0),
470 nsid: P::NSID,
471 payload,
472 created,
473 },
474 self.inner.bounds,
475 );
476 self.inner.settle(ring);
477 created
478 }
479
480 pub fn reserve(&self) -> Reservation {
481 let mut ring = self.inner.lock();
482 let (micros, cursor) = self.next_cursor(&mut ring);
483 ring.pending.insert(cursor);
484 drop(ring);
485 Reservation {
486 inner: Arc::clone(&self.inner),
487 cursor,
488 micros,
489 fulfilled: false,
490 }
491 }
492
493 pub fn replay(&self, after: EventCursor, bounds: ReplayBounds) -> Replayed {
494 let ring = self.inner.lock();
495 // The corresponding read side guarantee of `reserve`.
496 let horizon = ring.pending.iter().next().copied();
497 let visible = |entry: &&Entry| {
498 entry.event.created > after
499 && horizon.is_none_or(|horizon| entry.event.created < horizon)
500 };
501 let events: Vec<Arc<Event>> = ring
502 .entries
503 .iter()
504 .filter(visible)
505 .take(bounds.events.get())
506 // Why the first event gets to ignore the byte bound?
507 // Imagine a consumer whose next event is by itself wider
508 // than the entire bound, right -
509 // every batch it requests would come back empty,
510 // its cursor would never advance,
511 // it would ask again, repeat.
512 // Sending that one event alone over the bound
513 // is the only way.
514 .scan(0usize, |spent, entry| {
515 let first = *spent == 0;
516 *spent += entry.bytes;
517 (first || *spent <= bounds.bytes.get()).then(|| Arc::clone(&entry.event))
518 })
519 .collect();
520 let end = match ring.entries.iter().filter(visible).nth(events.len()) {
521 Some(_) => BatchEnd::Bounded,
522 None => BatchEnd::CaughtUp,
523 };
524 Replayed { events, end }
525 }
526
527 pub fn subscribe(&self) -> watch::Receiver<EventCursor> {
528 self.inner.head.subscribe()
529 }
530}
531
532pub struct Reservation {
533 inner: Arc<Inner>,
534 cursor: EventCursor,
535 micros: UnixMicros,
536 fulfilled: bool,
537}
538
539impl Reservation {
540 pub fn cursor(&self) -> EventCursor {
541 self.cursor
542 }
543
544 pub fn fulfill<P: Publish>(mut self, payload: &P) {
545 let payload = serde_json::to_value(payload).expect("event payload serializes to JSON");
546 let mut ring = self.inner.lock();
547 insert_sorted(
548 &mut ring,
549 Event {
550 rkey: Tid::from_time(self.micros.get(), 0),
551 nsid: P::NSID,
552 payload,
553 created: self.cursor,
554 },
555 self.inner.bounds,
556 );
557 ring.pending.remove(&self.cursor);
558 self.inner.settle(ring);
559 self.fulfilled = true;
560 }
561}
562
563impl Drop for Reservation {
564 fn drop(&mut self) {
565 if self.fulfilled {
566 return;
567 }
568 let mut ring = self.inner.lock();
569 ring.pending.remove(&self.cursor);
570 self.inner.settle(ring);
571 }
572}
573
574knot_types::scalar_newtype! {
575 pub struct GlobalSubscriberLimit(usize);
576 pub struct PerPeerSubscriberLimit(usize);
577}
578
579pub struct SubscriberGate {
580 global: Arc<Semaphore>,
581 per_peer_max: usize,
582 peers: Mutex<HashMap<IpAddr, usize>>,
583}
584
585impl SubscriberGate {
586 pub fn new(global_max: GlobalSubscriberLimit, per_peer_max: PerPeerSubscriberLimit) -> Self {
587 Self {
588 global: Arc::new(Semaphore::new(global_max.get().max(1))),
589 per_peer_max: per_peer_max.get().max(1),
590 peers: Mutex::new(HashMap::new()),
591 }
592 }
593
594 pub fn try_admit(self: &Arc<Self>, peer: IpAddr) -> Option<SubscriberPermit> {
595 let global = Arc::clone(&self.global).try_acquire_owned().ok()?;
596 let mut peers = self
597 .peers
598 .lock()
599 .unwrap_or_else(|poisoned| poisoned.into_inner());
600 let current = peers.get(&peer).copied().unwrap_or(0);
601 if current >= self.per_peer_max {
602 return None;
603 }
604 peers.insert(peer, current + 1);
605 Some(SubscriberPermit {
606 _global: global,
607 gate: Arc::clone(self),
608 peer,
609 })
610 }
611}
612
613pub struct SubscriberPermit {
614 _global: OwnedSemaphorePermit,
615 gate: Arc<SubscriberGate>,
616 peer: IpAddr,
617}
618
619impl Drop for SubscriberPermit {
620 fn drop(&mut self) {
621 let mut peers = self
622 .gate
623 .peers
624 .lock()
625 .unwrap_or_else(|poisoned| poisoned.into_inner());
626 if let Some(count) = peers.get_mut(&self.peer) {
627 *count -= 1;
628 if *count == 0 {
629 peers.remove(&self.peer);
630 }
631 }
632 }
633}
634
635#[cfg(test)]
636mod tests {
637 use super::*;
638
639 use knot_runtime::{ManualClock, UnixMicros};
640
641 fn bounds(events: usize, bytes: usize) -> ReplayBounds {
642 ReplayBounds::new(
643 ReplayEvents::new(events).unwrap(),
644 ReplayBytes::new(bytes).unwrap(),
645 )
646 }
647
648 fn log(capacity: usize) -> EventLog<ManualClock> {
649 EventLog::new(
650 ManualClock::new(UnixMicros::new(1_700_000_000_000_000)),
651 bounds(capacity, 1 << 20),
652 )
653 }
654
655 fn replay(log: &EventLog<ManualClock>, after: EventCursor, limit: usize) -> Vec<Arc<Event>> {
656 log.replay(after, bounds(limit, 1 << 30)).events
657 }
658
659 fn update() -> GitRefUpdate {
660 GitRefUpdate::new(
661 RepoDid::new("did:plc:limpet").unwrap(),
662 Some(OwnerDid::new("did:web:olaren.dev").unwrap()),
663 AccountDid::new("did:plc:nel").unwrap(),
664 )
665 }
666
667 fn wire<P: Publish>(log: &EventLog<ManualClock>, payload: &P) -> serde_json::Value {
668 log.publish(payload);
669 serde_json::to_value(&*replay(log, EventCursor::START, 1).remove(0)).unwrap()
670 }
671
672 #[test]
673 fn publish_nsids_are_valid_type_names() {
674 [
675 GitRefUpdate::NSID,
676 KnotMemberUpdate::NSID,
677 RepoCollaboratorUpdate::NSID,
678 ]
679 .iter()
680 .for_each(|nsid| {
681 assert!(knot_types::TypeName::new(*nsid).is_ok(), "{nsid}");
682 });
683 }
684
685 #[test]
686 fn a_frozen_clock_still_yields_strictly_increasing_cursors_and_distinct_rkeys() {
687 let log = log(8);
688 let cursors: Vec<_> = (0..3).map(|_| log.publish(&update())).collect();
689 assert!(cursors.windows(2).all(|pair| pair[0] < pair[1]));
690 let events = replay(&log, EventCursor::START, 8);
691 let rkeys: std::collections::BTreeSet<_> = events
692 .iter()
693 .map(|event| event.rkey.as_str().to_string())
694 .collect();
695 assert_eq!(rkeys.len(), 3);
696 }
697
698 #[test]
699 fn the_ring_evicts_the_oldest_event_past_capacity() {
700 let log = log(2);
701 let first = log.publish(&update());
702 log.publish(&update());
703 log.publish(&update());
704 let replayed = replay(&log, EventCursor::START, 8);
705 assert_eq!(replayed.len(), 2);
706 assert!(replayed.iter().all(|event| event.created > first));
707 }
708
709 #[test]
710 fn a_wide_event_evicts_by_bytes_long_before_the_ring_fills() {
711 let wide = |count: usize| {
712 update().with_changed_files(fill_changed(
713 (0..count).map(|index| format!("crates/knot-events/src/f{index}.rs")),
714 ))
715 };
716 let log = EventLog::new(
717 ManualClock::new(UnixMicros::new(1_700_000_000_000_000)),
718 bounds(1_024, 64 * 1_024),
719 );
720 (0..16).for_each(|_| {
721 log.publish(&wide(512));
722 });
723 let replayed = replay(&log, EventCursor::START, 1_024);
724 assert!(
725 (1..16).contains(&replayed.len()),
726 "the byte maximum evicts before the event maximum does: {}",
727 replayed.len()
728 );
729
730 let one = EventLog::new(
731 ManualClock::new(UnixMicros::new(1_700_000_000_000_000)),
732 bounds(1_024, 1),
733 );
734 let only = one.publish(&wide(512));
735 assert_eq!(
736 replay(&one, EventCursor::START, 8)
737 .iter()
738 .map(|event| event.created)
739 .collect::<Vec<EventCursor>>(),
740 vec![only],
741 "the ring keeps the one event wider than the whole byte maximum"
742 );
743 }
744
745 fn fill_changed(paths: impl Iterator<Item = String>) -> ChangedFiles {
746 let mut budget = knot_types::ChangedFilesBudget::new();
747 let _ = paths.into_iter().try_for_each(|path| {
748 budget.admit(knot_types::RepoPath::new(path).expect("test path is well-formed"))
749 });
750 budget.finish()
751 }
752
753 #[test]
754 fn replay_honors_the_cursor_and_the_limit() {
755 let log = log(8);
756 let cursors: Vec<_> = (0..4).map(|_| log.publish(&update())).collect();
757 let after_second = replay(&log, cursors[1], 8);
758 assert_eq!(
759 after_second
760 .iter()
761 .map(|event| event.created)
762 .collect::<Vec<_>>(),
763 cursors[2..].to_vec()
764 );
765 assert_eq!(replay(&log, EventCursor::START, 2).len(), 2);
766 assert!(replay(&log, cursors[3], 8).is_empty());
767 }
768
769 #[test]
770 fn a_replay_batch_stops_at_the_byte_maximum_and_reports_whether_more_remains() {
771 let log = EventLog::new(
772 ManualClock::new(UnixMicros::new(1_700_000_000_000_000)),
773 bounds(1_024, 1 << 20),
774 );
775 assert_eq!(
776 log.replay(EventCursor::START, bounds(8, 1 << 20)).end,
777 BatchEnd::CaughtUp,
778 "an empty ring has nothing left to send"
779 );
780 let wide = update().with_changed_files(fill_changed(
781 (0..512).map(|index| format!("crates/knot-events/src/f{index}.rs")),
782 ));
783 let cursors: Vec<EventCursor> = (0..8).map(|_| log.publish(&wide)).collect();
784
785 let batch = log.replay(EventCursor::START, bounds(1_024, 16 * 1_024));
786 assert!(
787 (1..8).contains(&batch.events.len()),
788 "the byte maximum stops the batch before the event maximum does: {}",
789 batch.events.len()
790 );
791 assert_eq!(batch.end, BatchEnd::Bounded);
792
793 let rest = log.replay(
794 batch.events.last().expect("the batch is nonempty").created,
795 bounds(1_024, 1 << 30),
796 );
797 assert_eq!(rest.end, BatchEnd::CaughtUp);
798 assert_eq!(
799 batch.events.len() + rest.events.len(),
800 cursors.len(),
801 "the two batches together are every event, with none repeated or skipped"
802 );
803 let head = log.replay(cursors[7], bounds(8, 1 << 20));
804 assert!(head.events.is_empty() && head.end == BatchEnd::CaughtUp);
805
806 let single = log.replay(EventCursor::START, bounds(1_024, 1));
807 assert_eq!(
808 single.events.len(),
809 1,
810 "an event wider than the whole batch maximum is sent alone"
811 );
812 assert_eq!(single.end, BatchEnd::Bounded);
813 }
814
815 #[test]
816 fn a_subscriber_observes_the_head_advance() {
817 let log = log(8);
818 let mut head = log.subscribe();
819 assert_eq!(*head.borrow_and_update(), EventCursor::START);
820 let created = log.publish(&update());
821 assert!(head.has_changed().unwrap());
822 assert_eq!(*head.borrow_and_update(), created);
823 }
824
825 #[test]
826 fn the_wire_event_matches_the_eventstream_shape() {
827 let wire = wire(&log(8), &update());
828 assert_eq!(wire["nsid"], "sh.tangled.git.refUpdate");
829 assert_eq!(wire["created"].as_i64().unwrap() % 1_000, 0);
830 assert_eq!(wire["rkey"].as_str().unwrap().len(), 13);
831 let payload = &wire["event"];
832 assert_eq!(payload["$type"], "sh.tangled.git.refUpdate");
833 assert_eq!(payload["committerDid"], "did:plc:nel");
834 assert_eq!(payload["ownerDid"], "did:web:olaren.dev");
835 assert_eq!(payload["repo"], "did:plc:limpet");
836 assert_eq!(payload["meta"], serde_json::Value::Null);
837 assert_eq!(payload["ref"], "");
838 assert_eq!(
839 payload["oldSha"], "",
840 "a record about no ref sends the empty sha"
841 );
842 assert_eq!(payload["newSha"], "");
843 }
844
845 #[test]
846 fn an_absent_sha_of_a_transition_is_the_null_oid_of_the_repo_object_format() {
847 let new = Oid::from_hex(&"cd".repeat(32)).unwrap();
848 let created = &wire(
849 &log(8),
850 &GitRefUpdate::new(
851 RepoDid::new("did:plc:limpet").unwrap(),
852 None,
853 AccountDid::new("did:plc:nel").unwrap(),
854 )
855 .on_ref(
856 RefName::new("refs/heads/fresh").unwrap(),
857 RefTransition::Create { new },
858 ObjectFormat::SHA256,
859 ),
860 )["event"];
861 assert_eq!(created["oldSha"], "0".repeat(64));
862 assert_eq!(created["newSha"], new.to_hex());
863
864 let old = Oid::from_hex(&"ab".repeat(20)).unwrap();
865 let rebuilt = update()
866 .on_ref(
867 RefName::new("refs/heads/fresh").unwrap(),
868 RefTransition::Create {
869 new: Oid::from_hex(&"cd".repeat(20)).unwrap(),
870 },
871 ObjectFormat::SHA1,
872 )
873 .on_ref(
874 RefName::new("refs/heads/gone").unwrap(),
875 RefTransition::Delete { old },
876 ObjectFormat::SHA1,
877 );
878 let payload = &wire(&log(8), &rebuilt)["event"];
879 assert_eq!(
880 payload["ref"], "refs/heads/gone",
881 "a later transition replaces the earlier one whole"
882 );
883 assert_eq!(payload["oldSha"], old.to_hex());
884 assert_eq!(payload["newSha"], "0".repeat(40));
885 }
886
887 fn peer(last: u8) -> IpAddr {
888 IpAddr::V4(std::net::Ipv4Addr::new(127, 0, 0, last))
889 }
890
891 #[test]
892 fn gate_enforces_limits_and_prunes() {
893 let global = Arc::new(SubscriberGate::new(
894 GlobalSubscriberLimit::new(2),
895 PerPeerSubscriberLimit::new(8),
896 ));
897 let first = global
898 .try_admit(peer(1))
899 .expect("first subscriber is admitted");
900 let _second = global
901 .try_admit(peer(2))
902 .expect("second subscriber is admitted");
903 assert!(
904 global.try_admit(peer(3)).is_none(),
905 "third subscriber is refused once the global limit is reached"
906 );
907 drop(first);
908 assert!(
909 global.try_admit(peer(3)).is_some(),
910 "freeing global slot admits waiting subscriber"
911 );
912
913 let per_peer = Arc::new(SubscriberGate::new(
914 GlobalSubscriberLimit::new(16),
915 PerPeerSubscriberLimit::new(2),
916 ));
917 let _socket = per_peer
918 .try_admit(peer(1))
919 .expect("first socket is admitted");
920 let second = per_peer
921 .try_admit(peer(1))
922 .expect("second socket is admitted");
923 assert!(
924 per_peer.try_admit(peer(1)).is_none(),
925 "third socket from same peer is refused at the per-peer limit"
926 );
927 assert!(
928 per_peer.try_admit(peer(2)).is_some(),
929 "different peer keeps its own budget"
930 );
931 drop(second);
932 assert!(
933 per_peer.try_admit(peer(1)).is_some(),
934 "freed per-peer slot is reusable"
935 );
936
937 let prune = Arc::new(SubscriberGate::new(
938 GlobalSubscriberLimit::new(16),
939 PerPeerSubscriberLimit::new(2),
940 ));
941 let permit = prune.try_admit(peer(1)).expect("admitted");
942 drop(permit);
943 assert!(
944 prune
945 .peers
946 .lock()
947 .unwrap_or_else(|poisoned| poisoned.into_inner())
948 .is_empty(),
949 "peer map prunes peer once its last socket closes"
950 );
951 }
952
953 #[test]
954 fn a_ref_update_includes_its_computed_meta_on_the_wire() {
955 let wire = wire(
956 &log(8),
957 &update().with_meta(RefUpdateMeta::new(
958 true,
959 vec![EmailCommitCount::new(
960 Email::new("nel@oyster.cafe"),
961 CommitCount::new(3),
962 )],
963 vec![LanguageSize::new(
964 LanguageName::new("Rust"),
965 LanguageBytes::new(1234),
966 )],
967 )),
968 );
969 let meta = &wire["event"]["meta"];
970 assert_eq!(meta["isDefaultRef"], true);
971 assert_eq!(
972 meta["commitCount"]["byEmail"][0]["email"],
973 "nel@oyster.cafe"
974 );
975 assert_eq!(meta["commitCount"]["byEmail"][0]["count"], 3);
976 assert_eq!(meta["langBreakdown"]["inputs"][0]["lang"], "Rust");
977 assert_eq!(meta["langBreakdown"]["inputs"][0]["size"], 1234);
978 }
979
980 #[test]
981 fn an_empty_breakdown_omits_the_optional_meta_arrays() {
982 let wire = wire(
983 &log(8),
984 &update().with_meta(RefUpdateMeta::new(false, Vec::new(), Vec::new())),
985 );
986 let meta = &wire["event"]["meta"];
987 assert_eq!(meta["isDefaultRef"], false);
988 assert!(meta["commitCount"].get("byEmail").is_none());
989 assert!(meta.get("langBreakdown").is_none());
990 }
991
992 #[test]
993 fn acl_updates_match_eventstream_shape() {
994 let log = log(8);
995 log.publish(&KnotMemberUpdate::added(
996 AccountDid::new("did:plc:nel").unwrap(),
997 ));
998 log.publish(&KnotMemberUpdate::removed(
999 AccountDid::new("did:plc:olaren").unwrap(),
1000 ));
1001 log.publish(&RepoCollaboratorUpdate::added(
1002 AccountDid::new("did:plc:nel").unwrap(),
1003 RepoDid::new("did:plc:limpet").unwrap(),
1004 ));
1005 log.publish(&RepoCollaboratorUpdate::removed(
1006 AccountDid::new("did:plc:nel").unwrap(),
1007 RepoDid::new("did:plc:limpet").unwrap(),
1008 ));
1009 let events = replay(&log, EventCursor::START, 8);
1010
1011 let member_added = serde_json::to_value(&*events[0]).unwrap();
1012 assert_eq!(member_added["nsid"], "sh.tangled.knot.memberUpdate");
1013 assert_eq!(member_added["event"]["op"], "add");
1014 assert_eq!(member_added["event"]["subject"], "did:plc:nel");
1015 assert!(member_added["event"].get("$type").is_none());
1016 let member_removed = serde_json::to_value(&*events[1]).unwrap();
1017 assert_eq!(member_removed["event"]["op"], "remove");
1018 assert_eq!(member_removed["event"]["subject"], "did:plc:olaren");
1019
1020 let collab_added = serde_json::to_value(&*events[2]).unwrap();
1021 assert_eq!(collab_added["nsid"], "sh.tangled.repo.collaboratorUpdate");
1022 assert_eq!(collab_added["event"]["op"], "add");
1023 assert_eq!(collab_added["event"]["subject"], "did:plc:nel");
1024 assert_eq!(collab_added["event"]["repo"], "did:plc:limpet");
1025 assert!(collab_added["event"].get("$type").is_none());
1026 let collab_removed = serde_json::to_value(&*events[3]).unwrap();
1027 assert_eq!(collab_removed["event"]["op"], "remove");
1028 assert_eq!(collab_removed["event"]["repo"], "did:plc:limpet");
1029 }
1030
1031 fn cursors(log: &EventLog<ManualClock>) -> Vec<EventCursor> {
1032 replay(log, EventCursor::START, 64)
1033 .iter()
1034 .map(|event| event.created)
1035 .collect()
1036 }
1037
1038 #[test]
1039 fn a_reservation_holds_back_later_events_until_it_is_fulfilled() {
1040 let log = log(8);
1041 let early = log.publish(&update());
1042 let reservation = log.reserve();
1043 let later = log.publish(&update());
1044 assert!(reservation.cursor() > early && reservation.cursor() < later);
1045 assert_eq!(
1046 cursors(&log),
1047 vec![early],
1048 "event published after reservation waits behind it"
1049 );
1050 let mid = reservation.cursor();
1051 reservation.fulfill(&update());
1052 assert_eq!(
1053 cursors(&log),
1054 vec![early, mid, later],
1055 "fulfilling reservation releases it and event queued behind it, in cursor order"
1056 );
1057 }
1058
1059 #[test]
1060 fn out_of_order_fulfillment_still_replays_in_cursor_order() {
1061 let log = log(8);
1062 let first = log.reserve();
1063 let second = log.reserve();
1064 let (c1, c2) = (first.cursor(), second.cursor());
1065 assert!(c1 < c2);
1066 second.fulfill(&update());
1067 assert!(
1068 cursors(&log).is_empty(),
1069 "later reservation stays hidden while earlier one is outstanding"
1070 );
1071 first.fulfill(&update());
1072 assert_eq!(
1073 cursors(&log),
1074 vec![c1, c2],
1075 "both surface in cursor order regardless of fulfillment order"
1076 );
1077 }
1078
1079 #[test]
1080 fn a_dropped_reservation_unblocks_the_horizon_without_an_event() {
1081 let log = log(8);
1082 let reservation = log.reserve();
1083 let later = log.publish(&update());
1084 assert!(
1085 cursors(&log).is_empty(),
1086 "later event waits behind unfulfilled reservation"
1087 );
1088 drop(reservation);
1089 assert_eq!(
1090 cursors(&log),
1091 vec![later],
1092 "dropping reservation surfaces queued event and leaves no gap"
1093 );
1094 }
1095
1096 #[test]
1097 fn the_head_holds_at_the_last_stable_event_until_a_reservation_is_fulfilled() {
1098 let log = log(8);
1099 let mut head = log.subscribe();
1100 let early = log.publish(&update());
1101 assert_eq!(*head.borrow_and_update(), early);
1102 let reservation = log.reserve();
1103 let later = log.publish(&update());
1104 assert_eq!(
1105 *head.borrow_and_update(),
1106 early,
1107 "head holds while lower-cursor reservation is pending"
1108 );
1109 reservation.fulfill(&update());
1110 assert_eq!(
1111 *head.borrow_and_update(),
1112 later,
1113 "fulfilling reservation advances head past released events"
1114 );
1115 }
1116
1117 #[test]
1118 fn an_anonymous_owner_is_omitted_from_the_wire() {
1119 let wire = wire(
1120 &log(8),
1121 &GitRefUpdate::new(
1122 RepoDid::new("did:plc:limpet").unwrap(),
1123 None,
1124 AccountDid::new("did:plc:nel").unwrap(),
1125 ),
1126 );
1127 assert!(wire["event"].get("ownerDid").is_none());
1128 }
1129}