Something went wrong. Try again.
Monorepo for Tangled tangled.org
Something went wrong. Try again.
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541use std::collections::{HashMap, VecDeque};use std::str::FromStr;use std::sync::{Arc, Mutex, MutexGuard};use std::time::Duration;
use jacquard_common::DefaultStr;use jacquard_common::types::cid::IpldCid;use jacquard_common::types::string::{AtUri, Cid};use serde::Deserialize;use smallvec::SmallVec;use tokio::sync::oneshot;use tokio::time::Instant;
// a settlement only has to outlive the round trip from the writer's pds back to usconst RETENTION: Duration = Duration::from_secs(30);// best effort, counted in distinct at-uris so a writer rewriting one record// cant push everyone else outconst CAPACITY: usize = 4096;// enough revisions to still answer a writer whose record was superseded mid flightconst MAX_REVISIONS: usize = 8;// the detail can echo caller-controlled bytes (parse errors)const DETAIL_LIMIT: usize = 200;
/// why a record bobbin saw will not show up in its index#[derive(Clone, Copy, Debug, Eq, PartialEq)]pub enum Rejection { UnknownCollection, Undecodable, MissingBody, NotIndexed,}
impl Rejection { pub const fn as_str(self) -> &'static str { match self { Self::UnknownCollection => "unknownCollection", Self::Undecodable => "undecodable", Self::MissingBody => "missingBody", Self::NotIndexed => "notIndexed", } }}
/// what bobbin did with a record once it came off the firehose#[derive(Clone, Debug, Eq, PartialEq)]pub enum RecordOutcome { Indexed, Deleted, Rejected { reason: Rejection, detail: Option<Box<str>>, },}
impl RecordOutcome { pub fn rejected(reason: Rejection, detail: Option<&str>) -> Self { Self::Rejected { reason, detail: detail.map(clamp), } }
pub const fn as_str(&self) -> &'static str { match self { Self::Indexed => "indexed", Self::Deleted => "deleted", Self::Rejected { .. } => "rejected", } }
pub const fn rejection(&self) -> Option<Rejection> { match self { Self::Rejected { reason, .. } => Some(*reason), Self::Indexed | Self::Deleted => None, } }
pub fn detail(&self) -> Option<&str> { match self { Self::Rejected { detail, .. } => detail.as_deref(), Self::Indexed | Self::Deleted => None, } }}
fn clamp(s: &str) -> Box<str> { let mut end = DETAIL_LIMIT.min(s.len()); while !s.is_char_boundary(end) { end -= 1; } s[..end].into()}
#[derive(Clone, Copy, Debug, Eq, Hash, PartialEq)]pub struct ParsedCid(IpldCid);
impl From<IpldCid> for ParsedCid { fn from(cid: IpldCid) -> Self { Self(cid) }}
impl FromStr for ParsedCid { type Err = <IpldCid as FromStr>::Err;
fn from_str(s: &str) -> Result<Self, Self::Err> { s.parse::<IpldCid>().map(Self) }}
impl serde::Serialize for ParsedCid { fn serialize<S: serde::Serializer>(&self, serializer: S) -> Result<S::Ok, S::Error> { serializer.collect_str(&self.0) }}
impl<'de> Deserialize<'de> for ParsedCid { fn deserialize<D>(deserializer: D) -> Result<Self, D::Error> where D: serde::Deserializer<'de>, { let raw = Cid::<DefaultStr>::deserialize(deserializer)?; raw.to_ipld().map(Self).map_err(serde::de::Error::custom) }}
#[derive(Clone, Debug, Eq, PartialEq)]pub struct Settlement { pub source: AtUri<DefaultStr>, pub cid: Option<ParsedCid>, pub outcome: RecordOutcome,}
#[derive(Debug, Default)]pub struct Settlements { inner: Mutex<Inner>,}
#[derive(Debug, Default)]struct Inner { // per-revision history so a superseded record can still be resolved recent: HashMap<AtUri<DefaultStr>, SmallVec<[Recent; 1]>>, // first-seen order, exactly one entry per recent key order: VecDeque<AtUri<DefaultStr>>, waiters: HashMap<AtUri<DefaultStr>, Vec<Waiter>>, next_seq: u64,}
#[derive(Debug)]struct Recent { at: Instant, cid: Option<ParsedCid>, outcome: RecordOutcome,}
#[derive(Debug)]struct Waiter { seq: u64, cid: Option<ParsedCid>, tx: oneshot::Sender<Settlement>,}
pub enum Watch { Settled(Settlement), /// the registration's Drop cancels the wait, so timeouts and /// abandoned requests don't leave a waiter behind Waiting(Registration, oneshot::Receiver<Settlement>),}
/// one entry in the waiter list. dropping it removes the entry, so a/// settlement that arrives later is only remembered, not sentpub struct Registration { watch: Arc<Settlements>, source: AtUri<DefaultStr>, seq: u64,}
impl Drop for Registration { fn drop(&mut self) { let mut inner = self.watch.lock(); let Some(waiting) = inner.waiters.get_mut(&self.source) else { return; }; waiting.retain(|w| w.seq != self.seq); if waiting.is_empty() { inner.waiters.remove(&self.source); } }}
impl Settlements { pub fn new() -> Self { Self::default() }
fn lock(&self) -> MutexGuard<'_, Inner> { self.inner.lock().unwrap_or_else(|e| e.into_inner()) }
pub fn settle(&self, settlement: Settlement) { let now = Instant::now(); let winners = { let mut inner = self.lock(); let winners = inner.take_matching(&settlement); inner.remember(now, settlement.clone()); inner.prune(now); winners }; for tx in winners { let _ = tx.send(settlement.clone()); } }
pub fn watch(self: &Arc<Self>, source: &AtUri<DefaultStr>, cid: Option<ParsedCid>) -> Watch { let now = Instant::now(); let mut inner = self.lock(); inner.prune(now); if let Some(recent) = inner.find(source, cid) { return Watch::Settled(Settlement { source: source.clone(), cid: recent.cid, outcome: recent.outcome.clone(), }); } let seq = inner.next_seq(); let (tx, rx) = oneshot::channel(); inner .waiters .entry(source.clone()) .or_default() .push(Waiter { seq, cid, tx }); let registration = Registration { watch: self.clone(), source: source.clone(), seq, }; Watch::Waiting(registration, rx) }}
impl Inner { fn next_seq(&mut self) -> u64 { self.next_seq += 1; self.next_seq }
fn take_matching(&mut self, s: &Settlement) -> Vec<oneshot::Sender<Settlement>> { if let Some(waiting) = self.waiters.remove(&s.source) { let (hit, miss): (Vec<_>, Vec<_>) = waiting .into_iter() .partition(|w| w.cid.is_none_or(|want| s.cid == Some(want))); if !miss.is_empty() { self.waiters.insert(s.source.clone(), miss); } hit.into_iter().map(|w| w.tx).collect() } else { Vec::new() } }
fn find(&self, source: &AtUri<DefaultStr>, cid: Option<ParsedCid>) -> Option<&Recent> { self.recent .get(source)? .iter() .rev() .find(|r| cid.is_none_or(|want| r.cid == Some(want))) }
fn remember(&mut self, now: Instant, settlement: Settlement) { let entries = match self.recent.entry(settlement.source) { std::collections::hash_map::Entry::Vacant(v) => { self.order.push_back(v.key().clone()); v.insert(SmallVec::new()) } std::collections::hash_map::Entry::Occupied(o) => o.into_mut(), }; entries.push(Recent { at: now, cid: settlement.cid, outcome: settlement.outcome, }); entries.retain(|r| now.duration_since(r.at) < RETENTION); while entries.len() > MAX_REVISIONS { entries.remove(0); } }
fn prune(&mut self, now: Instant) { while let Some(source) = self.order.pop_front() { let over = self.recent.len() > CAPACITY; let keep = match self.recent.get_mut(&source) { None => false, Some(entries) => { entries.retain(|r| now.duration_since(r.at) < RETENTION); !over && !entries.is_empty() } }; if keep { self.order.push_front(source); break; } self.recent.remove(&source); } }}
#[cfg(test)]mod tests { use super::*;
fn uri(rkey: &str) -> AtUri<DefaultStr> { AtUri::new_owned(format!("at://did:plc:alice/org.tangled.feed.star/{rkey}")).unwrap() }
const CID_A: &str = "bafyreieqygohnz2zqyvtvktbjpvhutphobcmbsnt4q5lc36ri7vpcmoz4i"; const CID_B: &str = "bafkreigh2akiscaildc7gnvtklbsfhdgwz72eolmpckbqr5ej26byp3uli";
fn cid(s: &str) -> ParsedCid { s.parse().unwrap() }
// a structurally valid cidv1 whose digest is a counter, so every mint differs fn mint(i: u32) -> ParsedCid { let mut raw = vec![0x01u8, 0x55, 0x12, 0x20]; raw.extend_from_slice(&[0u8; 28]); raw.extend_from_slice(&i.to_be_bytes()); ParsedCid::from(IpldCid::try_from(raw.as_slice()).unwrap()) }
fn indexed(rkey: &str, c: Option<ParsedCid>) -> Settlement { Settlement { source: uri(rkey), cid: c, outcome: RecordOutcome::Indexed, } }
#[test] fn a_long_detail_is_clamped_on_a_char_boundary() { let long = "\u{1f600}".repeat(200); let outcome = RecordOutcome::rejected(Rejection::Undecodable, Some(&long)); let detail = outcome.detail().expect("detail kept"); assert!(detail.len() <= DETAIL_LIMIT); assert!(long.starts_with(detail)); }
fn settled(watch: Watch) -> Option<Settlement> { match watch { Watch::Settled(s) => Some(s), Watch::Waiting(..) => None, } }
#[tokio::test(start_paused = true)] async fn settlement_before_the_ask_is_still_there() { let watch = Arc::new(Settlements::new()); watch.settle(indexed("a", Some(cid(CID_A)))); let got = settled(watch.watch(&uri("a"), Some(cid(CID_A)))).expect("already settled"); assert_eq!(got.outcome, RecordOutcome::Indexed); }
#[tokio::test(start_paused = true)] async fn a_waiter_wakes_on_its_own_cid() { let watch = Arc::new(Settlements::new()); let Watch::Waiting(_reg, rx) = watch.watch(&uri("a"), Some(cid(CID_B))) else { panic!("nothing has settled yet"); }; watch.settle(indexed("a", Some(cid(CID_A)))); watch.settle(Settlement { source: uri("a"), cid: Some(cid(CID_B)), outcome: RecordOutcome::rejected(Rejection::Undecodable, None), }); let got = rx.await.expect("waiter woken"); assert_eq!( got.outcome, RecordOutcome::rejected(Rejection::Undecodable, None) ); }
#[tokio::test(start_paused = true)] async fn a_cidless_waiter_takes_the_next_settlement() { let watch = Arc::new(Settlements::new()); let Watch::Waiting(_reg, rx) = watch.watch(&uri("a"), None) else { panic!("nothing has settled yet"); }; watch.settle(Settlement { source: uri("a"), cid: None, outcome: RecordOutcome::Deleted, }); assert_eq!(rx.await.unwrap().outcome, RecordOutcome::Deleted); }
// a second write to the same uri must not eat the answer the first // writer is still waiting for #[tokio::test(start_paused = true)] async fn a_superseded_revision_keeps_its_own_outcome() { let watch = Arc::new(Settlements::new()); watch.settle(Settlement { source: uri("a"), cid: Some(cid(CID_A)), outcome: RecordOutcome::rejected(Rejection::Undecodable, None), }); watch.settle(indexed("a", Some(cid(CID_B)))); let old = settled(watch.watch(&uri("a"), Some(cid(CID_A)))).expect("older revision"); assert_eq!( old.outcome, RecordOutcome::rejected(Rejection::Undecodable, None) ); let new = settled(watch.watch(&uri("a"), Some(cid(CID_B)))).expect("newer revision"); assert_eq!(new.outcome, RecordOutcome::Indexed); let any = settled(watch.watch(&uri("a"), None)).expect("newest wins without a cid"); assert_eq!(any.cid, Some(cid(CID_B))); }
#[tokio::test(start_paused = true)] async fn an_unrelated_revision_does_not_wake_a_waiter() { let watch = Arc::new(Settlements::new()); let Watch::Waiting(_reg, mut rx) = watch.watch(&uri("a"), Some(cid(CID_B))) else { panic!("nothing has settled yet"); }; watch.settle(indexed("a", Some(cid(CID_A)))); assert!(rx.try_recv().is_err()); }
#[tokio::test(start_paused = true)] async fn dropping_the_registration_takes_the_waiter_back_out() { let watch = Arc::new(Settlements::new()); let Watch::Waiting(reg, rx) = watch.watch(&uri("a"), None) else { panic!("nothing has settled yet"); }; drop(reg); assert!(watch.lock().waiters.is_empty()); drop(rx); }
#[tokio::test(start_paused = true)] async fn settlements_expire() { let watch = Arc::new(Settlements::new()); watch.settle(indexed("a", None)); tokio::time::advance(RETENTION + Duration::from_secs(1)).await; watch.settle(indexed("b", None)); assert!(settled(watch.watch(&uri("a"), None)).is_none()); assert!(settled(watch.watch(&uri("b"), None)).is_some()); }
#[tokio::test(start_paused = true)] async fn a_later_revision_outlives_an_expired_earlier_one() { let watch = Arc::new(Settlements::new()); watch.settle(indexed("a", Some(cid(CID_A)))); tokio::time::advance(RETENTION - Duration::from_secs(1)).await; watch.settle(indexed("a", Some(cid(CID_B)))); tokio::time::advance(Duration::from_secs(2)).await; watch.settle(indexed("z", None)); assert!(settled(watch.watch(&uri("a"), Some(cid(CID_A)))).is_none()); let got = settled(watch.watch(&uri("a"), Some(cid(CID_B)))).expect("newer survives"); assert_eq!(got.outcome, RecordOutcome::Indexed); }
#[tokio::test(start_paused = true)] async fn the_recent_map_stays_bounded() { let watch = Arc::new(Settlements::new()); let keys = vec!["a", "b", "c", "d", "e"]; for key in keys { watch.settle(indexed(key, None)); } for i in 0..(MAX_REVISIONS * 3) { watch.settle(indexed("c", Some(mint(i as u32)))); } for i in 0..3 { watch.settle(indexed("a", Some(mint(i as u32)))); } let inner = watch.lock(); assert_eq!(inner.order.len(), inner.recent.len()); for entries in inner.recent.values() { assert!(entries.len() <= MAX_REVISIONS); } }
#[tokio::test(start_paused = true)] async fn a_rewrite_flood_does_not_evict_other_records() { let watch = Arc::new(Settlements::new()); watch.settle(indexed("b", None)); for i in 0..(CAPACITY + 100) { watch.settle(indexed("a", Some(mint(i as u32)))); } let watch_b = watch.watch(&uri("b"), None); assert!(matches!(watch_b, Watch::Settled(_))); let inner = watch.lock(); assert_eq!(inner.order.len(), inner.recent.len()); assert_eq!(inner.order.len(), 2); }
#[tokio::test(start_paused = true)] async fn only_the_last_few_revisions_of_a_record_survive() { let watch = Arc::new(Settlements::new()); let u = uri("a"); for i in 0..(MAX_REVISIONS + 5) { watch.settle(indexed("a", Some(mint(i as u32)))); } for i in 0..5 { assert!( settled(watch.watch(&u, Some(mint(i as u32)))).is_none(), "cid c{i} should be gone" ); } for i in 5..(MAX_REVISIONS + 5) { let s = settled(watch.watch(&u, Some(mint(i as u32)))).expect("should be findable"); assert_eq!(s.cid, Some(mint(i as u32))); } }
#[tokio::test(start_paused = true)] async fn too_many_records_evicts_the_oldest() { let watch = Arc::new(Settlements::new()); for i in 0..(CAPACITY + 10) { watch.settle(indexed(&format!("r{i}"), None)); } let inner = watch.lock(); assert!(inner.recent.len() <= CAPACITY); drop(inner); assert!( settled(watch.watch(&uri("r0"), None)).is_none(), "first should be gone" ); assert!( settled(watch.watch(&uri(&format!("r{}", CAPACITY + 9)), None)).is_some(), "last should survive" ); }
#[tokio::test(start_paused = true)] async fn expires_after_retention_with_no_pressure() { let watch = Arc::new(Settlements::new()); watch.settle(indexed("a", None)); tokio::time::advance(RETENTION + Duration::from_secs(1)).await; assert!(settled(watch.watch(&uri("a"), None)).is_none()); }}