Something went wrong. Try again.
Monorepo for Tangled tangled.org
Something went wrong. Try again.
1234567891011121314151617181920212223242526272829303132333435363738394041424344454647484950515253545556575859606162636465666768697071727374757677787980818283848586878889909192939495969798991001011021031041051061071081091101111121131141151161171181191201211221231241251261271281291301311321331341351361371381391401411421431441451461471481491501511521531541551561571581591601611621631641651661671681691701711721731741751761771781791801811821831841851861871881891901911921931941951961971981992002012022032042052062072082092102112122132142152162172182192202212222232242252262272282292302312322332342352362372382392402412422432442452462472482492502512522532542552562572582592602612622632642652662672682692702712722732742752762772782792802812822832842852862872882892902912922932942952962972982993003013023033043053063073083093103113123133143153163173183193203213223233243253263273283293303313323333343353363373383393403413423433443453463473483493503513523533543553563573583593603613623633643653663673683693703713723733743753763773783793803813823833843853863873883893903913923933943953963973983994004014024034044054064074084094104114124134144154164174184194204214224234244254264274284294304314324334344354364374384394404414424434444454464474484494504514524534544554564574584594604614624634644654664674684694704714724734744754764774784794804814824834844854864874884894904914924934944954964974984995005015025035045055065075085095105115125135145155165175185195205215225235245255265275285295305315325335345355365375385395405415425435445455465475485495505515525535545555565575585595605615625635645655665675685695705715725735745755765775785795805815825835845855865875885895905915925935945955965975985996006016026036046056066076086096106116126136146156166176186196206216226236246256266276286296306316326336346356366376386396406416426436446456466476486496506516526536546556566576586596606616626636646656666676686696706716726736746756766776786796806816826836846856866876886896906916926936946956966976986997007017027037047057067077087097107117127137147157167177187197207217227237247257267277287297307317327337347357367377387397407417427437447457467477487497507517527537547557567577587597607617627637647657667677687697707717727737747757767777787797807817827837847857867877887897907917927937947957967977987998008018028038048058068078088098108118128138148158168178188198208218228238248258268278288298308318328338348358368378388398408418428438448458468478488498508518528538548558568578588598608618628638648658668678688698708718728738748758768778788798808818828838848858868878888898908918928938948958968978988999009019029039049059069079089099109119129139149159169179189199209219229239249259269279289299309319329339349359369379389399409419429439449459469479489499509519529539549559569579589599609619629639649659669679689699709719729739749759769779789799809819829839849859869879889899909919929939949959969979989991000100110021003100410051006100710081009101010111012101310141015101610171018101910201021102210231024102510261027102810291030103110321033103410351036103710381039104010411042104310441045104610471048104910501051105210531054105510561057105810591060106110621063106410651066106710681069107010711072107310741075107610771078107910801081108210831084108510861087108810891090109110921093109410951096109710981099110011011102110311041105110611071108110911101111111211131114111511161117111811191120112111221123112411251126112711281129113011311132113311341135113611371138113911401141114211431144114511461147114811491150115111521153115411551156115711581159116011611162116311641165116611671168116911701171117211731174117511761177117811791180118111821183118411851186118711881189119011911192119311941195119611971198119912001201120212031204120512061207120812091210121112121213121412151216121712181219122012211222122312241225122612271228122912301231123212331234123512361237123812391240124112421243124412451246124712481249125012511252125312541255125612571258125912601261126212631264126512661267126812691270127112721273127412751276127712781279128012811282128312841285128612871288128912901291129212931294129512961297129812991300130113021303130413051306130713081309131013111312131313141315131613171318131913201321132213231324132513261327132813291330133113321333133413351336133713381339134013411342134313441345134613471348134913501351135213531354135513561357135813591360136113621363136413651366136713681369137013711372137313741375137613771378137913801381138213831384138513861387138813891390139113921393139413951396139713981399140014011402140314041405140614071408140914101411141214131414141514161417141814191420142114221423142414251426142714281429143014311432143314341435143614371438143914401441144214431444144514461447144814491450145114521453145414551456145714581459146014611462146314641465146614671468146914701471147214731474147514761477147814791480148114821483148414851486148714881489149014911492149314941495149614971498149915001501150215031504150515061507150815091510151115121513151415151516151715181519152015211522152315241525152615271528152915301531153215331534153515361537153815391540154115421543154415451546154715481549155015511552155315541555155615571558155915601561156215631564156515661567156815691570157115721573157415751576157715781579158015811582158315841585158615871588158915901591159215931594159515961597159815991600160116021603160416051606160716081609161016111612161316141615161616171618161916201621162216231624162516261627162816291630163116321633163416351636163716381639164016411642164316441645164616471648164916501651165216531654165516561657165816591660166116621663166416651666166716681669167016711672167316741675167616771678167916801681168216831684168516861687168816891690169116921693169416951696169716981699170017011702170317041705170617071708170917101711171217131714171517161717171817191720172117221723172417251726172717281729173017311732173317341735173617371738173917401741174217431744174517461747174817491750175117521753175417551756175717581759176017611762176317641765176617671768176917701771177217731774177517761777177817791780178117821783178417851786178717881789179017911792179317941795179617971798179918001801180218031804180518061807180818091810181118121813181418151816181718181819182018211822182318241825182618271828182918301831183218331834183518361837183818391840184118421843184418451846184718481849185018511852185318541855185618571858185918601861186218631864186518661867186818691870187118721873187418751876187718781879188018811882188318841885188618871888188918901891189218931894189518961897189818991900190119021903190419051906190719081909191019111912191319141915191619171918191919201921192219231924192519261927192819291930193119321933193419351936193719381939194019411942194319441945194619471948194919501951195219531954195519561957195819591960196119621963196419651966196719681969197019711972197319741975197619771978197919801981198219831984198519861987198819891990199119921993199419951996199719981999200020012002200320042005200620072008200920102011201220132014201520162017201820192020202120222023202420252026202720282029203020312032203320342035203620372038203920402041204220432044204520462047204820492050205120522053205420552056205720582059206020612062206320642065206620672068206920702071207220732074207520762077207820792080208120822083208420852086208720882089209020912092209320942095209620972098209921002101210221032104210521062107210821092110211121122113211421152116211721182119212021212122212321242125212621272128212921302131213221332134213521362137213821392140214121422143214421452146214721482149215021512152215321542155215621572158215921602161216221632164216521662167216821692170217121722173217421752176217721782179218021812182218321842185218621872188218921902191219221932194219521962197219821992200220122022203220422052206220722082209221022112212221322142215221622172218221922202221222222232224222522262227222822292230223122322233223422352236223722382239224022412242224322442245224622472248224922502251225222532254225522562257225822592260226122622263226422652266226722682269227022712272227322742275227622772278227922802281228222832284228522862287228822892290229122922293229422952296229722982299230023012302230323042305230623072308230923102311231223132314231523162317231823192320232123222323232423252326232723282329233023312332233323342335233623372338233923402341234223432344234523462347234823492350235123522353235423552356235723582359236023612362236323642365236623672368236923702371237223732374237523762377237823792380238123822383238423852386238723882389239023912392239323942395239623972398239924002401240224032404240524062407240824092410241124122413241424152416241724182419242024212422242324242425242624272428242924302431243224332434243524362437243824392440244124422443244424452446244724482449245024512452245324542455245624572458245924602461246224632464246524662467246824692470247124722473247424752476247724782479248024812482248324842485248624872488248924902491249224932494249524962497249824992500250125022503250425052506250725082509251025112512251325142515251625172518251925202521252225232524252525262527252825292530253125322533253425352536253725382539254025412542254325442545254625472548254925502551255225532554255525562557255825592560256125622563256425652566256725682569257025712572257325742575257625772578257925802581258225832584258525862587258825892590259125922593259425952596259725982599260026012602260326042605260626072608260926102611261226132614261526162617261826192620262126222623262426252626262726282629263026312632263326342635263626372638263926402641264226432644264526462647264826492650265126522653265426552656265726582659266026612662266326642665266626672668266926702671267226732674267526762677267826792680268126822683268426852686268726882689269026912692269326942695269626972698269927002701270227032704270527062707270827092710271127122713271427152716271727182719272027212722272327242725272627272728272927302731273227332734273527362737273827392740274127422743274427452746274727482749275027512752275327542755275627572758275927602761276227632764276527662767276827692770277127722773277427752776277727782779278027812782278327842785278627872788278927902791279227932794279527962797279827992800280128022803280428052806280728082809281028112812281328142815281628172818281928202821282228232824282528262827282828292830283128322833283428352836283728382839284028412842284328442845284628472848284928502851285228532854285528562857285828592860286128622863286428652866286728682869287028712872287328742875287628772878287928802881288228832884288528862887288828892890289128922893289428952896289728982899290029012902290329042905290629072908290929102911291229132914291529162917291829192920292129222923292429252926292729282929293029312932293329342935293629372938293929402941294229432944294529462947294829492950295129522953295429552956295729582959296029612962296329642965296629672968296929702971297229732974297529762977297829792980298129822983298429852986298729882989299029912992299329942995299629972998299930003001300230033004300530063007300830093010301130123013301430153016301730183019302030213022302330243025302630273028302930303031303230333034303530363037303830393040304130423043304430453046304730483049305030513052305330543055305630573058305930603061306230633064306530663067306830693070307130723073307430753076307730783079308030813082308330843085308630873088308930903091309230933094309530963097309830993100310131023103310431053106310731083109311031113112311331143115311631173118311931203121312231233124312531263127312831293130313131323133313431353136313731383139314031413142314331443145314631473148314931503151315231533154315531563157315831593160316131623163316431653166316731683169317031713172317331743175317631773178317931803181318231833184318531863187318831893190319131923193319431953196319731983199320032013202320332043205320632073208320932103211321232133214321532163217321832193220322132223223322432253226322732283229323032313232323332343235323632373238323932403241324232433244324532463247324832493250325132523253325432553256325732583259326032613262326332643265326632673268326932703271327232733274327532763277327832793280328132823283328432853286328732883289329032913292329332943295329632973298329933003301330233033304330533063307330833093310331133123313331433153316331733183319332033213322332333243325332633273328332933303331333233333334333533363337333833393340334133423343334433453346334733483349335033513352335333543355335633573358335933603361336233633364336533663367336833693370337133723373337433753376337733783379338033813382338333843385338633873388338933903391339233933394339533963397339833993400340134023403340434053406340734083409341034113412341334143415341634173418341934203421342234233424342534263427342834293430343134323433343434353436343734383439344034413442344334443445344634473448344934503451345234533454345534563457345834593460346134623463346434653466346734683469347034713472347334743475347634773478347934803481348234833484348534863487348834893490349134923493349434953496349734983499350035013502350335043505350635073508350935103511351235133514351535163517351835193520352135223523352435253526352735283529353035313532353335343535353635373538353935403541354235433544354535463547354835493550355135523553355435553556355735583559356035613562356335643565356635673568356935703571357235733574357535763577357835793580358135823583358435853586358735883589359035913592359335943595359635973598359936003601360236033604360536063607360836093610361136123613361436153616361736183619362036213622362336243625362636273628362936303631363236333634363536363637363836393640364136423643364436453646364736483649365036513652365336543655365636573658365936603661366236633664366536663667366836693670367136723673367436753676367736783679368036813682368336843685368636873688368936903691369236933694369536963697369836993700370137023703370437053706370737083709371037113712371337143715371637173718371937203721372237233724372537263727372837293730373137323733373437353736373737383739374037413742374337443745374637473748374937503751375237533754375537563757375837593760376137623763376437653766376737683769377037713772377337743775377637773778377937803781378237833784378537863787378837893790379137923793379437953796379737983799380038013802380338043805380638073808380938103811381238133814381538163817381838193820382138223823382438253826382738283829383038313832383338343835383638373838383938403841384238433844384538463847384838493850385138523853385438553856385738583859386038613862386338643865386638673868386938703871387238733874387538763877387838793880388138823883388438853886388738883889389038913892389338943895389638973898389939003901390239033904390539063907390839093910391139123913391439153916391739183919392039213922392339243925392639273928392939303931393239333934393539363937393839393940394139423943394439453946394739483949395039513952395339543955395639573958395939603961396239633964396539663967396839693970397139723973397439753976397739783979398039813982398339843985398639873988398939903991399239933994399539963997399839994000400140024003400440054006400740084009401040114012401340144015401640174018401940204021402240234024402540264027402840294030403140324033403440354036403740384039404040414042404340444045404640474048404940504051405240534054405540564057405840594060406140624063406440654066406740684069407040714072407340744075407640774078407940804081408240834084408540864087408840894090409140924093409440954096409740984099410041014102410341044105410641074108410941104111411241134114411541164117411841194120412141224123412441254126412741284129413041314132413341344135413641374138413941404141414241434144414541464147414841494150415141524153415441554156415741584159416041614162416341644165416641674168416941704171417241734174417541764177417841794180418141824183418441854186418741884189419041914192419341944195419641974198419942004201420242034204420542064207420842094210421142124213421442154216421742184219422042214222422342244225422642274228422942304231use std::collections::{HashSet, VecDeque};use std::num::{NonZeroU64, NonZeroUsize};use std::sync::Arc;use std::sync::atomic::{AtomicU64, Ordering};use std::time::Duration;
use bobbin_edge_index::{ ClaimIndex, Coverage, CoverageWatch, EdgeStore, HydrantCursor, ParsedCid, PromotionSignal, RecordOutcome, Rejection, RepoTransition, Settlement, Settlements, StreamHealth, delete_record_indexes, upsert_record_indexes,};use bobbin_record_lru::RecordStore;use bobbin_resolver::{ HydrantClient, IdentityResolver, NormalizeRepoRefs, decode_record, synthesize_created_at,};use bobbin_runtime::{ Clock, Entropy, NetworkError, RuntimeHasher, UnixMicros, WsConn, WsMessage, WsStream, WsTransport,};use bobbin_types::edges::{Edge, ExtractError, Record};use bobbin_types::ids::{RepoIdent, SubjectRef, owner_did_from_aturi};use bobbin_types::org_tangled::repo::manifest::Manifest;use bobbin_types::record::RecordBody;use bobbin_types::search::{SearchSink, SearchableRecord};use bytes::Bytes;use futures::StreamExt;use jacquard_common::DefaultStr;use jacquard_common::types::did::Did;use jacquard_common::types::ident::AtIdentifier;use jacquard_common::types::nsid::Nsid;use jacquard_common::types::recordkey::Rkey;use jacquard_common::types::string::{AtStrError, AtUri, Cid};use jacquard_common::types::tid::Tid;use thiserror::Error;use tokio::time::Instant;use tokio_stream::wrappers::ReceiverStream;use tokio_util::sync::CancellationToken;use tracing::{debug, info, warn};use url::Url;
mod analytics;mod frame;mod metrics;pub mod moderation;mod resolver;mod shadow;mod warming;use frame::HydrantStreamErrorFrame;pub use frame::{AccountFrame, FrameKind, HydrantFrame, IdentityFrame, RecordAction, RecordFrame};use metrics::{EdgeDrop, FrameDrop};pub use resolver::{RepoIdResolver, Resolution};pub use shadow::{WarmingShadowBuffer, WarmingShadowSnapshot};pub use warming::{ParkedUpsert, WarmingBuffer, WarmingBufferSnapshot};
const TANGLED_PREFIX: &str = "sh.tangled.";const ORG_TANGLED_PREFIX: &str = "org.tangled.";const RECONNECT_INITIAL_DELAY: Duration = Duration::from_millis(500);const RECONNECT_MAX_DELAY: Duration = Duration::from_secs(30);const PING_INTERVAL: Duration = Duration::from_secs(20);const PONG_TIMEOUT: Duration = Duration::from_secs(15);const CONNECT_TIMEOUT: Duration = Duration::from_secs(15);const HEAD_TIMEOUT: Duration = Duration::from_secs(5);const READY_SKEW: Duration = Duration::from_secs(60);const FRAME_CHANNEL_DEPTH: usize = 256;const CONTROL_CHANNEL_DEPTH: usize = 16;const READER_HOLD_LIMIT: usize = 64;const SEND_TIMEOUT: Duration = Duration::from_secs(10);const NORMAL_CLOSE: u16 = 1000;const WARMING_FLUSH_PARALLELISM: usize = 64;pub const DEFAULT_INGEST_PARALLELISM: NonZeroUsize = match NonZeroUsize::new(16) { Some(n) => n, None => unreachable!(),};pub const DEFAULT_IDLE_PROMOTE_MIN_EVENTS: NonZeroU64 = match NonZeroU64::new(256) { Some(n) => n, None => unreachable!(),};
pub use analytics::{Analytics, PosthogConfig};
#[derive(Clone, Debug)]pub struct IngestConfig { pub hydrant_base: Url, pub start_cursor: HydrantCursor, pub parallelism: NonZeroUsize, pub idle_promote_min_events: NonZeroU64,}
impl IngestConfig { pub fn new(hydrant_base: Url) -> Self { Self { hydrant_base, start_cursor: HydrantCursor::new(0), parallelism: DEFAULT_INGEST_PARALLELISM, idle_promote_min_events: DEFAULT_IDLE_PROMOTE_MIN_EVENTS, } }
fn stream_url(&self, cursor: HydrantCursor) -> Result<Url, IngestError> { let mut url = websocket_url(&self.hydrant_base)?; url.set_path("/stream"); url.query_pairs_mut() .clear() .append_pair("cursor", &cursor.raw().to_string()); Ok(url) }}
fn websocket_url(base: &Url) -> Result<Url, IngestError> { let mut url = base.clone(); match url.scheme() { "http" => url .set_scheme("ws") .map_err(|_| IngestError::Url("set ws scheme"))?, "https" => url .set_scheme("wss") .map_err(|_| IngestError::Url("set wss scheme"))?, "ws" | "wss" => {} other => return Err(IngestError::UnknownScheme(other.to_owned())), } Ok(url)}
#[derive(Debug, Error)]pub enum IngestError { #[error("invalid hydrant url: {0}")] Url(&'static str), #[error("unsupported url scheme: {0}")] UnknownScheme(String), #[error("network: {0}")] Network(#[from] NetworkError), #[error("frame decode: {0}")] Decode(#[from] serde_json::Error), #[error("invalid at-uri synthesized from frame: {0}")] InvalidAtUri(#[from] AtStrError), #[error("record extraction: {0}")] Extract(#[from] ExtractError), #[error("hydrant did not respond to ping within {0:?}")] PongTimeout(Duration), #[error("hydrant did not finish the websocket handshake within {0:?}")] ConnectTimeout(Duration), #[error("websocket send blocked for at least {0:?}, treating link as dead")] SendTimeout(Duration), #[error("hydrant disconnected because bobbin's stream consumer fell behind: {message}")] ConsumerTooSlow { message: String }, #[error("hydrant signaled stream error {code}: {message}")] HydrantStream { code: String, message: String },}
#[derive(Clone, Copy, Debug, Eq, PartialEq, Hash)]pub enum DisconnectKind { Url, UnknownScheme, Network, Decode, InvalidAtUri, Extract, ConnectTimeout, PongTimeout, SendTimeout, ConsumerTooSlow, HydrantStream,}
impl DisconnectKind { pub const fn as_str(self) -> &'static str { match self { Self::Url => "url", Self::UnknownScheme => "unknown_scheme", Self::Network => "network", Self::Decode => "decode", Self::InvalidAtUri => "invalid_at_uri", Self::Extract => "extract", Self::ConnectTimeout => "connect_timeout", Self::PongTimeout => "pong_timeout", Self::SendTimeout => "send_timeout", Self::ConsumerTooSlow => "consumer_too_slow", Self::HydrantStream => "hydrant_stream", } }
pub fn from_error(err: &IngestError) -> Self { match err { IngestError::Url(_) => Self::Url, IngestError::UnknownScheme(_) => Self::UnknownScheme, IngestError::Network(_) => Self::Network, IngestError::Decode(_) => Self::Decode, IngestError::InvalidAtUri(_) => Self::InvalidAtUri, IngestError::Extract(_) => Self::Extract, IngestError::ConnectTimeout(_) => Self::ConnectTimeout, IngestError::PongTimeout(_) => Self::PongTimeout, IngestError::SendTimeout(_) => Self::SendTimeout, IngestError::ConsumerTooSlow { .. } => Self::ConsumerTooSlow, IngestError::HydrantStream { .. } => Self::HydrantStream, } }}
#[derive(Clone, Debug, Eq, PartialEq)]pub struct DisconnectSnapshot { pub kind: DisconnectKind, pub message: String, pub at_unix_micros: UnixMicros, pub last_cursor: HydrantCursor,}
#[derive(Default)]pub struct DisconnectSink { last: std::sync::Mutex<Option<DisconnectSnapshot>>, count: std::sync::atomic::AtomicU64,}
impl DisconnectSink { pub fn new() -> Self { Self::default() }
pub fn record(&self, snap: DisconnectSnapshot) { *self.last.lock().expect("disconnect sink mutex poisoned") = Some(snap); self.count .fetch_add(1, std::sync::atomic::Ordering::Relaxed); }
pub fn snapshot(&self) -> Option<DisconnectSnapshot> { self.last .lock() .expect("disconnect sink mutex poisoned") .clone() }
pub fn count(&self) -> u64 { self.count.load(std::sync::atomic::Ordering::Relaxed) }}
#[derive(Clone, Copy, Debug, Eq, PartialEq)]enum SessionOutcome { Progressed, Empty,}
#[derive(Debug)]struct SessionEnd { outcome: SessionOutcome, error: Option<IngestError>,}
pub struct IngestRuntime<S: SearchSink + 'static> { pub store: Arc<EdgeStore>, pub claims: Arc<ClaimIndex>, pub coverage: Arc<CoverageWatch>, pub search: Arc<S>, pub records: Arc<dyn RecordStore>, pub resolver: Arc<RepoIdResolver>, pub identity: Arc<IdentityResolver>, pub clock: Arc<dyn Clock>, pub entropy: Arc<dyn Entropy>, pub stream_health: Arc<StreamHealth>, pub ws: Arc<dyn WsTransport>, pub hydrant: Option<HydrantClient>, pub cancel: CancellationToken, pub disconnects: Option<Arc<DisconnectSink>>, pub warming_shadow: Option<Arc<WarmingShadowBuffer>>, pub warming_buffer: Option<Arc<WarmingBuffer>>, pub settlements: Option<Arc<Settlements>>, pub analytics: Option<Arc<Analytics>>,}
impl<S: SearchSink + 'static> Clone for IngestRuntime<S> { fn clone(&self) -> Self { Self { store: self.store.clone(), claims: self.claims.clone(), coverage: self.coverage.clone(), search: self.search.clone(), records: self.records.clone(), resolver: self.resolver.clone(), identity: self.identity.clone(), clock: self.clock.clone(), entropy: self.entropy.clone(), stream_health: self.stream_health.clone(), ws: self.ws.clone(), hydrant: self.hydrant.clone(), cancel: self.cancel.clone(), disconnects: self.disconnects.clone(), warming_shadow: self.warming_shadow.clone(), warming_buffer: self.warming_buffer.clone(), settlements: self.settlements.clone(), analytics: self.analytics.clone(), } }}
impl<S: SearchSink + 'static> IngestRuntime<S> { fn pipeline_ctx(&self) -> PipelineCtx<'_, S> { PipelineCtx { resolver: &self.resolver, identity: &self.identity, store: &self.store, claims: &self.claims, coverage: &self.coverage, records: &*self.records, search: &self.search, shadow: self.warming_shadow.as_deref(), buffer: self.warming_buffer.as_deref(), settlements: self.settlements.as_deref(), analytics: self.analytics.as_deref(), } }}
struct PipelineCtx<'a, S: SearchSink + 'static> { resolver: &'a RepoIdResolver, identity: &'a IdentityResolver, store: &'a EdgeStore, claims: &'a ClaimIndex, coverage: &'a CoverageWatch, records: &'a dyn RecordStore, search: &'a S, shadow: Option<&'a WarmingShadowBuffer>, buffer: Option<&'a WarmingBuffer>, settlements: Option<&'a Settlements>, analytics: Option<&'a Analytics>,}
pub async fn run<S: SearchSink + 'static>( config: IngestConfig, runtime: IngestRuntime<S>,) -> Result<(), IngestError> { let frames = FrameCounter::default(); let warming_flusher = spawn_warming_flusher(&runtime); let result = run_inner(config, &runtime, &frames).await; if let Some(handle) = warming_flusher && let Err(join) = handle.await { warn!(?join, "warming flusher task panicked"); } result}
async fn run_inner<S: SearchSink + 'static>( config: IngestConfig, runtime: &IngestRuntime<S>, frames: &FrameCounter,) -> Result<(), IngestError> { let mut backoff = RECONNECT_INITIAL_DELAY; let mut replay_head = None; loop { let cursor = next_connect_cursor(runtime.coverage.snapshot(), config.start_cursor); let opened = runtime.clock.now_instant(); runtime .stream_health .connecting(runtime.clock.now_unix_micros()); if replay_head.is_none() && !runtime.coverage.snapshot().is_ready() { replay_head = fetch_replay_head(runtime).await; } let SessionEnd { outcome, error } = run_session(&config, cursor, replay_head, runtime, frames).await; let open_for = runtime.clock.now_instant().duration_since(opened); if runtime.cancel.is_cancelled() { info!( last_cursor = runtime.coverage.snapshot().last_cursor().raw(), "ingest stopped after shutdown signal" ); return Ok(()); } match (outcome, &error) { (SessionOutcome::Progressed, None) => info!( open_secs = open_for.as_secs(), "hydrant stream closed after delivering frames, reconnecting" ), (SessionOutcome::Empty, None) => warn!( open_secs = open_for.as_secs(), "hydrant stream closed without delivering frames" ), (_, Some(err)) => warn!( ?err, open_secs = open_for.as_secs(), "hydrant stream errored" ), } if let Some(err) = error.as_ref() { let kind = DisconnectKind::from_error(err); metrics::disconnected(kind); if let Some(sink) = runtime.disconnects.as_ref() { sink.record(DisconnectSnapshot { kind, message: err.to_string(), at_unix_micros: runtime.clock.now_unix_micros(), last_cursor: runtime.coverage.snapshot().last_cursor(), }); } } runtime.stream_health.disconnected( runtime.clock.now_unix_micros(), error.as_ref().map(|err| err.to_string()).as_deref(), ); let made_progress = matches!(outcome, SessionOutcome::Progressed); if made_progress { backoff = RECONNECT_INITIAL_DELAY; } else { tokio::select! { biased; _ = runtime.cancel.cancelled() => return Ok(()), _ = runtime.clock.sleep(jittered(backoff, &*runtime.entropy)) => {} } backoff = (backoff * 2).min(RECONNECT_MAX_DELAY); } }}
fn spawn_warming_flusher<S: SearchSink + 'static>( runtime: &IngestRuntime<S>,) -> Option<tokio::task::JoinHandle<()>> { runtime.warming_buffer.as_ref()?; let rt = runtime.clone(); Some(tokio::spawn(async move { let buffer = rt .warming_buffer .as_deref() .expect("warming flusher only spawns when buffer is set"); let mut rx = rt.coverage.subscribe(); let reason = loop { if rx.borrow_and_update().is_ready() { break FlushReason::Ready; } tokio::select! { biased; _ = rt.cancel.cancelled() => break FlushReason::Cancelled, res = rx.changed() => match res { Ok(()) => continue, Err(_) => break FlushReason::CoverageDropped, }, } }; flush_warming_buffer(&rt, buffer, reason).await; }))}
#[derive(Clone, Copy, Debug, Eq, PartialEq)]enum FlushReason { Ready, Cancelled, CoverageDropped,}
impl FlushReason { fn as_str(self) -> &'static str { match self { FlushReason::Ready => "ready", FlushReason::Cancelled => "cancelled", FlushReason::CoverageDropped => "coverage_dropped", } }}
async fn flush_warming_buffer<S: SearchSink + 'static>( runtime: &IngestRuntime<S>, buffer: &WarmingBuffer, reason: FlushReason,) { let drained = buffer.drain_for_promote().await; if drained.is_empty() { return; } if reason != FlushReason::Ready { info!( target: "bobbin_ingest::warming", abandoned_entries = drained.len(), reason = reason.as_str(), "abandoning parked items on non-ready flush", ); return; } let hasher = buffer.hasher().clone(); let unique: HashSet<RepoIdent, RuntimeHasher> = drained .iter() .flat_map(|(_, deps)| deps.iter().cloned()) .fold(HashSet::with_hasher(hasher), |mut acc, dep| { acc.insert(dep); acc }); if !unique.is_empty() { let resolver = runtime.resolver.clone(); let _: Vec<()> = futures::stream::iter(unique) .map(|key| { let resolver = resolver.clone(); async move { let _ = resolver.resolve(&key.owner, &key.rkey).await; } }) .buffer_unordered(WARMING_FLUSH_PARALLELISM) .collect() .await; } let upserts: Vec<ParkedUpsert> = drained.into_iter().map(|(u, _)| u).collect(); let count = upserts.len(); let ctx = runtime.pipeline_ctx(); finalize_drained(&ctx, upserts).await; info!( target: "bobbin_ingest::warming", flushed_entries = count, reason = reason.as_str(), "drained warming buffer", );}
#[derive(Clone, Copy, Debug, Eq, PartialEq)]struct FramesDelivered(u64);
#[derive(Clone, Debug, Default)]struct FrameCounter(Arc<AtomicU64>);
impl FrameCounter { fn delivered(&self) -> FramesDelivered { FramesDelivered(self.0.load(Ordering::Acquire)) }
fn deliver(&self) { self.0.fetch_add(1, Ordering::Release); }}
fn promote_if_caught_up<S: SearchSink + 'static>( runtime: &IngestRuntime<S>, min_events: NonZeroU64, delivered: FramesDelivered,) { let snap = runtime.coverage.snapshot(); if snap.is_ready() || snap.events_processed() < min_events.get() { return; } runtime.coverage.update(|c| c.force_ready()); info!( target: "bobbin_ingest::coverage", events_processed = snap.events_processed(), frames_delivered = delivered.0, last_cursor = snap.last_cursor().raw(), "a live socket didn't deliver a frame across a keepalive interval, promoting coverage to ready", );}
/// the newest event hydrant holds before we subscribe, so a warming replay can/// promote the moment it gets there. `None` leaves promotion to the idle checkasync fn fetch_replay_head<S: SearchSink + 'static>( runtime: &IngestRuntime<S>,) -> Option<HydrantCursor> { let client = runtime.hydrant.as_ref()?; let head = tokio::select! { biased; _ = runtime.cancel.cancelled() => return None, res = client.stream_head() => res, _ = runtime.clock.sleep(HEAD_TIMEOUT) => { warn!(timeout = ?HEAD_TIMEOUT, "hydrant didn't report its stream head, coverage waits for an idle stream"); return None; } }; match head { Ok(Some(id)) => { info!( target: "bobbin_ingest::coverage", head = id, "coverage promotes once the replay reaches hydrant's stream head", ); Some(HydrantCursor::new(id)) } Ok(None) => { debug!(target: "bobbin_ingest::coverage", "hydrant holds no events yet, coverage waits for an idle stream"); None } Err(err) => { warn!( ?err, "couldn't read hydrant's stream head, coverage waits for an idle stream" ); None } }}
fn promote_at_head<S: SearchSink + 'static>(runtime: &IngestRuntime<S>, head: HydrantCursor) { let snap = runtime.coverage.snapshot(); if snap.is_ready() || snap.last_cursor() < head { return; } runtime.coverage.update(|c| c.force_ready()); info!( target: "bobbin_ingest::coverage", events_processed = snap.events_processed(), last_cursor = snap.last_cursor().raw(), head = head.raw(), "the replay reached hydrant's stream head, promoting coverage to ready", );}
fn next_connect_cursor(snapshot: Coverage, start: HydrantCursor) -> HydrantCursor { if snapshot.events_processed() == 0 { start } else { HydrantCursor::new(snapshot.last_cursor().raw().saturating_add(1)) }}
fn jittered(base: Duration, entropy: &dyn Entropy) -> Duration { let base_ms = u64::try_from(base.as_millis()).unwrap_or(u64::MAX); let cap_ms = (base_ms / 4).max(1); base + Duration::from_millis(entropy.next_u64() % cap_ms)}
async fn run_session<S: SearchSink + 'static>( config: &IngestConfig, cursor: HydrantCursor, replay_head: Option<HydrantCursor>, runtime: &IngestRuntime<S>, frames: &FrameCounter,) -> SessionEnd { let url = match config.stream_url(cursor) { Ok(u) => u, Err(e) => { return SessionEnd { outcome: SessionOutcome::Empty, error: Some(e), }; } }; info!(%url, "connecting to hydrant /stream"); let connect = tokio::select! { biased; _ = runtime.cancel.cancelled() => { return SessionEnd { outcome: SessionOutcome::Empty, error: None }; } res = runtime.ws.connect(url) => res.map_err(IngestError::Network), _ = runtime.clock.sleep(CONNECT_TIMEOUT) => { Err(IngestError::ConnectTimeout(CONNECT_TIMEOUT)) } }; let WsConn { sink: mut ws_sink, stream: ws_stream, } = match connect { Ok(c) => c, Err(e) => { return SessionEnd { outcome: SessionOutcome::Empty, error: Some(e), }; } }; runtime .stream_health .connected(runtime.clock.now_unix_micros());
let (frame_tx, frame_rx) = tokio::sync::mpsc::channel::<HydrantFrame>(FRAME_CHANNEL_DEPTH); let parallelism = config.parallelism.get(); let processor_runtime = runtime.clone(); let processor = tokio::spawn(async move { let cancel = processor_runtime.cancel.clone(); let prep_rt = processor_runtime.clone(); let claim_rt = processor_runtime.clone(); let resolve_rt = processor_runtime.clone(); let commit_rt = processor_runtime;
let pipeline = ReceiverStream::new(frame_rx) .map(move |frame| prep_stage(frame, prep_rt.clone())) .buffered(parallelism) // then not buffered, resolve reads what this stage writes .then(move |staged| claim_stage(staged, claim_rt.clone())) .map(move |staged| resolve_stage(staged, resolve_rt.clone(), replay_head)) .buffered(parallelism) .for_each(move |staged| { commit_stage(staged, commit_rt.clone(), parallelism, replay_head) });
tokio::select! { biased; _ = cancel.cancelled() => {}, _ = pipeline => {}, } });
let (control_tx, mut control_rx) = tokio::sync::mpsc::channel::<WsEvent>(CONTROL_CHANNEL_DEPTH); let session_cancel = runtime.cancel.child_token(); let reader_cancel = session_cancel.clone(); let reader = tokio::spawn(reader_loop( ws_stream, frame_tx, control_tx, frames.clone(), reader_cancel, ));
let mut next_ping = runtime.clock.now_instant() + PING_INTERVAL; let mut pong_deadline: Option<Instant> = None; let mut quiet: Option<FramesDelivered> = None;
let writer_error: Option<IngestError> = loop { tokio::select! { biased; _ = runtime.cancel.cancelled() => { let _ = timed_send( &mut ws_sink, WsMessage::Close { code: NORMAL_CLOSE, reason: "bobbin shutdown".to_owned() }, ).await; break None; } _ = runtime.clock.sleep_until(next_ping) => { next_ping = runtime.clock.now_instant() + PING_INTERVAL; if pong_deadline.is_none() { let delivered = frames.delivered(); if quiet.replace(delivered) == Some(delivered) { promote_if_caught_up(runtime, config.idle_promote_min_events, delivered); } if let Err(e) = timed_send(&mut ws_sink, WsMessage::Ping(Bytes::new())).await { break Some(e); } pong_deadline = Some(runtime.clock.now_instant() + PONG_TIMEOUT); } } _ = wait_until(pong_deadline, runtime.clock.as_ref()) => { break Some(IngestError::PongTimeout(PONG_TIMEOUT)); } evt = control_rx.recv() => { let Some(evt) = evt else { break None; }; match evt { WsEvent::IncomingPing(payload) => { if let Err(e) = timed_send(&mut ws_sink, WsMessage::Pong(payload)).await { break Some(e); } } WsEvent::IncomingPong => { pong_deadline = None; } } } } };
session_cancel.cancel(); drop(ws_sink); drop(control_rx);
let reader_end = reader.await.unwrap_or(SessionEnd { outcome: SessionOutcome::Empty, error: None, }); if let Err(join) = processor.await { warn!(?join, "frame processor task panicked"); } SessionEnd { outcome: reader_end.outcome, error: writer_error.or(reader_end.error), }}
async fn timed_send( sink: &mut Box<dyn bobbin_runtime::WsSink>, msg: WsMessage,) -> Result<(), IngestError> { match tokio::time::timeout(SEND_TIMEOUT, sink.send(msg)).await { Ok(Ok(())) => Ok(()), Ok(Err(e)) => Err(IngestError::Network(e)), Err(_elapsed) => Err(IngestError::SendTimeout(SEND_TIMEOUT)), }}
async fn reader_loop( mut ws_stream: Box<dyn WsStream>, frame_tx: tokio::sync::mpsc::Sender<HydrantFrame>, control_tx: tokio::sync::mpsc::Sender<WsEvent>, frames: FrameCounter, cancel: CancellationToken,) -> SessionEnd { let mut outcome = SessionOutcome::Empty; let mut held: VecDeque<HydrantFrame> = VecDeque::new();
let error: Option<IngestError> = loop { let step = if held.is_empty() { tokio::select! { biased; _ = cancel.cancelled() => ReaderStep::Cancelled, msg = ws_stream.next() => ReaderStep::WsMessage(msg), } } else if held.len() < READER_HOLD_LIMIT { tokio::select! { biased; _ = cancel.cancelled() => ReaderStep::Cancelled, permit_result = frame_tx.reserve() => match permit_result { Ok(permit) => { let frame = held .pop_front() .expect("non-empty held when reserve succeeds"); permit.send(frame); ReaderStep::Sent } Err(_) => ReaderStep::FrameSinkClosed, }, msg = ws_stream.next() => ReaderStep::WsMessage(msg), } } else { tokio::select! { biased; _ = cancel.cancelled() => ReaderStep::Cancelled, permit_result = frame_tx.reserve() => match permit_result { Ok(permit) => { let frame = held .pop_front() .expect("held at limit when reserve succeeds"); permit.send(frame); ReaderStep::Sent } Err(_) => ReaderStep::FrameSinkClosed, }, } };
match step { ReaderStep::Cancelled => break None, ReaderStep::FrameSinkClosed => break None, ReaderStep::Sent => { outcome = SessionOutcome::Progressed; } ReaderStep::WsMessage(msg) => { let Some(msg) = msg else { break None; }; let parsed = match msg { Ok(m) => m, Err(e) => break Some(IngestError::Network(e)), }; match parsed { WsMessage::Text(text) => { let frame = match classify_text_frame(&text) { Ok(f) => f, Err(e) => break Some(e), }; frames.deliver(); held.push_back(frame); } WsMessage::Binary(_) => { debug!("hydrant sent unexpected binary frame, ignoring"); } WsMessage::Ping(payload) => { if control_tx .send(WsEvent::IncomingPing(payload)) .await .is_err() { break None; } } WsMessage::Pong(_) => { if control_tx.send(WsEvent::IncomingPong).await.is_err() { break None; } } WsMessage::Close { code, reason } => { debug!(code, %reason, "hydrant closed stream"); break None; } } } } }; SessionEnd { outcome, error }}
enum ReaderStep { Cancelled, FrameSinkClosed, Sent, WsMessage(Option<Result<WsMessage, NetworkError>>),}
fn classify_text_frame(text: &str) -> Result<HydrantFrame, IngestError> { #[derive(serde::Deserialize)] struct PeekType<'a> { #[serde(rename = "type", borrow)] kind: Option<std::borrow::Cow<'a, str>>, } let is_error_frame = serde_json::from_str::<PeekType>(text) .ok() .and_then(|p| p.kind) .as_deref() == Some("error"); if is_error_frame { return match serde_json::from_str::<HydrantStreamErrorFrame>(text) { Ok(err_frame) => Err(classify_hydrant_error(err_frame)), Err(decode_err) => Err(IngestError::Decode(decode_err)), }; } serde_json::from_str::<HydrantFrame>(text).map_err(IngestError::Decode)}
fn classify_hydrant_error(frame: HydrantStreamErrorFrame) -> IngestError { let HydrantStreamErrorFrame { error, message } = frame; let message = message.unwrap_or_default(); match error.as_str() { "ConsumerTooSlow" => IngestError::ConsumerTooSlow { message }, _ => IngestError::HydrantStream { code: error, message, }, }}
#[derive(Debug)]enum WsEvent { IncomingPing(Bytes), IncomingPong,}
async fn wait_until(deadline: Option<Instant>, clock: &dyn Clock) { match deadline { Some(d) => clock.sleep_until(d).await, None => std::future::pending::<()>().await, }}
#[derive(Clone, Copy, Debug, Eq, PartialEq)]enum Regime { Replay, Live, NonRecord,}
impl Regime { fn as_str(self) -> &'static str { match self { Self::Replay => "replay", Self::Live => "live", Self::NonRecord => "non_record", } }}
struct Pending { cursor: HydrantCursor, signal: PromotionSignal, regime: Regime, op: PendingOp,}
impl Pending { // replays keep their stored live flag, so only frames past hydrant's head are new fn is_fresh(&self, coverage: Coverage, head: Option<HydrantCursor>) -> bool { self.regime == Regime::Live && head.map_or(coverage.is_ready() || self.signal.caught_up(), |head| { self.cursor > head }) }}
enum PendingOp { Noop, Identity(Option<IdentityFrame>), Account(Option<AccountFrame>), Rejected { source: AtUri<DefaultStr>, cid: Option<Cid<DefaultStr>>, outcome: RecordOutcome, }, Upsert(Box<UpsertPieces>), Parked { nsid: Nsid<DefaultStr>, }, Delete { source: AtUri<DefaultStr>, nsid: Nsid<DefaultStr>, rev: Tid, },}
struct Prepared { pending: Pending, prepare_start: Instant, prepare_end: Instant,}
struct Resolved { pending: Pending, prepare_start: Instant, prepare_end: Instant, resolve_start: Instant, resolve_end: Instant,}
fn pending_nsid(op: &PendingOp) -> Option<&Nsid<DefaultStr>> { match op { PendingOp::Upsert(pieces) => Some(&pieces.nsid), PendingOp::Delete { nsid, .. } => Some(nsid), PendingOp::Parked { nsid, .. } => Some(nsid), PendingOp::Noop | PendingOp::Identity(_) | PendingOp::Account(_) | PendingOp::Rejected { .. } => None, }}
fn pending_edge_count(op: &PendingOp) -> u64 { match op { PendingOp::Upsert(pieces) => pieces.edges.len() as u64, _ => 0, }}
async fn prep_stage<S: SearchSink + 'static>( frame: HydrantFrame, rt: IngestRuntime<S>,) -> Prepared { let now = rt.clock.now_unix_micros(); let prepare_start = rt.clock.now_instant(); let ctx = rt.pipeline_ctx(); let pending = prepare_frame(frame, &ctx, now).await; let prepare_end = rt.clock.now_instant(); Prepared { pending, prepare_start, prepare_end, }}
/// the one stage that writes shared state before commit, so it runs serially/// repo claims have to land in cursor order or the parallel resolve below reads a half applied renameasync fn claim_stage<S: SearchSink + 'static>(staged: Prepared, rt: IngestRuntime<S>) -> Prepared { let ctx = rt.pipeline_ctx(); let Prepared { pending, prepare_start, prepare_end, } = staged; Prepared { pending: claim_pending(pending, &ctx).await, prepare_start, prepare_end, }}
async fn claim_pending<S: SearchSink + 'static>( pending: Pending, ctx: &PipelineCtx<'_, S>,) -> Pending { match &pending.op { PendingOp::Upsert(pieces) => { evict_from_buffer(ctx.buffer, &pieces.source).await; // visibility is the ClaimIndex's call now, the resolver only serves legacy at-uri lookups if let Record::Repo(repo) = &pieces.parsed { claim_repo(&pieces.source, repo, ctx).await; } } PendingOp::Rejected { source, .. } => evict_from_buffer(ctx.buffer, source).await, PendingOp::Delete { source, nsid, .. } => { evict_from_buffer(ctx.buffer, source).await; if nsid.as_ref() == "org.tangled.repo.manifest" && let Some(repo_did) = owner_did_from_aturi(source) && let Some(ident) = ctx.resolver.lookup_by_repo_did(&repo_did).await { ctx.resolver.forget(&ident.owner, &ident.rkey).await; } } PendingOp::Noop | PendingOp::Identity(_) | PendingOp::Account(_) | PendingOp::Parked { .. } => {} } pending}
async fn claim_repo<S: SearchSink + 'static>( source: &AtUri<DefaultStr>, repo: &Manifest<DefaultStr>, ctx: &PipelineCtx<'_, S>,) { let Some(repo_did) = owner_did_from_aturi(source) else { return; }; let Ok(slug) = Rkey::new_owned(repo.declaration.name.0.to_lowercase()) else { return; }; let ident = RepoIdent::new(repo.declaration.owner.clone(), slug); if let Some(shadow) = ctx.shadow { shadow.note_observed(&ident.owner, &ident.rkey).await; } ctx.resolver .observe( ident.owner.clone(), ident.rkey.clone(), Some(repo_did), None, &repo.created_at, ) .await; finalize_drained(ctx, take_observed(ctx.buffer, &ident).await).await;}
async fn resolve_stage<S: SearchSink + 'static>( staged: Prepared, rt: IngestRuntime<S>, replay_head: Option<HydrantCursor>,) -> Resolved { let resolve_start = rt.clock.now_instant(); let ctx = rt.pipeline_ctx(); let pending = resolve_pending(staged.pending, &ctx, replay_head).await; let resolve_end = rt.clock.now_instant(); Resolved { pending, prepare_start: staged.prepare_start, prepare_end: staged.prepare_end, resolve_start, resolve_end, }}
async fn commit_stage<S: SearchSink + 'static>( staged: Resolved, rt: IngestRuntime<S>, parallelism: usize, replay_head: Option<HydrantCursor>,) { let nsid = pending_nsid(&staged.pending.op).cloned(); let edge_count = pending_edge_count(&staged.pending.op); let regime = staged.pending.regime; let cursor = staged.pending.cursor.raw(); let Resolved { pending, prepare_start, prepare_end, resolve_start, resolve_end, } = staged; let commit_start = rt.clock.now_instant(); // runs before commit so deletes can read the cached record before it drops, and only for fresh // frames so replays dont resend analytics let analytics = rt .analytics .as_ref() .filter(|_| pending.is_fresh(rt.coverage.snapshot(), replay_head)); let mut repo_event_version = None; if let Some(analytics) = analytics { match &pending.op { PendingOp::Upsert(pieces) => { analytics.observe_upsert( &pieces.source, &pieces.parsed, pieces.cid.as_ref(), &pieces.rev, &*rt.records, ); let version = pieces .cid .as_ref() .map_or(pieces.source.as_ref(), |c| c.as_ref()); repo_event_version = Some((version.to_owned(), pieces.rev.clone())); } PendingOp::Delete { source, nsid, rev } => { analytics.observe_delete(source, nsid, rev, &*rt.records); } _ => {} } } let (settled, transitions) = commit_pending(pending, &rt.pipeline_ctx()).await; if let Some((analytics, (version, rev))) = analytics.zip(repo_event_version) { for transition in &transitions { analytics.observe_repo(transition, &version, &rev); } } let commit_end = rt.clock.now_instant(); if let Some(settled) = &settled { metrics::settled(settled); } if let Some((watch, settled)) = rt.settlements.as_deref().zip(settled) { watch.settle(settled); } if let Some(head) = replay_head { promote_at_head(&rt, head); } tracing::trace!( target: "bobbin_ingest::stage", cursor, regime = regime.as_str(), nsid = %nsid.as_ref().map(Nsid::as_str).unwrap_or(""), edge_count, prepare_us = prepare_end.duration_since(prepare_start).as_micros() as u64, queue_resolve_wait_us = resolve_start.duration_since(prepare_end).as_micros() as u64, resolve_us = resolve_end.duration_since(resolve_start).as_micros() as u64, queue_commit_wait_us = commit_start.duration_since(resolve_end).as_micros() as u64, commit_us = commit_end.duration_since(commit_start).as_micros() as u64, total_us = commit_end.duration_since(prepare_start).as_micros() as u64, parallelism, "pipeline stage timings", );}
async fn prepare_frame<S: SearchSink + 'static>( frame: HydrantFrame, ctx: &PipelineCtx<'_, S>, now: UnixMicros,) -> Pending { let cursor = HydrantCursor::new(frame.id); let signal = promotion_signal(frame.record.as_ref(), now); let regime = match frame.record.as_ref() { Some(r) if r.live => Regime::Live, Some(_) => Regime::Replay, None => Regime::NonRecord, }; let op = match frame.kind { FrameKind::Record => prepare_record(frame.record, ctx).await, FrameKind::Identity => PendingOp::Identity(frame.identity), FrameKind::Account => PendingOp::Account(frame.account), FrameKind::Other => { debug!(id = frame.id, "ignoring unknown hydrant frame kind"); PendingOp::Noop } }; Pending { cursor, signal, regime, op, }}
async fn prepare_record<S: SearchSink + 'static>( record: Option<RecordFrame>, ctx: &PipelineCtx<'_, S>,) -> PendingOp { let Some(record) = record else { debug!("record-typed frame missing payload, skipping"); metrics::frame_dropped(FrameDrop::MissingPayload); return PendingOp::Noop; }; if !record.collection.as_ref().starts_with(TANGLED_PREFIX) && !record.collection.as_ref().starts_with(ORG_TANGLED_PREFIX) { return PendingOp::Noop; } // hydrant only announces an identity when it changes, so nothing tells us // the handle of an author we first meet through a replayed record ctx.identity.warm(&record.did); let nsid = record.collection.clone(); let source = match build_source_uri(&record) { Ok(s) => s, Err(e) => { warn!(?e, "invalid frame source, dropping record"); metrics::frame_dropped(FrameDrop::InvalidSource); return PendingOp::Noop; } }; let cid = record.cid.clone(); match record.action { RecordAction::Create | RecordAction::Update => { let Some(raw) = record.record else { debug!(collection = %nsid, "create/update missing record body, clearing cache"); return rejected(source, cid, Rejection::MissingBody, None); }; let raw_bytes = Bytes::copy_from_slice(raw.get().as_bytes()); let bytes = match fallback_rfc3339(&record.rkey, &record.rev) .and_then(|fallback| synthesize_created_at(&raw_bytes, &fallback)) { Some(patched) => Bytes::from(patched), None => raw_bytes, }; let parsed = match decode_record(&record.collection, &bytes) { Ok(parsed) => parsed, Err(ExtractError::UnknownCollection(name)) => { debug!(collection = %name, "unknown sh.tangled.* collection, clearing cache"); return rejected(source, cid, Rejection::UnknownCollection, Some(&name)); } Err(e) => { warn!(?e, collection = %record.collection, "record decode failed, clearing cache"); return rejected(source, cid, Rejection::Undecodable, Some(&e.to_string())); } }; let edges = match parsed.extract_edges(&source) { Ok(es) => es, Err(e) => { warn!(?e, "edge extraction failed, clearing cache"); return rejected(source, cid, Rejection::Undecodable, Some(&e.to_string())); } }; let _ = ctx.store.intern_source(&source); PendingOp::Upsert(Box::new(UpsertPieces { source, nsid, parsed, bytes, cid: record.cid, rev: record.rev, edges, })) } RecordAction::Delete => PendingOp::Delete { source, nsid, rev: record.rev, }, RecordAction::Other => { debug!(collection = %nsid, "ignoring unknown record action"); metrics::frame_dropped(FrameDrop::UnknownAction); PendingOp::Noop } }}
fn rejected( source: AtUri<DefaultStr>, cid: Option<Cid<DefaultStr>>, reason: Rejection, detail: Option<&str>,) -> PendingOp { PendingOp::Rejected { source, cid, outcome: RecordOutcome::rejected(reason, detail), }}
fn fallback_rfc3339(rkey: &Rkey<DefaultStr>, rev: &Tid) -> Option<String> { let tid = Tid::new(rkey.as_ref()).ok().unwrap_or_else(|| rev.clone()); Some(tid_time(&tid)?.to_rfc3339_opts(chrono::SecondsFormat::Micros, true))}
fn tid_time(tid: &Tid) -> Option<chrono::DateTime<chrono::Utc>> { chrono::DateTime::from_timestamp_micros(i64::try_from(tid.timestamp()).ok()?)}
async fn evict_from_buffer(buffer: Option<&WarmingBuffer>, source: &AtUri<DefaultStr>) { if let Some(buffer) = buffer && !buffer.is_sealed() { buffer.evict_source(source).await; }}
async fn take_observed(buffer: Option<&WarmingBuffer>, ident: &RepoIdent) -> Vec<ParkedUpsert> { match buffer { Some(buffer) => buffer.take_observed(&ident.owner, &ident.rkey).await, None => Vec::new(), }}
async fn resolve_pending<S: SearchSink + 'static>( pending: Pending, ctx: &PipelineCtx<'_, S>, replay_head: Option<HydrantCursor>,) -> Pending { let fresh = pending.is_fresh(ctx.coverage.snapshot(), replay_head); let Pending { cursor, signal, regime, op, } = pending; let op = match op { PendingOp::Upsert(pieces) => match try_park_warming(ctx, cursor, fresh, pieces).await { ParkOutcome::Parked { nsid } => { return Pending { cursor, signal, regime, op: PendingOp::Parked { nsid }, }; } ParkOutcome::Passthrough(mut pieces) => { let edges = std::mem::take(&mut pieces.edges); pieces.edges = normalize_subjects(edges, ctx.resolver, ctx.coverage, ctx.shadow).await; PendingOp::Upsert(pieces) } }, other => other, }; Pending { cursor, signal, regime, op, }}
struct UpsertPieces { source: AtUri<DefaultStr>, nsid: Nsid<DefaultStr>, parsed: Record, bytes: Bytes, cid: Option<Cid<DefaultStr>>, rev: Tid, edges: Vec<Edge>,}
impl From<ParkedUpsert> for UpsertPieces { fn from(u: ParkedUpsert) -> Self { Self { source: u.source, nsid: u.nsid, parsed: u.parsed, bytes: u.bytes, cid: u.cid, rev: u.rev, edges: u.edges, } }}
enum ParkOutcome { Parked { nsid: Nsid<DefaultStr> }, Passthrough(Box<UpsertPieces>),}
async fn try_park_warming<S: SearchSink + 'static>( ctx: &PipelineCtx<'_, S>, cursor: HydrantCursor, fresh: bool, pieces: Box<UpsertPieces>,) -> ParkOutcome { let Some(buffer) = ctx.buffer else { return ParkOutcome::Passthrough(pieces); }; if ctx.coverage.snapshot().is_ready() || buffer.is_sealed() { return ParkOutcome::Passthrough(pieces); } let deps = collect_unresolved_deps(&pieces.edges, ctx.resolver).await; if deps.is_empty() { return ParkOutcome::Passthrough(pieces); } let nsid = pieces.nsid.clone(); let pieces = *pieces; let upsert = ParkedUpsert { cursor, fresh, source: pieces.source, nsid: pieces.nsid, parsed: pieces.parsed, bytes: pieces.bytes, cid: pieces.cid, rev: pieces.rev, edges: pieces.edges, }; let deps_for_shadow = ctx.shadow.is_some().then(|| deps.clone()); match buffer.try_park(upsert, deps).await { Ok(()) => { if let Some((shadow, noted)) = ctx.shadow.zip(deps_for_shadow) { let _ = futures::future::join_all(noted.into_iter().map(|dep| async move { shadow.note_unresolved(dep.owner, dep.rkey).await; })) .await; } ParkOutcome::Parked { nsid } } Err(returned) => ParkOutcome::Passthrough(Box::new(returned.into())), }}
async fn collect_unresolved_deps(edges: &[Edge], resolver: &RepoIdResolver) -> Vec<RepoIdent> { futures::stream::iter(edges) .fold(Vec::new(), |mut acc, edge| async move { let Some(uri) = edge.subject.as_uri() else { return acc; }; let Some((owner, rkey)) = parse_repo_subject_uri(uri) else { return acc; }; if resolver.cached_resolution(&owner, &rkey).await.is_some() { return acc; } let candidate = RepoIdent::new(owner, rkey); if !acc.contains(&candidate) { acc.push(candidate); } acc }) .await}
async fn finalize_drained<S: SearchSink + 'static>( ctx: &PipelineCtx<'_, S>, drained: Vec<ParkedUpsert>,) { for upsert in drained { let ParkedUpsert { cursor: _, fresh, source, nsid: _, parsed, bytes, cid, rev, edges, } = upsert; // parked upserts skipped analytics while waiting on deps, so emit now if they arrived fresh if fresh && let Some(analytics) = ctx.analytics { analytics.observe_upsert(&source, &parsed, cid.as_ref(), &rev, ctx.records); } let edges = normalize_subjects(edges, ctx.resolver, ctx.coverage, None).await; let settled_cid = settle_cid(cid.as_ref()); cache_body(ctx.records, &source, cid, bytes); // parked records wait on legacy repo at-uris, so they are never manifests or declarations upsert_record_indexes(ctx.store, ctx.claims, &source, edges, &parsed); index_search(ctx.search, ctx.resolver, &source, parsed).await; let settled = Settlement { source, cid: settled_cid, outcome: RecordOutcome::Indexed, }; metrics::settled(&settled); if let Some(watch) = ctx.settlements { watch.settle(settled); } }}
// hydrant already decoded these cids from car bytes, so garbage here means a bug upstreamfn settle_cid(cid: Option<&Cid<DefaultStr>>) -> Option<ParsedCid> { cid.and_then(|c| { c.to_ipld() .map(ParsedCid::from) .inspect_err(|e| warn!(?e, "unparseable frame cid, settling without it")) .ok() })}
async fn commit_pending<S: SearchSink + 'static>( pending: Pending, ctx: &PipelineCtx<'_, S>,) -> (Option<Settlement>, Vec<RepoTransition>) { let store = ctx.store; let claims = ctx.claims; let mut transitions = Vec::new(); let (coverage, search, records) = (ctx.coverage, ctx.search, ctx.records); let (resolver, identity) = (ctx.resolver, ctx.identity); let Pending { cursor, signal, regime: _, op, } = pending; // the settlement goes out only after its op is applied, so a writer asking // about this record sees the same thing a reader would let settled = match op { PendingOp::Noop | PendingOp::Parked { .. } => None, PendingOp::Identity(observed) => { if let Some(observed) = observed { match observed.handle { Some(handle) => identity.observe(observed.did, handle), None => identity.refresh(&observed.did), } } None } PendingOp::Account(account) => { if let Some(account) = account && !account.active { identity.deactivate(account.did); } None } PendingOp::Rejected { source, cid, outcome, } => { records.remove(&source); Some(Settlement { source, cid: settle_cid(cid.as_ref()), outcome, }) } PendingOp::Upsert(pieces) => { let UpsertPieces { source, nsid: _, parsed, bytes, cid, rev: _, edges, } = *pieces; let settled_cid = settle_cid(cid.as_ref()); cache_body(records, &source, cid, bytes); transitions = upsert_record_indexes(store, claims, &source, edges, &parsed); index_search(search, resolver, &source, parsed).await; Some(Settlement { source, cid: settled_cid, outcome: RecordOutcome::Indexed, }) } PendingOp::Delete { source, nsid, .. } => { delete_record_indexes(store, claims, &source, &nsid); records.remove(&source); search.remove(&source).await; Some(Settlement { source, cid: None, outcome: RecordOutcome::Deleted, }) } }; coverage.update(|c| c.advance(cursor).maybe_promote(signal)); (settled, transitions)}
async fn index_search<S: SearchSink>( search: &S, resolver: &RepoIdResolver, source: &AtUri<DefaultStr>, parsed: Record,) { let Some(searchable) = SearchableRecord::try_from_record(parsed) else { return; }; let Some(searchable) = searchable.normalize(resolver).await else { return; }; search.upsert(searchable.to_search_doc(source)).await;}
#[cfg(test)]#[allow(clippy::too_many_arguments)]async fn handle_frame<S: SearchSink + 'static>( frame: HydrantFrame, store: &EdgeStore, claims: &ClaimIndex, coverage: &CoverageWatch, search: &S, records: &dyn RecordStore, resolver: &RepoIdResolver, clock: &dyn Clock, now: UnixMicros,) { let _ = clock; let identity = IdentityResolver::detached(RuntimeHasher::default()); let ctx = PipelineCtx { resolver, identity: &identity, store, claims, coverage, records, search, shadow: None, buffer: None, settlements: None, analytics: None, }; let pending = claim_pending(prepare_frame(frame, &ctx, now).await, &ctx).await; let pending = resolve_pending(pending, &ctx, None).await; commit_pending(pending, &ctx).await;}
fn promotion_signal(record: Option<&RecordFrame>, now: UnixMicros) -> PromotionSignal { PromotionSignal { rev_micros: record.map(|r| r.rev.timestamp()), now_micros: now.raw(), skew_micros: READY_SKEW.as_micros() as u64, }}
fn cache_body( records: &dyn RecordStore, source: &AtUri<DefaultStr>, cid: Option<Cid<DefaultStr>>, bytes: Bytes,) { match cid { Some(cid) => records.put( source.clone(), Arc::new(RecordBody { uri: source.clone(), cid, value: bytes, }), ), None => records.remove(source), }}
async fn normalize_subjects( edges: Vec<Edge>, resolver: &RepoIdResolver, coverage: &CoverageWatch, shadow: Option<&WarmingShadowBuffer>,) -> Vec<Edge> { let warming = shadow.is_some() && !coverage.snapshot().is_ready(); futures::stream::iter(edges) .filter_map(|edge| async move { let Some(uri) = edge.subject.as_uri() else { return Some(edge); }; let Some((owner, rkey)) = parse_repo_subject_uri(uri) else { return Some(edge); }; if warming && let Some(shadow) = shadow && resolver.cached_resolution(&owner, &rkey).await.is_none() { shadow .note_unresolved(owner.clone(), rkey.clone()) .await; } match resolver.resolve(&owner, &rkey).await { Resolution::Mapped(repo_did) => Some(Edge { subject: SubjectRef::Did(repo_did), ..edge }), Resolution::NoRepoDid => { warn!( target: "bobbin_ingest::normalize", kind = %edge.kind, owner = owner.as_ref(), rkey = rkey.as_ref(), source = edge.source.as_ref(), "dropping edge: target repo has no repoDid, no canonical DID subject available", ); metrics::edge_dropped(EdgeDrop::NoRepoDid); None } Resolution::Unresolvable => { warn!( target: "bobbin_ingest::normalize", kind = %edge.kind, owner = owner.as_ref(), rkey = rkey.as_ref(), source = edge.source.as_ref(), "dropping edge: repo unresolvable, rkey-form subject will not match bare-DID queries", ); metrics::edge_dropped(EdgeDrop::RepoUnresolvable); None } } }) .collect() .await}
fn parse_repo_subject_uri(uri: &AtUri<DefaultStr>) -> Option<(Did<DefaultStr>, Rkey<DefaultStr>)> { let collection = uri.collection()?; if collection.as_ref() != "sh.tangled.repo" { return None; } let AtIdentifier::Did(authority) = uri.authority() else { return None; }; let rkey = uri.rkey()?; let owner = Did::new_owned(authority.as_ref()).ok()?; let rkey = Rkey::new_owned(rkey.as_ref()).ok()?; Some((owner, rkey))}
fn build_source_uri(r: &RecordFrame) -> Result<AtUri<DefaultStr>, IngestError> { Ok(AtUri::from_parts_owned( r.did.as_ref(), r.collection.as_ref(), r.rkey.as_ref(), )?)}
#[cfg(test)]mod tests { use super::*; use bobbin_edge_index::{Coverage, TicketStateKind}; use bobbin_record_lru::{CacheCapacity, LruRecordStore, NoopRecordStore, RecordStore}; use bobbin_runtime::{OsEntropy, RuntimeHasher, SystemClock, WsConnectFuture, WsTransport}; use bobbin_slingshot_client::SlingshotClient; use bobbin_types::search::NoopSearchSink; use jacquard_common::types::datetime::Datetime; use jacquard_common::types::handle::Handle; use jacquard_common::types::nsid::Nsid; use jacquard_common::types::tid::Tid; use serde_json::json; use wiremock::matchers::{method, path, query_param}; use wiremock::{Mock, MockServer, ResponseTemplate};
const VALID_CID: &str = "bafyreieqygohnz2zqyvtvktbjpvhutphobcmbsnt4q5lc36ri7vpcmoz4i";
struct ScriptedWs(std::sync::Mutex<Option<WsConn>>);
impl ScriptedWs { fn undialed() -> Arc<dyn WsTransport> { Arc::new(Self(std::sync::Mutex::new(None))) }
fn dialing(conn: WsConn) -> Arc<dyn WsTransport> { Arc::new(Self(std::sync::Mutex::new(Some(conn)))) } }
impl WsTransport for ScriptedWs { fn connect(&self, url: Url) -> WsConnectFuture { let conn = self.0.lock().expect("scripted ws lock").take(); Box::pin(async move { conn.ok_or_else(|| { NetworkError::Connect(format!("no scripted stream left for {url}")) }) }) } }
enum Backlog { Drained, Delivering(u64), }
struct PongingSink { ws: tokio::sync::mpsc::Sender<Result<WsMessage, NetworkError>>, backlog: Backlog, }
impl bobbin_runtime::WsSink for PongingSink { fn send<'a>(&'a mut self, msg: WsMessage) -> bobbin_runtime::WsSendFuture<'a> { Box::pin(async move { let WsMessage::Ping(payload) = msg else { return Ok(()); }; let next = match &mut self.backlog { Backlog::Drained => None, Backlog::Delivering(id) => { *id += 1; Some(*id) } }; let closed = || NetworkError::Connect("scripted stream closed".to_owned()); if let Some(id) = next { self.ws.send(other_frame(id)).await.map_err(|_| closed())?; } self.ws .send(Ok(WsMessage::Pong(payload))) .await .map_err(|_| closed()) }) } }
fn did_subj(s: &str) -> SubjectRef { SubjectRef::Did(Did::new_owned(s).unwrap()) }
fn rkey(s: &str) -> Rkey<DefaultStr> { Rkey::new_owned(s).unwrap() }
#[allow(clippy::type_complexity)] fn fresh() -> ( Arc<EdgeStore>, Arc<ClaimIndex>, Arc<CoverageWatch>, Arc<RepoIdResolver>, ) { ( Arc::new(EdgeStore::new(RuntimeHasher::default())), Arc::new(ClaimIndex::new()), Arc::new(CoverageWatch::new()), Arc::new(RepoIdResolver::detached(RuntimeHasher::default())), ) }
fn now() -> UnixMicros { SystemClock::new().now_unix_micros() }
fn sys_clock() -> SystemClock { SystemClock::new() }
fn detached_identity() -> IdentityResolver { IdentityResolver::detached(RuntimeHasher::default()) }
async fn drain_identity_warming(identity: &Arc<IdentityResolver>) { let settled_before = identity.stats().warm_resolved + identity.stats().warm_failed; let cancel = CancellationToken::new(); let warmer = tokio::spawn(identity.clone().run_warming(cancel.clone())); tokio::time::timeout(Duration::from_secs(5), async { while identity.stats().warm_resolved + identity.stats().warm_failed == settled_before { tokio::task::yield_now().await; } }) .await .expect("identity warmer never settled"); cancel.cancel(); warmer.await.unwrap(); }
fn parse_frame(value: serde_json::Value) -> HydrantFrame { let text = serde_json::to_string(&value).expect("serialize fixture"); serde_json::from_str(&text).expect("deserialize fixture") }
fn fresh_tid() -> Tid { Tid::now_0() }
#[tokio::test] async fn ignores_non_tangled_collections() { let (store, claims, cov, resolver) = fresh(); let frame: HydrantFrame = parse_frame(json!({ "id": 1, "type": "record", "record": { "live": false, "did": "did:plc:nel", "rev": fresh_tid().as_str(), "collection": "app.bsky.feed.post", "rkey": "abcabcabcabcz", "action": "create", "record": {"$type": "app.bsky.feed.post", "text": "hi"} } })); handle_frame( frame, &store, &claims, &cov, &NoopSearchSink, &NoopRecordStore, &resolver, &sys_clock(), now(), ) .await; assert_eq!(store.key_count(), 0); assert_eq!(cov.snapshot().events_processed(), 1); }
#[tokio::test] async fn create_then_delete_round_trips_a_star() { let (store, claims, cov, resolver) = fresh(); let create: HydrantFrame = parse_frame(json!({ "id": 10, "type": "record", "record": { "live": false, "did": "did:plc:olaren", "rev": fresh_tid().as_str(), "collection": "org.tangled.feed.star", "rkey": "abcabcabcabcz", "action": "create", "record": { "$type": "org.tangled.feed.star", "createdAt": "2026-05-01T00:00:00Z", "subject": {"$type": "org.tangled.feed.star#repo", "identity": "did:plc:abalone"} } } })); handle_frame( create, &store, &claims, &cov, &NoopSearchSink, &NoopRecordStore, &resolver, &sys_clock(), now(), ) .await; let key = bobbin_types::ids::EdgeKey::new( Nsid::new_static("org.tangled.feed.star").unwrap(), did_subj("did:plc:abalone"), ); assert_eq!(store.count(&key), 1);
let delete: HydrantFrame = parse_frame(json!({ "id": 11, "type": "record", "record": { "live": false, "did": "did:plc:olaren", "rev": fresh_tid().as_str(), "collection": "org.tangled.feed.star", "rkey": "abcabcabcabcz", "action": "delete", "record": null } })); handle_frame( delete, &store, &claims, &cov, &NoopSearchSink, &NoopRecordStore, &resolver, &sys_clock(), now(), ) .await; assert_eq!(store.count(&key), 0); }
#[tokio::test] async fn update_replaces_prior_edges() { let (store, claims, cov, resolver) = fresh(); let mk = |subject_did: &Did<DefaultStr>, id: u64| -> HydrantFrame { parse_frame(json!({ "id": id, "type": "record", "record": { "live": false, "did": "did:plc:olaren", "rev": fresh_tid().as_str(), "collection": "org.tangled.feed.star", "rkey": "abcabcabcabcz", "action": "update", "record": { "$type": "org.tangled.feed.star", "createdAt": "2026-05-01T00:00:00Z", "subject": {"$type": "org.tangled.feed.star#repo", "identity": subject_did.as_ref()} } } })) }; handle_frame( mk(&Did::new_owned("did:plc:abalone").unwrap(), 1), &store, &claims, &cov, &NoopSearchSink, &NoopRecordStore, &resolver, &sys_clock(), now(), ) .await; handle_frame( mk(&Did::new_owned("did:plc:uni").unwrap(), 2), &store, &claims, &cov, &NoopSearchSink, &NoopRecordStore, &resolver, &sys_clock(), now(), ) .await;
let kind = Nsid::new_static("org.tangled.feed.star").unwrap(); let old = bobbin_types::ids::EdgeKey::new(kind.clone(), did_subj("did:plc:abalone")); let new = bobbin_types::ids::EdgeKey::new(kind, did_subj("did:plc:uni")); assert_eq!(store.count(&old), 0); assert_eq!(store.count(&new), 1); }
#[test] fn replayed_live_frames_are_not_fresh_until_the_tip() { let pending = |regime, cursor, rev_micros| Pending { cursor: HydrantCursor::new(cursor), signal: PromotionSignal { rev_micros: Some(rev_micros), now_micros: 1_000_000_000, skew_micros: 1_000_000, }, regime, op: PendingOp::Noop, }; let warming = Coverage::default(); let ready = warming.force_ready(); let head = Some(HydrantCursor::new(10)); assert!(!pending(Regime::Live, 10, 999_999_999).is_fresh(ready, head)); assert!(pending(Regime::Live, 11, 1).is_fresh(warming, head)); assert!(!pending(Regime::Replay, 11, 999_999_999).is_fresh(ready, head)); // without a head it falls back to coverage and the rev assert!(!pending(Regime::Live, 1, 1).is_fresh(warming, None)); assert!(pending(Regime::Live, 1, 999_999_999).is_fresh(warming, None)); assert!(pending(Regime::Live, 1, 1).is_fresh(ready, None)); }
#[tokio::test] async fn prepare_frame_tags_regime_from_live_flag() { let (store, claims, cov, resolver) = fresh(); let search = NoopSearchSink; let records = NoopRecordStore; let identity = detached_identity(); let ctx = PipelineCtx { resolver: &resolver, identity: &identity, store: &store, claims: &claims, coverage: &cov, records: &records, search: &search, shadow: None, buffer: None, settlements: None, analytics: None, }; let mk = |live: bool| -> HydrantFrame { parse_frame(json!({ "id": 1, "type": "record", "record": { "live": live, "did": "did:plc:olaren", "rev": fresh_tid().as_str(), "collection": "org.tangled.feed.star", "rkey": "abcabcabcabcz", "action": "create", "record": { "$type": "org.tangled.feed.star", "createdAt": "2026-05-01T00:00:00Z", "subject": {"$type": "org.tangled.feed.star#repo", "identity": "did:plc:abalone"} } } })) }; let live_pending = prepare_frame(mk(true), &ctx, now()).await; assert_eq!(live_pending.regime, Regime::Live); let replay_pending = prepare_frame(mk(false), &ctx, now()).await; assert_eq!(replay_pending.regime, Regime::Replay);
let identity: HydrantFrame = parse_frame(json!({ "id": 9, "type": "identity", })); let id_pending = prepare_frame(identity, &ctx, now()).await; assert_eq!(id_pending.regime, Regime::NonRecord); }
#[tokio::test] async fn create_with_cid_warms_record_lru() { let (store, claims, cov, resolver) = fresh(); let lru = LruRecordStore::new(CacheCapacity::from_bytes(64 * 1024)); let frame: HydrantFrame = parse_frame(json!({ "id": 1, "type": "record", "record": { "live": false, "did": "did:plc:olaren", "rev": fresh_tid().as_str(), "collection": "org.tangled.feed.star", "rkey": "abcabcabcabcz", "action": "create", "cid": VALID_CID, "record": { "$type": "org.tangled.feed.star", "createdAt": "2026-05-01T00:00:00Z", "subject": {"$type": "org.tangled.feed.star#repo", "identity": "did:plc:abalone"} } } })); let source = AtUri::new_owned("at://did:plc:olaren/org.tangled.feed.star/abcabcabcabcz").unwrap(); handle_frame( frame, &store, &claims, &cov, &NoopSearchSink, &lru, &resolver, &sys_clock(), now(), ) .await; let cached = lru.get(&source).expect("hydrant cid must seed the lru"); assert_eq!(cached.cid.as_ref(), VALID_CID); let parsed: serde_json::Value = serde_json::from_slice(&cached.value).unwrap(); assert_eq!(parsed["subject"]["identity"], "did:plc:abalone"); }
#[tokio::test] async fn create_without_cid_clears_record_lru() { let (store, claims, cov, resolver) = fresh(); let lru = LruRecordStore::new(CacheCapacity::from_bytes(64 * 1024)); let source = AtUri::new_owned("at://did:plc:olaren/org.tangled.feed.star/abcabcabcabcz").unwrap(); let cid: Cid<DefaultStr> = VALID_CID.parse().unwrap(); lru.put( source.clone(), Arc::new(RecordBody { uri: source.clone(), cid, value: bytes::Bytes::from_static(b"{\"stale\":true}"), }), ); let frame: HydrantFrame = parse_frame(json!({ "id": 1, "type": "record", "record": { "live": false, "did": "did:plc:olaren", "rev": fresh_tid().as_str(), "collection": "org.tangled.feed.star", "rkey": "abcabcabcabcz", "action": "update", "record": { "$type": "org.tangled.feed.star", "createdAt": "2026-05-01T00:00:00Z", "subject": {"$type": "org.tangled.feed.star#repo", "identity": "did:plc:abalone"} } } })); handle_frame( frame, &store, &claims, &cov, &NoopSearchSink, &lru, &resolver, &sys_clock(), now(), ) .await; assert!( lru.get(&source).is_none(), "missing cid means we cannot trust the body, so the lru must be cleared", ); }
#[tokio::test] async fn delete_evicts_record_lru() { let (store, claims, cov, resolver) = fresh(); let lru = LruRecordStore::new(CacheCapacity::from_bytes(64 * 1024)); let source = AtUri::new_owned("at://did:plc:olaren/org.tangled.feed.star/abcabcabcabcz").unwrap(); let cid: Cid<DefaultStr> = VALID_CID.parse().unwrap(); lru.put( source.clone(), Arc::new(RecordBody { uri: source.clone(), cid, value: bytes::Bytes::from_static(b"{}"), }), ); let frame: HydrantFrame = parse_frame(json!({ "id": 1, "type": "record", "record": { "live": false, "did": "did:plc:olaren", "rev": fresh_tid().as_str(), "collection": "org.tangled.feed.star", "rkey": "abcabcabcabcz", "action": "delete", "record": null } })); handle_frame( frame, &store, &claims, &cov, &NoopSearchSink, &lru, &resolver, &sys_clock(), now(), ) .await; assert!(lru.get(&source).is_none()); }
#[tokio::test] async fn live_recent_event_promotes_coverage_to_ready() { let (store, claims, cov, resolver) = fresh(); assert!(!cov.snapshot().is_ready()); let frame: HydrantFrame = parse_frame(json!({ "id": 99, "type": "record", "record": { "live": true, "did": "did:plc:olaren", "rev": fresh_tid().as_str(), "collection": "org.tangled.feed.star", "rkey": "abcabcabcabcz", "action": "create", "record": { "$type": "org.tangled.feed.star", "createdAt": "2026-05-01T00:00:00Z", "subject": {"$type": "org.tangled.feed.star#repo", "identity": "did:plc:abalone"} } } })); handle_frame( frame, &store, &claims, &cov, &NoopSearchSink, &NoopRecordStore, &resolver, &sys_clock(), now(), ) .await; assert!(cov.snapshot().is_ready()); assert_eq!(cov.snapshot().last_cursor(), HydrantCursor::new(99)); }
#[tokio::test] async fn live_but_stale_rev_does_not_promote() { let (store, claims, cov, resolver) = fresh(); let stale_tid = Tid::from_time(1_000_000, 0); let frame: HydrantFrame = parse_frame(json!({ "id": 7, "type": "record", "record": { "live": true, "did": "did:plc:olaren", "rev": stale_tid.as_str(), "collection": "org.tangled.feed.star", "rkey": "abcabcabcabcz", "action": "create", "record": { "$type": "org.tangled.feed.star", "createdAt": "2026-05-01T00:00:00Z", "subject": {"$type": "org.tangled.feed.star#repo", "identity": "did:plc:abalone"} } } })); handle_frame( frame, &store, &claims, &cov, &NoopSearchSink, &NoopRecordStore, &resolver, &sys_clock(), now(), ) .await; assert!(!cov.snapshot().is_ready()); assert!(matches!(cov.snapshot(), Coverage::Warming { .. })); }
#[test] fn classify_consumer_too_slow_frame_returns_typed_variant() { let text = r#"{"type":"error","error":"ConsumerTooSlow","message":"stream socket send blocked for at least 30 seconds"}"#; match classify_text_frame(text) { Err(IngestError::ConsumerTooSlow { message }) => { assert!( message.contains("30 seconds"), "message field preserved verbatim, got: {message}" ); } other => panic!("expected ConsumerTooSlow variant, got: {other:?}"), } }
#[test] fn classify_unknown_hydrant_error_falls_back_to_generic_variant() { let text = r#"{"type":"error","error":"NewFutureCode","message":"some new failure mode"}"#; match classify_text_frame(text) { Err(IngestError::HydrantStream { code, message }) => { assert_eq!(code, "NewFutureCode"); assert_eq!(message, "some new failure mode"); } other => panic!("expected HydrantStream variant, got: {other:?}"), } }
#[test] fn classify_error_frame_without_message_uses_empty_string() { let text = r#"{"type":"error","error":"ConsumerTooSlow"}"#; match classify_text_frame(text) { Err(IngestError::ConsumerTooSlow { message }) => assert!(message.is_empty()), other => panic!("expected ConsumerTooSlow with empty message, got: {other:?}"), } }
#[test] fn classify_normal_record_frame_unchanged() { let text = r#"{"id":42,"type":"record"}"#; let frame = classify_text_frame(text).expect("normal record frame must parse"); assert_eq!(frame.id, 42); assert_eq!(frame.kind, FrameKind::Record); }
#[test] fn classify_hydrant_identity_and_account_frames() { for text in [ r#"{"id":1,"type":"account","account":{"did":"did:web:guestbook.gaze.systems","active":true,"status":"desynchronized"}}"#, r#"{"id":2,"type":"identity","identity":{"did":"did:web:guestbook.gaze.systems","handle":"guestbook.gaze.systems"}}"#, r#"{"id":43,"type":"identity","identity":{"did":"did:plc:olaren"}}"#, r#"{"id":44,"type":"account","account":{"did":"did:plc:olaren","active":false}}"#, ] { let frame = classify_text_frame(text) .unwrap_or_else(|e| panic!("hydrant frame must parse: {text} -> {e:?}")); assert!(frame.identity.is_some() || frame.account.is_some()); } }
#[test] fn classify_garbage_object_returns_decode_error() { let text = r#"{"random":"object","without":"required fields"}"#; match classify_text_frame(text) { Err(IngestError::Decode(_)) => {} other => panic!("expected Decode error, got: {other:?}"), } }
struct ScriptedWsStream { messages: std::collections::VecDeque<Result<WsMessage, NetworkError>>, }
impl WsStream for ScriptedWsStream { fn next<'a>(&'a mut self) -> bobbin_runtime::WsMessageFuture<'a> { let msg = self.messages.pop_front(); Box::pin(async move { msg }) } }
#[tokio::test] async fn reader_loop_surfaces_consumer_too_slow_from_error_frame() { let mut messages = std::collections::VecDeque::new(); messages.push_back(Ok(WsMessage::Text( r#"{"type":"error","error":"ConsumerTooSlow","message":"stream delivery blocked"}"# .to_owned(), ))); let stream: Box<dyn WsStream> = Box::new(ScriptedWsStream { messages }); let (frame_tx, _frame_rx) = tokio::sync::mpsc::channel::<HydrantFrame>(FRAME_CHANNEL_DEPTH); let (control_tx, _control_rx) = tokio::sync::mpsc::channel::<WsEvent>(CONTROL_CHANNEL_DEPTH); let cancel = CancellationToken::new();
let end = reader_loop( stream, frame_tx, control_tx, FrameCounter::default(), cancel, ) .await;
match end.error { Some(IngestError::ConsumerTooSlow { message }) => { assert_eq!(message, "stream delivery blocked"); } other => panic!("expected ConsumerTooSlow SessionEnd error, got: {other:?}"), } assert_eq!(end.outcome, SessionOutcome::Empty); }
#[tokio::test] async fn reader_loop_surfaces_unknown_hydrant_error_distinctly() { let mut messages = std::collections::VecDeque::new(); messages.push_back(Ok(WsMessage::Text( r#"{"type":"error","error":"NewFutureCode","message":"new mode"}"#.to_owned(), ))); let stream: Box<dyn WsStream> = Box::new(ScriptedWsStream { messages }); let (frame_tx, _frame_rx) = tokio::sync::mpsc::channel::<HydrantFrame>(FRAME_CHANNEL_DEPTH); let (control_tx, _control_rx) = tokio::sync::mpsc::channel::<WsEvent>(CONTROL_CHANNEL_DEPTH); let cancel = CancellationToken::new();
let end = reader_loop( stream, frame_tx, control_tx, FrameCounter::default(), cancel, ) .await;
match end.error { Some(IngestError::HydrantStream { code, message }) => { assert_eq!(code, "NewFutureCode"); assert_eq!(message, "new mode"); } other => panic!("expected HydrantStream SessionEnd error, got: {other:?}"), } }
struct ChannelWsStream { rx: tokio::sync::mpsc::Receiver<Result<WsMessage, NetworkError>>, }
impl WsStream for ChannelWsStream { fn next<'a>(&'a mut self) -> bobbin_runtime::WsMessageFuture<'a> { Box::pin(async move { self.rx.recv().await }) } }
fn star_frame_text(id: u64, rkey: &Rkey<DefaultStr>) -> String { json!({ "id": id, "type": "record", "record": { "live": false, "did": "did:plc:olaren", "rev": fresh_tid().as_str(), "collection": "org.tangled.feed.star", "rkey": rkey.as_ref(), "action": "create", "record": { "$type": "org.tangled.feed.star", "createdAt": "2026-05-01T00:00:00Z", "subject": {"$type": "org.tangled.feed.star#repo", "identity": "did:plc:abalone"} } } }) .to_string() }
#[tokio::test] async fn pong_forwards_promptly_when_frame_channel_is_full() { let (ws_tx, ws_rx) = tokio::sync::mpsc::channel::<Result<WsMessage, NetworkError>>(8); let stream: Box<dyn WsStream> = Box::new(ChannelWsStream { rx: ws_rx }); let (frame_tx, frame_rx) = tokio::sync::mpsc::channel::<HydrantFrame>(1); let (control_tx, mut control_rx) = tokio::sync::mpsc::channel::<WsEvent>(CONTROL_CHANNEL_DEPTH); let cancel = CancellationToken::new();
let prefill: HydrantFrame = parse_frame(json!({ "id": 0, "type": "record", "record": { "live": false, "did": "did:plc:olaren", "rev": fresh_tid().as_str(), "collection": "org.tangled.feed.star", "rkey": "prefilrkey001", "action": "create", "record": { "$type": "org.tangled.feed.star", "createdAt": "2026-05-01T00:00:00Z", "subject": {"$type": "org.tangled.feed.star#repo", "identity": "did:plc:abalone"} } } })); frame_tx .try_send(prefill) .expect("depth-1 frame channel must accept the prefill");
let reader_handle = tokio::spawn(reader_loop( stream, frame_tx.clone(), control_tx.clone(), FrameCounter::default(), cancel.clone(), ));
ws_tx .send(Ok(WsMessage::Text(star_frame_text( 1, &rkey("starrkeyaa001"), )))) .await .unwrap(); ws_tx.send(Ok(WsMessage::Pong(Bytes::new()))).await.unwrap();
let pong_event = tokio::time::timeout(Duration::from_millis(500), control_rx.recv()) .await .expect("pong must surface inside 500ms even when frame_tx is saturated. A reader blocked on a frame send would never poll the next ws message") .expect("control_tx was not closed"); assert!( matches!(pong_event, WsEvent::IncomingPong), "first control event must be the pong, not a held text", );
let too_slow = "{\"type\":\"error\",\"error\":\"ConsumerTooSlow\",\"message\":\"saturated\"}" .to_string(); ws_tx.send(Ok(WsMessage::Text(too_slow))).await.unwrap();
let end = tokio::time::timeout(Duration::from_secs(1), reader_handle) .await .expect("reader must exit promptly once ConsumerTooSlow is read") .expect("reader task must not panic"); match end.error { Some(IngestError::ConsumerTooSlow { message }) => assert_eq!(message, "saturated"), other => panic!("expected ConsumerTooSlow disconnect, got: {other:?}"), }
drop(frame_rx); }
#[tokio::test] async fn reader_drains_held_frames_once_processor_catches_up() { let (ws_tx, ws_rx) = tokio::sync::mpsc::channel::<Result<WsMessage, NetworkError>>(8); let stream: Box<dyn WsStream> = Box::new(ChannelWsStream { rx: ws_rx }); let (frame_tx, mut frame_rx) = tokio::sync::mpsc::channel::<HydrantFrame>(1); let (control_tx, _control_rx) = tokio::sync::mpsc::channel::<WsEvent>(CONTROL_CHANNEL_DEPTH); let cancel = CancellationToken::new();
let prefill: HydrantFrame = parse_frame(json!({ "id": 0, "type": "record", "record": { "live": false, "did": "did:plc:olaren", "rev": fresh_tid().as_str(), "collection": "org.tangled.feed.star", "rkey": "prefilrkey001", "action": "create", "record": { "$type": "org.tangled.feed.star", "createdAt": "2026-05-01T00:00:00Z", "subject": {"$type": "org.tangled.feed.star#repo", "identity": "did:plc:abalone"} } } })); frame_tx.try_send(prefill).unwrap();
let reader_handle = tokio::spawn(reader_loop( stream, frame_tx.clone(), control_tx, FrameCounter::default(), cancel.clone(), ));
ws_tx .send(Ok(WsMessage::Text(star_frame_text( 1, &rkey("heldrkeyaa001"), )))) .await .unwrap(); ws_tx .send(Ok(WsMessage::Text(star_frame_text( 2, &rkey("heldrkeyaa002"), )))) .await .unwrap();
let _drained_prefill = frame_rx.recv().await.expect("prefilled frame drains"); let first = tokio::time::timeout(Duration::from_millis(500), frame_rx.recv()) .await .expect("first held frame must reach frame_rx after slot opens") .expect("frame_tx still open"); assert_eq!(first.id, 1); let second = tokio::time::timeout(Duration::from_millis(500), frame_rx.recv()) .await .expect("second held frame must reach frame_rx after slot opens") .expect("frame_tx still open"); assert_eq!(second.id, 2);
cancel.cancel(); let _ = tokio::time::timeout(Duration::from_secs(1), reader_handle) .await .expect("reader must stop after cancel"); }
#[test] fn first_connect_uses_configured_start_cursor() { let start = HydrantCursor::new(42); assert_eq!(next_connect_cursor(Coverage::default(), start), start); }
#[test] fn reconnect_resumes_strictly_after_last_seen() { let snap = Coverage::default().advance(HydrantCursor::new(7)); assert_eq!( next_connect_cursor(snap, HydrantCursor::new(0)), HydrantCursor::new(8), ); }
#[test] fn reconnect_overrides_configured_start() { let snap = Coverage::default().advance(HydrantCursor::new(100)); assert_eq!( next_connect_cursor(snap, HydrantCursor::new(50)), HydrantCursor::new(101), ); }
#[test] fn first_connect_uses_start_even_when_first_frame_id_would_be_zero() { let start = HydrantCursor::new(7); let snap = Coverage::default(); assert_eq!(snap.last_cursor(), HydrantCursor::new(0)); assert_eq!(snap.events_processed(), 0); assert_eq!(next_connect_cursor(snap, start), start); }
#[test] fn reconnect_after_processing_id_zero_advances_to_one() { let snap = Coverage::default().advance(HydrantCursor::new(0)); assert_eq!(snap.events_processed(), 1); assert_eq!( next_connect_cursor(snap, HydrantCursor::new(99)), HydrantCursor::new(1), "events_processed disambiguates 'never seen' from 'saw id 0'", ); }
#[tokio::test] async fn identity_frame_advances_cursor_only() { let (store, claims, cov, resolver) = fresh(); let frame: HydrantFrame = parse_frame(json!({ "id": 5, "type": "identity", "identity": { "did": "did:plc:olaren", "handle": "olaren.dev", "is_active": true, "status": "active" } })); handle_frame( frame, &store, &claims, &cov, &NoopSearchSink, &NoopRecordStore, &resolver, &sys_clock(), now(), ) .await; assert_eq!(store.key_count(), 0); assert_eq!(cov.snapshot().last_cursor(), HydrantCursor::new(5)); assert!(!cov.snapshot().is_ready()); }
#[tokio::test] async fn identity_frames_refresh_incomplete_docs_and_observe_complete_ones() { let server = MockServer::start().await; Mock::given(method("GET")) .and(path("/xrpc/blue.microcosm.identity.resolveMiniDoc")) .and(query_param("identifier", "did:plc:cached")) .respond_with(ResponseTemplate::new(200).set_body_json(json!({ "did": "did:plc:cached", "handle": "cached.example.com", "pds": "https://cached-pds.example.com" }))) .expect(1) .mount(&server) .await; Mock::given(method("GET")) .and(path("/xrpc/blue.microcosm.identity.resolveMiniDoc")) .and(query_param("identifier", "did:plc:inactive")) .respond_with(ResponseTemplate::new(200).set_body_json(json!({ "did": "did:plc:inactive", "handle": "inactive.example.com", "pds": "https://inactive-pds.example.com" }))) .expect(1) .mount(&server) .await;
let (store, claims, coverage, resolver) = fresh(); let identity = Arc::new(IdentityResolver::with_slingshot( SlingshotClient::with_default_http(Url::parse(&server.uri()).unwrap()).unwrap(), Arc::new(SystemClock::new()), RuntimeHasher::default(), )); let records = NoopRecordStore; let search = NoopSearchSink; let ctx = PipelineCtx { resolver: &resolver, identity: &identity, store: &store, claims: &claims, coverage: &coverage, records: &records, search: &search, shadow: None, buffer: None, settlements: None, analytics: None, }; let apply = async |frame: HydrantFrame| { let pending = prepare_frame(frame, &ctx, now()).await; let pending = resolve_pending(pending, &ctx, None).await; commit_pending(pending, &ctx).await; };
let cached = Did::new_static("did:plc:cached").unwrap(); identity .resolve_minidoc(&AtIdentifier::Did(cached.clone())) .await .unwrap(); apply(parse_frame(json!({ "id": 1, "type": "identity", "identity": {"did": "did:plc:cached", "handle": "renamed.example.com"} }))) .await; let cached_doc = identity.get_by_did(&cached).unwrap(); assert_eq!(cached_doc.handle.as_ref(), "renamed.example.com"); assert_eq!( cached_doc.pds, Some(Url::parse("https://cached-pds.example.com").unwrap()) ); assert_eq!(identity.stats().warm_queued, 0);
let inactive = Did::new_static("did:plc:inactive").unwrap(); identity.deactivate(inactive.clone()); apply(parse_frame(json!({ "id": 2, "type": "identity", "identity": {"did": "did:plc:inactive", "handle": "frame.example.com"} }))) .await; assert_eq!(identity.stats().warm_queued, 1); drain_identity_warming(&identity).await;
let rehydrated = identity.get_by_did(&inactive).unwrap(); assert_eq!(rehydrated.handle.as_ref(), "inactive.example.com"); assert_eq!( rehydrated.pds, Some(Url::parse("https://inactive-pds.example.com").unwrap()) ); }
#[tokio::test] async fn identity_and_account_frames_drive_the_identity_resolver() { let (store, claims, coverage, resolver) = fresh(); let identity = detached_identity(); let records = NoopRecordStore; let search = NoopSearchSink; let ctx = PipelineCtx { resolver: &resolver, identity: &identity, store: &store, claims: &claims, coverage: &coverage, records: &records, search: &search, shadow: None, buffer: None, settlements: None, analytics: None, }; let olaren = Did::new_static("did:plc:olaren").unwrap(); let apply = async |frame: HydrantFrame| { let pending = prepare_frame(frame, &ctx, now()).await; let pending = resolve_pending(pending, &ctx, None).await; commit_pending(pending, &ctx).await; };
apply(parse_frame(json!({ "id": 5, "type": "identity", "identity": {"did": "did:plc:olaren", "handle": "olaren.dev"} }))) .await; let doc = identity .get_by_did(&olaren) .expect("identity frame should seed the resolver"); assert_eq!(doc.handle.as_ref(), "olaren.dev");
apply(parse_frame(json!({ "id": 6, "type": "identity", "identity": {"did": "did:plc:olaren"} }))) .await; let kept = identity .get_by_did(&olaren) .expect("a missing handle must not drop the cached one"); assert_eq!(kept.handle.as_ref(), "olaren.dev");
apply(parse_frame(json!({ "id": 7, "type": "identity", "identity": {"did": "did:plc:olaren", "handle": "olaren.dev"} }))) .await; apply(parse_frame(json!({ "id": 8, "type": "account", "account": {"did": "did:plc:olaren", "active": false, "status": "deactivated"} }))) .await; assert_eq!( identity.get_by_did(&olaren), Err(bobbin_resolver::IdentityResolveError::NotFound), "an inactive account frame is the only liveness signal hydrant sends", ); }
#[tokio::test] async fn replayed_records_warm_their_author_once() { let (store, claims, coverage, resolver) = fresh(); // a real clock, so the warm queue is live instead of inert let identity = IdentityResolver::with_slingshot( bobbin_slingshot_client::SlingshotClient::with_default_http( url::Url::parse("http://127.0.0.1:1/").unwrap(), ) .unwrap(), Arc::new(SystemClock::new()), RuntimeHasher::default(), ); let records = NoopRecordStore; let search = NoopSearchSink; let ctx = PipelineCtx { resolver: &resolver, identity: &identity, store: &store, claims: &claims, coverage: &coverage, records: &records, search: &search, shadow: None, buffer: None, settlements: None, analytics: None, }; let star = |id: u64, rkey: &str| { parse_frame(json!({ "id": id, "type": "record", "record": { "live": false, "did": "did:plc:nel", "rev": fresh_tid().as_str(), "collection": "org.tangled.feed.star", "rkey": rkey, "action": "create", "record": { "$type": "org.tangled.feed.star", "createdAt": "2026-05-01T00:00:00Z", "subject": {"$type": "org.tangled.feed.star#repo", "identity": "did:plc:abalone"} } } })) }; let apply = async |frame: HydrantFrame| { let pending = prepare_frame(frame, &ctx, now()).await; let pending = resolve_pending(pending, &ctx, None).await; commit_pending(pending, &ctx).await; };
apply(star(1, "abcabcabcabcz")).await; assert!(store.key_count() > 0, "the record must land in the index"); assert_eq!( identity.stats().warm_queued, 1, "a replayed record is the only announcement of its author", );
apply(star(2, "abcabcabcabcy")).await; assert_eq!( identity.stats().warm_queued, 1, "the queue dedups, so a busy author costs one lookup", ); }
#[tokio::test] async fn records_from_known_authors_queue_no_warm() { let (store, claims, coverage, resolver) = fresh(); let identity = IdentityResolver::with_slingshot( bobbin_slingshot_client::SlingshotClient::with_default_http( url::Url::parse("http://127.0.0.1:1/").unwrap(), ) .unwrap(), Arc::new(SystemClock::new()), RuntimeHasher::default(), ); identity.observe( Did::new_static("did:plc:nel").unwrap(), Handle::new_static("nel.dev").unwrap(), ); let records = NoopRecordStore; let search = NoopSearchSink; let ctx = PipelineCtx { resolver: &resolver, identity: &identity, store: &store, claims: &claims, coverage: &coverage, records: &records, search: &search, shadow: None, buffer: None, settlements: None, analytics: None, }; let pending = prepare_frame( parse_frame(json!({ "id": 1, "type": "record", "record": { "live": false, "did": "did:plc:nel", "rev": fresh_tid().as_str(), "collection": "org.tangled.feed.star", "rkey": "abcabcabcabcz", "action": "create", "record": { "$type": "org.tangled.feed.star", "createdAt": "2026-05-01T00:00:00Z", "subject": {"$type": "org.tangled.feed.star#repo", "identity": "did:plc:abalone"} } } })), &ctx, now(), ) .await; let pending = resolve_pending(pending, &ctx, None).await; commit_pending(pending, &ctx).await; assert_eq!(identity.stats().warm_queued, 0); }
#[tokio::test] async fn account_frame_advances_cursor_only() { let (store, claims, cov, resolver) = fresh(); let frame: HydrantFrame = parse_frame(json!({ "id": 6, "type": "account", "account": {"did": "did:plc:olaren", "active": true} })); handle_frame( frame, &store, &claims, &cov, &NoopSearchSink, &NoopRecordStore, &resolver, &sys_clock(), now(), ) .await; assert_eq!(store.key_count(), 0); assert_eq!(cov.snapshot().last_cursor(), HydrantCursor::new(6)); }
#[tokio::test] async fn unknown_frame_kind_advances_cursor_without_panic() { let (store, claims, cov, resolver) = fresh(); let frame: HydrantFrame = parse_frame(json!({"id": 8, "type": "future_event"})); assert_eq!(frame.kind, FrameKind::Other); handle_frame( frame, &store, &claims, &cov, &NoopSearchSink, &NoopRecordStore, &resolver, &sys_clock(), now(), ) .await; assert_eq!(cov.snapshot().last_cursor(), HydrantCursor::new(8)); }
fn fresh_runtime(cancel: CancellationToken) -> IngestRuntime<NoopSearchSink> { IngestRuntime { store: Arc::new(EdgeStore::new(RuntimeHasher::default())), claims: Arc::new(ClaimIndex::new()), coverage: Arc::new(CoverageWatch::new()), search: Arc::new(NoopSearchSink), records: Arc::new(NoopRecordStore) as Arc<dyn RecordStore>, resolver: Arc::new(RepoIdResolver::detached(RuntimeHasher::default())), identity: Arc::new(IdentityResolver::detached(RuntimeHasher::default())), clock: Arc::new(SystemClock::new()), entropy: Arc::new(OsEntropy), stream_health: Arc::new(StreamHealth::new()), ws: ScriptedWs::undialed(), hydrant: None, cancel, disconnects: None, warming_shadow: None, warming_buffer: None, settlements: None, analytics: None, } }
#[tokio::test(start_paused = true)] async fn cancel_token_short_circuits_reconnect_sleep() { let cfg = IngestConfig::new(Url::parse("ws://127.0.0.1:1").unwrap()); let cancel = CancellationToken::new(); let runtime = fresh_runtime(cancel.clone()); let task = tokio::spawn(async move { run(cfg, runtime).await }); tokio::time::advance(Duration::from_millis(10)).await; cancel.cancel(); let outcome = tokio::time::timeout(Duration::from_secs(1), task) .await .expect("ingest must stop within timeout once cancel fires"); assert!(matches!(outcome, Ok(Ok(()))), "got {outcome:?}"); }
struct HangingWs;
impl WsTransport for HangingWs { fn connect(&self, _url: Url) -> WsConnectFuture { Box::pin(std::future::pending::<Result<WsConn, NetworkError>>()) } }
/// one second at a time: a timer registered while the clock is jumping ahead /// would otherwise be scheduled past the jump and never fire async fn advance_secs(seconds: u64) { for _ in 0..seconds { tokio::time::advance(Duration::from_secs(1)).await; for _ in 0..4 { tokio::task::yield_now().await; } } }
#[tokio::test(start_paused = true)] async fn a_handshake_that_never_answers_is_retried_instead_of_wedging_the_stream() { let cfg = IngestConfig::new(Url::parse("ws://127.0.0.1:1").expect("hydrant url")); let cancel = CancellationToken::new(); let mut runtime = fresh_runtime(cancel.clone()); runtime.ws = Arc::new(HangingWs); let disconnects = Arc::new(DisconnectSink::new()); runtime.disconnects = Some(disconnects.clone()); let task = tokio::spawn(async move { run(cfg, runtime).await }); tokio::task::yield_now().await;
advance_secs(CONNECT_TIMEOUT.as_secs() + 2).await; assert_eq!( disconnects.count(), 1, "a connect that never answers must give up and report itself", );
advance_secs(RECONNECT_MAX_DELAY.as_secs() + CONNECT_TIMEOUT.as_secs() + 2).await; assert!( disconnects.count() >= 2, "a hung connect must not keep the ingest from trying again, got {} attempts", disconnects.count(), ); assert_eq!( disconnects.snapshot().map(|s| s.kind), Some(DisconnectKind::ConnectTimeout), );
cancel.cancel(); let outcome = tokio::time::timeout(Duration::from_secs(1), task) .await .expect("ingest must stop within timeout once cancel fires"); assert!(matches!(outcome, Ok(Ok(()))), "got {outcome:?}"); }
struct Keepalives { coverage: Arc<CoverageWatch>, cancel: CancellationToken, session: tokio::task::JoinHandle<SessionEnd>, }
fn keepalives(backlog: Backlog) -> Keepalives { let (ws, ws_rx) = tokio::sync::mpsc::channel(8); let cancel = CancellationToken::new(); let mut runtime = fresh_runtime(cancel.clone()); runtime.ws = ScriptedWs::dialing(WsConn { sink: Box::new(PongingSink { ws, backlog }), stream: Box::new(ChannelWsStream { rx: ws_rx }), }); runtime .coverage .update(|c| c.advance(HydrantCursor::new(1))); let coverage = runtime.coverage.clone(); let mut cfg = IngestConfig::new(Url::parse("ws://127.0.0.1:1").expect("hydrant url")); cfg.idle_promote_min_events = NonZeroU64::new(1).expect("one event is nonzero"); let session = tokio::spawn(async move { run_session( &cfg, HydrantCursor::new(0), None, &runtime, &FrameCounter::default(), ) .await }); Keepalives { coverage, cancel, session, } }
async fn promoted_within(k: &Keepalives, intervals: u32) -> bool { let mut coverage = k.coverage.subscribe(); tokio::time::timeout( PING_INTERVAL * intervals, coverage.wait_for(|c| c.is_ready()), ) .await .is_ok() }
fn other_frame(id: u64) -> Result<WsMessage, NetworkError> { Ok(WsMessage::Text( json!({"id": id, "type": "future_event"}).to_string(), )) }
#[tokio::test(start_paused = true)] async fn a_socket_answering_keepalives_without_a_frame_promotes_coverage() { let idle = keepalives(Backlog::Drained); assert!( promoted_within(&idle, 4).await, "the socket answers keepalives and doesn't deliver a record, so the ingest is at the tip", ); idle.cancel.cancel(); idle.session.await.expect("session panicked"); }
#[tokio::test(start_paused = true)] async fn a_socket_delivering_a_record_every_keepalive_stays_warming() { let busy = keepalives(Backlog::Delivering(1)); assert!( !promoted_within(&busy, 8).await, "a record arrives in every interval, so the ingest is behind the tip", ); busy.cancel.cancel(); busy.session.await.expect("session panicked"); }
struct HeadResponder { status: http::StatusCode, head: serde_json::Value, paths: std::sync::Mutex<Vec<String>>, }
impl bobbin_runtime::MemHttpResponder for HeadResponder { fn respond( &self, request: &bobbin_runtime::HttpRequest, ) -> bobbin_runtime::MemHttpResponse { self.paths .lock() .unwrap() .push(request.url.path().to_owned()); let body = if self.status == http::StatusCode::OK { bobbin_runtime::MemHttpBody::ok_json(json!({ "id": self.head }).to_string().into()) } else { bobbin_runtime::MemHttpBody::status_only(self.status) }; bobbin_runtime::MemHttpResponse { latency: Duration::ZERO, result: Ok(body), } } }
struct HeadReplay { coverage: Arc<CoverageWatch>, responder: Arc<HeadResponder>, cancel: CancellationToken, task: tokio::task::JoinHandle<Result<(), IngestError>>, }
/// a cold ingest against a hydrant that answers `/stream/head` with `head` /// (or `status`) and replays frames `1..=frames` before going quiet fn head_replay(status: http::StatusCode, head: serde_json::Value, frames: u64) -> HeadReplay { let (ws, ws_rx) = tokio::sync::mpsc::channel(16); (1..=frames).for_each(|id| ws.try_send(other_frame(id)).expect("room for the replay")); let cancel = CancellationToken::new(); let mut runtime = fresh_runtime(cancel.clone()); runtime.ws = ScriptedWs::dialing(WsConn { sink: Box::new(PongingSink { ws, backlog: Backlog::Drained, }), stream: Box::new(ChannelWsStream { rx: ws_rx }), }); let responder = Arc::new(HeadResponder { status, head, paths: std::sync::Mutex::new(Vec::new()), }); let http = bobbin_runtime::MemHttpTransport::shared(responder.clone(), runtime.clock.clone()); runtime.hydrant = Some( HydrantClient::new(Url::parse("ws://127.0.0.1:1").expect("hydrant url"), http) .expect("hydrant client"), ); let coverage = runtime.coverage.clone(); let cfg = IngestConfig::new(Url::parse("ws://127.0.0.1:1").expect("hydrant url")); let task = tokio::spawn(async move { run(cfg, runtime).await }); HeadReplay { coverage, responder, cancel, task, } }
async fn stop(replay: HeadReplay) { replay.cancel.cancel(); let outcome = replay.task.await.expect("ingest panicked"); assert!(matches!(outcome, Ok(())), "got {outcome:?}"); }
#[tokio::test(start_paused = true)] async fn a_replay_reaching_the_hydrant_head_promotes_without_waiting_for_idle() { let replay = head_replay(http::StatusCode::OK, json!(3), 3); let started = tokio::time::Instant::now(); let mut coverage = replay.coverage.subscribe(); coverage .wait_for(|c| c.is_ready()) .await .expect("coverage watch outlives the ingest"); assert!( started.elapsed() < PING_INTERVAL, "promotion came from the head, not the idle keepalive check, took {:?}", started.elapsed(), ); assert_eq!( replay.coverage.snapshot().last_cursor(), HydrantCursor::new(3) ); assert_eq!( *replay.responder.paths.lock().unwrap(), vec!["/stream/head"] ); stop(replay).await; }
#[tokio::test(start_paused = true)] async fn a_replay_short_of_the_hydrant_head_stays_warming() { let replay = head_replay(http::StatusCode::OK, json!(4), 3); assert!( tokio::time::timeout( PING_INTERVAL * 4, replay.coverage.subscribe().wait_for(|c| c.is_ready()) ) .await .is_err(), "frame 4 never arrived, so the replay hasn't delivered what hydrant held", ); assert_eq!(replay.coverage.snapshot().events_processed(), 3); stop(replay).await; }
#[tokio::test(start_paused = true)] async fn a_hydrant_without_a_head_leaves_promotion_to_the_idle_check() { for (status, head) in [ (http::StatusCode::NOT_FOUND, json!(null)), (http::StatusCode::OK, json!(null)), ] { let replay = head_replay(status, head, 3); assert!( tokio::time::timeout( PING_INTERVAL * 4, replay.coverage.subscribe().wait_for(|c| c.is_ready()) ) .await .is_err(), "no head and fewer than the idle minimum of events must stay warming ({status})", ); assert_eq!( replay.coverage.snapshot().events_processed(), 3, "a missing head must not keep the stream from replaying ({status})", ); stop(replay).await; } }
#[tokio::test] async fn a_quiet_socket_leaves_a_cold_ingest_warming() { let runtime = fresh_runtime(CancellationToken::new()); runtime .coverage .update(|c| c.advance(HydrantCursor::new(1))); promote_if_caught_up( &runtime, DEFAULT_IDLE_PROMOTE_MIN_EVENTS, FramesDelivered(0), ); assert!( !runtime.coverage.snapshot().is_ready(), "a stream one event in just connected, so the default 256-event minimum leaves it warming", ); }
#[test] fn jittered_stays_within_one_quarter_of_base() { let base = Duration::from_secs(1); let cap = base + Duration::from_millis(250); let entropy = OsEntropy; (0..50).for_each(|_| { let j = jittered(base, &entropy); assert!(j >= base, "jitter must not undershoot"); assert!(j <= cap, "jitter must not exceed +25%, got {:?}", j); }); }
#[tokio::test] async fn explicit_subject_did_skips_normalization() { let (store, claims, cov, resolver) = fresh(); let star: HydrantFrame = parse_frame(json!({ "id": 1, "type": "record", "record": { "live": false, "did": "did:plc:olaren", "rev": fresh_tid().as_str(), "collection": "org.tangled.feed.star", "rkey": "abcabcabcabcz", "action": "create", "record": { "$type": "org.tangled.feed.star", "createdAt": "2026-05-01T00:00:00Z", "subject": {"$type": "org.tangled.feed.star#repo", "identity": "did:plc:abalone"} } } })); handle_frame( star, &store, &claims, &cov, &NoopSearchSink, &NoopRecordStore, &resolver, &sys_clock(), now(), ) .await; let key = bobbin_types::ids::EdgeKey::new( Nsid::new_static("org.tangled.feed.star").unwrap(), did_subj("did:plc:abalone"), ); assert_eq!(store.count(&key), 1); }
fn ticket_frame(repo: &str, rkey: &str) -> HydrantFrame { parse_frame(json!({ "id": 1, "type": "record", "record": { "live": false, "did": repo, "rev": fresh_tid().as_str(), "collection": "org.tangled.track.ticket", "rkey": rkey, "action": "create", "record": { "$type": "org.tangled.track.ticket", "createdAt": "2026-05-01T00:00:00Z", "title": { "$type": "org.tangled.markup.markdown#inline", "text": "bug" }, "state": "open" } } })) }
#[tokio::test] async fn ticket_keys_on_its_repo_with_its_own_state() { let (store, claims, cov, resolver) = fresh(); handle_frame( ticket_frame("did:plc:abalone", "1"), &store, &claims, &cov, &NoopSearchSink, &NoopRecordStore, &resolver, &sys_clock(), now(), ) .await; let key = bobbin_types::ids::EdgeKey::new( Nsid::new_static(bobbin_types::edges::TICKET_ISSUE_KIND).unwrap(), did_subj("did:plc:abalone"), ); assert_eq!(store.count(&key), 1); assert_eq!( store .count_ticket_state(&key, TicketStateKind::Open) .count .get(), 1, "hydrant ingest materializes the ticket's state", ); }
#[derive(Default)] struct RecordingSearchSink { docs: tokio::sync::Mutex<Vec<bobbin_types::search::SearchDoc>>, }
impl SearchSink for RecordingSearchSink { async fn upsert(&self, doc: bobbin_types::search::SearchDoc) { self.docs.lock().await.push(doc); } async fn remove(&self, _uri: &AtUri<DefaultStr>) {} }
#[tokio::test] async fn search_index_keys_tickets_on_their_repo() { let (store, claims, cov, resolver) = fresh(); let search = RecordingSearchSink::default(); handle_frame( ticket_frame("did:plc:abalone", "1"), &store, &claims, &cov, &search, &NoopRecordStore, &resolver, &sys_clock(), now(), ) .await; let docs = search.docs.lock().await; assert_eq!(docs.len(), 1, "ticket should produce one search doc"); assert_eq!( docs[0].repo, Some(Did::new_owned("did:plc:abalone").unwrap()) ); }
#[tokio::test] async fn manifest_lists_under_its_owner_once_declared() { let (store, claims, cov, resolver) = fresh(); let key = bobbin_types::ids::EdgeKey::new( Nsid::new_static("org.tangled.repo.manifest").unwrap(), did_subj("did:plc:nel"), ); handle_frame( manifest_frame("did:plc:abalone", json!({})), &store, &claims, &cov, &NoopSearchSink, &NoopRecordStore, &resolver, &sys_clock(), now(), ) .await; assert!( store.sources_for(&key).is_empty(), "nobody declared the repo yet" ); handle_frame( declaration_frame("did:plc:nel", "abalone", "did:plc:abalone"), &store, &claims, &cov, &NoopSearchSink, &NoopRecordStore, &resolver, &sys_clock(), now(), ) .await; let owner = Did::new_owned("did:plc:nel").unwrap(); let ident = bobbin_types::ids::RepoIdent::new(owner.clone(), Rkey::new_owned("abalone").unwrap()); assert_eq!( resolver.lookup_by_name(&owner, "abalone").await, Some(ident.clone()) ); assert_eq!( resolver .lookup_by_repo_did(&Did::new_owned("did:plc:abalone").unwrap()) .await, Some(ident), ); assert_eq!( store.sources_for(&key), vec![AtUri::new_owned("at://did:plc:abalone/org.tangled.repo.manifest/self").unwrap()], "the repo lists under its declared owner, not the repo account", ); }
fn declaration_frame(owner: &str, slug: &str, repo: &str) -> HydrantFrame { parse_frame(json!({ "id": 2, "type": "record", "record": { "live": false, "did": owner, "rev": fresh_tid().as_str(), "collection": "org.tangled.repo.declaration", "rkey": slug, "action": "create", "record": { "$type": "org.tangled.repo.declaration", "createdAt": "2026-05-01T00:00:00Z", "repo": repo } } })) }
#[tokio::test] async fn renamed_manifest_hides_until_the_new_owner_declares_it() { let (store, claims, cov, resolver) = fresh(); macro_rules! feed { ($id:expr, $frame:expr) => {{ let mut frame = $frame; frame.id = $id; handle_frame( frame, &store, &claims, &cov, &NoopSearchSink, &NoopRecordStore, &resolver, &sys_clock(), now(), ) .await }}; } let manifest = AtUri::new_owned("at://did:plc:abalone/org.tangled.repo.manifest/self").unwrap(); let listed = |subject: SubjectRef| { store.sources_for(&bobbin_types::ids::EdgeKey::new( Nsid::new_static("org.tangled.repo.manifest").unwrap(), subject, )) };
feed!(1, manifest_frame("did:plc:abalone", json!({}))); feed!( 2, declaration_frame("did:plc:nel", "abalone", "did:plc:abalone") ); assert_eq!(listed(did_subj("did:plc:nel")), vec![manifest.clone()]);
feed!( 3, manifest_frame( "did:plc:abalone", json!({ "declaration": { "owner": "did:plc:bob", "name": "bar" } }), ) ); assert!(listed(did_subj("did:plc:nel")).is_empty()); assert!(listed(did_subj("did:plc:bob")).is_empty()); assert!(listed(SubjectRef::Global).is_empty()); assert_eq!(claims.repo_by_name("did:plc:nel", "abalone"), None);
feed!( 4, declaration_frame("did:plc:bob", "bar", "did:plc:abalone") ); assert!(listed(did_subj("did:plc:nel")).is_empty()); assert_eq!(listed(did_subj("did:plc:bob")), vec![manifest.clone()]); assert_eq!(listed(SubjectRef::Global), vec![manifest.clone()]); assert_eq!( claims.repo_by_name("did:plc:bob", "bar").as_deref(), Some("did:plc:abalone") );
// the old owner cleaning up their stale declaration leaves the repo alone feed!( 5, parse_frame(json!({ "id": 5, "type": "record", "record": { "live": false, "did": "did:plc:nel", "rev": fresh_tid().as_str(), "collection": "org.tangled.repo.declaration", "rkey": "abalone", "action": "delete" } })) ); assert_eq!(listed(did_subj("did:plc:bob")), vec![manifest]); }
#[tokio::test] async fn org_collaborator_indexes_without_attestation() { let (store, claims, cov, resolver) = fresh(); let frame = parse_frame(json!({ "id": 1, "type": "record", "record": { "live": false, "did": "did:plc:abalone", "rev": fresh_tid().as_str(), "collection": "org.tangled.graph.collaborator", "rkey": "did:plc:nel", "action": "create", "record": { "$type": "org.tangled.graph.collaborator", "createdAt": "2026-05-01T00:00:00Z" } } })); handle_frame( frame, &store, &claims, &cov, &NoopSearchSink, &NoopRecordStore, &resolver, &sys_clock(), now(), ) .await; let by = bobbin_types::ids::EdgeKey::new( Nsid::new_static("org.tangled.graph.collaborator.by").unwrap(), did_subj("did:plc:nel"), ); assert_eq!(store.count(&by), 1); }
fn manifest_frame(repo: &str, extra: serde_json::Value) -> HydrantFrame { let mut record = json!({ "$type": "org.tangled.repo.manifest", "createdAt": "2026-05-01T00:00:00Z", "declaration": { "owner": "did:plc:nel", "name": "abalone" } }); record .as_object_mut() .unwrap() .extend(extra.as_object().unwrap().clone()); parse_frame(json!({ "id": 1, "type": "record", "record": { "live": false, "did": repo, "rev": fresh_tid().as_str(), "collection": "org.tangled.repo.manifest", "rkey": "self", "action": "create", "record": record } })) }
#[tokio::test] async fn fork_record_edges_against_its_source() { let (store, claims, cov, resolver) = fresh(); handle_frame( manifest_frame( "did:plc:olaren", json!({ "source": { "$type": "org.tangled.repo.defs#sourceTangled", "did": "did:plc:abalone" } }), ), &store, &claims, &cov, &NoopSearchSink, &NoopRecordStore, &resolver, &sys_clock(), now(), ) .await; handle_frame( declaration_frame("did:plc:nel", "abalone", "did:plc:olaren"), &store, &claims, &cov, &NoopSearchSink, &NoopRecordStore, &resolver, &sys_clock(), now(), ) .await; let key = bobbin_types::ids::EdgeKey::new( Nsid::new_static(bobbin_types::edges::REPO_SOURCE_EDGE_KIND).unwrap(), did_subj("did:plc:abalone"), ); assert_eq!(store.count(&key), 1); }
#[tokio::test] async fn delete_repo_record_evicts_resolver_cache() { let (store, claims, cov, resolver) = fresh(); let owner = Did::new_owned("did:plc:nel").unwrap(); let rkey = Rkey::new_owned("abalone").unwrap(); handle_frame( manifest_frame("did:plc:abalone", json!({})), &store, &claims, &cov, &NoopSearchSink, &NoopRecordStore, &resolver, &sys_clock(), now(), ) .await; assert!( resolver.cached_resolution(&owner, &rkey).await.is_some(), "observe must seed the cache", ); let delete: HydrantFrame = parse_frame(json!({ "id": 2, "type": "record", "record": { "live": false, "did": "did:plc:abalone", "rev": fresh_tid().as_str(), "collection": "org.tangled.repo.manifest", "rkey": "self", "action": "delete" } })); handle_frame( delete, &store, &claims, &cov, &NoopSearchSink, &NoopRecordStore, &resolver, &sys_clock(), now(), ) .await; assert!( resolver.cached_resolution(&owner, &rkey).await.is_none(), "deleting the manifest must clear the resolver cache so future observes are not blocked by a stale Authoritative entry", ); assert_eq!( resolver.lookup_by_name(&owner, "abalone").await, None, "a deleted repo must stop answering on its url", ); }
#[tokio::test] async fn cancel_short_circuits_a_hung_ws_connect() { let _listener = tokio::net::TcpListener::bind("127.0.0.1:0") .await .expect("bind sink listener"); let port = _listener.local_addr().expect("local addr").port(); let cfg = IngestConfig::new(Url::parse(&format!("ws://127.0.0.1:{port}")).expect("hydrant url")); let cancel = CancellationToken::new(); let runtime = fresh_runtime(cancel.clone()); let task = tokio::spawn(async move { run(cfg, runtime).await }); tokio::time::sleep(Duration::from_millis(100)).await; cancel.cancel(); let outcome = tokio::time::timeout(Duration::from_secs(2), task) .await .expect("cancel must short-circuit the hung ws connect"); assert!(matches!(outcome, Ok(Ok(()))), "got {outcome:?}"); }
#[tokio::test] async fn run_session_returns_after_remote_close_when_outer_cancel_unfired() { let cfg = IngestConfig::new(Url::parse("ws://127.0.0.1:1").unwrap()); let cancel = CancellationToken::new(); let mut runtime = fresh_runtime(cancel.clone()); runtime.ws = ScriptedWs::dialing(WsConn { sink: Box::new(OkSink), stream: Box::new(ScriptedWsStream { messages: VecDeque::from([Ok(WsMessage::Close { code: 1000, reason: "bye".to_owned(), })]), }), });
let res = tokio::time::timeout( Duration::from_secs(3), run_session( &cfg, HydrantCursor::new(0), None, &runtime, &FrameCounter::default(), ), ) .await; assert!( res.is_ok(), "run_session must return after a remote Close even when outer cancel never fires - regression for a session-scoped task hanging on the parent token", ); }
struct HangingSink; impl bobbin_runtime::WsSink for HangingSink { fn send<'a>(&'a mut self, _: WsMessage) -> bobbin_runtime::WsSendFuture<'a> { Box::pin(std::future::pending()) } }
#[tokio::test(start_paused = true)] async fn timed_send_surfaces_send_timeout_when_sink_pends_forever() { let mut sink: Box<dyn bobbin_runtime::WsSink> = Box::new(HangingSink); let task = tokio::spawn(async move { timed_send(&mut sink, WsMessage::Ping(Bytes::new())).await }); tokio::time::advance(SEND_TIMEOUT + Duration::from_secs(1)).await; let result = task.await.expect("task panicked"); match result { Err(IngestError::SendTimeout(d)) => assert_eq!(d, SEND_TIMEOUT), other => panic!( "expected SendTimeout, got {other:?}. A bare ws_sink.send.await would hang forever on a half-dead socket and starve the writer's pong-deadline arm", ), } }
struct OkSink; impl bobbin_runtime::WsSink for OkSink { fn send<'a>(&'a mut self, _: WsMessage) -> bobbin_runtime::WsSendFuture<'a> { Box::pin(async move { Ok(()) }) } }
#[tokio::test] async fn timed_send_returns_ok_when_sink_succeeds_promptly() { let mut sink: Box<dyn bobbin_runtime::WsSink> = Box::new(OkSink); let result = timed_send(&mut sink, WsMessage::Ping(Bytes::new())).await; assert!(matches!(result, Ok(())), "got {result:?}"); }
#[test] fn classify_error_frame_with_id_dispatches_as_error_not_unknown_kind() { let text = r#"{"id":42,"type":"error","error":"ConsumerTooSlow","message":"slow"}"#; match classify_text_frame(text) { Err(IngestError::ConsumerTooSlow { message }) => assert_eq!(message, "slow"), other => panic!( "type=\"error\" must dispatch to the error path even when id is present, got: {other:?}" ), } }
#[tokio::test] async fn buffered_pipeline_preserves_cursor_order_under_resolve_latency_skew() { use bobbin_record_lru::RecordStore; use bobbin_types::record::RecordBody; use std::sync::Mutex;
let server = wiremock::MockServer::start().await; wiremock::Mock::given(wiremock::matchers::method("GET")) .and(wiremock::matchers::path("/xrpc/com.atproto.repo.getRecord")) .respond_with( wiremock::ResponseTemplate::new(404).set_delay(Duration::from_millis(150)), ) .mount(&server) .await;
let client = bobbin_slingshot_client::SlingshotClient::with_default_http( Url::parse(&server.uri()).unwrap(), ) .unwrap(); let clock: Arc<dyn Clock> = Arc::new(SystemClock::new()); let resolver = Arc::new(RepoIdResolver::with_slingshot( client, clock.clone(), RuntimeHasher::default(), ));
let owner = Did::new_owned("did:plc:nel").unwrap(); let abalone = Did::new_owned("did:plc:abalone").unwrap(); let fast_rkeys: [Rkey<DefaultStr>; 2] = [rkey("fastrkeyaa01"), rkey("fastrkeyaa02")]; let slow_rkeys: [Rkey<DefaultStr>; 2] = [rkey("slowrkeyaa01"), rkey("slowrkeyaa02")]; for r in &fast_rkeys { resolver .observe( owner.clone(), r.clone(), Some(abalone.clone()), None, &Datetime::raw_str("2026-05-01T00:00:00Z"), ) .await; }
#[derive(Default)] struct Capturing { urls: Mutex<Vec<AtUri<DefaultStr>>>, } impl RecordStore for Capturing { fn get(&self, _uri: &AtUri<DefaultStr>) -> Option<Arc<RecordBody>> { None } fn put(&self, uri: AtUri<DefaultStr>, _body: Arc<RecordBody>) { self.urls.lock().unwrap().push(uri); } fn remove(&self, _uri: &AtUri<DefaultStr>) {} } let capturing = Arc::new(Capturing::default());
let runtime: IngestRuntime<NoopSearchSink> = IngestRuntime { store: Arc::new(EdgeStore::new(RuntimeHasher::default())), claims: Arc::new(ClaimIndex::new()), coverage: Arc::new(CoverageWatch::new()), search: Arc::new(NoopSearchSink), records: capturing.clone() as Arc<dyn RecordStore>, resolver, identity: Arc::new(IdentityResolver::detached(RuntimeHasher::default())), clock, entropy: Arc::new(OsEntropy), stream_health: Arc::new(StreamHealth::new()), ws: ScriptedWs::undialed(), hydrant: None, cancel: CancellationToken::new(), disconnects: None, warming_shadow: None, warming_buffer: None, settlements: None, analytics: None, };
let parallelism = 4usize; let (frame_tx, frame_rx) = tokio::sync::mpsc::channel::<HydrantFrame>(64); let pipeline_rt = runtime.clone(); let pipeline = tokio::spawn(async move { let prep_rt = pipeline_rt.clone(); let resolve_rt = pipeline_rt.clone(); let commit_rt = pipeline_rt; ReceiverStream::new(frame_rx) .then(move |frame| prep_stage(frame, prep_rt.clone())) .map(move |staged| resolve_stage(staged, resolve_rt.clone(), None)) .buffered(parallelism) .for_each(move |staged| commit_stage(staged, commit_rt.clone(), parallelism, None)) .await; });
let mk = |id: u64, idx: u64, repo_rkey: &Rkey<DefaultStr>| -> HydrantFrame { parse_frame(json!({ "id": id, "type": "record", "record": { "live": false, "did": "did:plc:olaren", "rev": fresh_tid().as_str(), "collection": "org.tangled.feed.star", "rkey": format!("starrkeya{idx:04}"), "action": "create", "cid": VALID_CID, "record": { "$type": "org.tangled.feed.star", "createdAt": "2026-05-01T00:00:00Z", "subject": { "$type": "org.tangled.feed.star#record", "record": { "uri": format!("at://did:plc:nel/sh.tangled.repo/{}", repo_rkey.as_ref()), "cid": VALID_CID, }, }, } } })) };
frame_tx.send(mk(1, 1, &slow_rkeys[0])).await.unwrap(); frame_tx.send(mk(2, 2, &fast_rkeys[0])).await.unwrap(); frame_tx.send(mk(3, 3, &slow_rkeys[1])).await.unwrap(); frame_tx.send(mk(4, 4, &fast_rkeys[1])).await.unwrap(); drop(frame_tx);
pipeline.await.unwrap();
let captured = capturing.urls.lock().unwrap(); let rkeys: Vec<String> = captured .iter() .filter_map(|uri| uri.rkey().map(|r| r.as_ref().to_owned())) .collect(); assert_eq!( rkeys, vec![ "starrkeya0001".to_owned(), "starrkeya0002".to_owned(), "starrkeya0003".to_owned(), "starrkeya0004".to_owned(), ], "buffered(N) must preserve cursor order even when resolves complete out of order, with ~150ms slow vs cache-hit fast as the latency skew here", ); assert_eq!(runtime.coverage.snapshot().last_cursor().raw(), 4); }
fn settling_runtime(watch: &Arc<Settlements>) -> IngestRuntime<NoopSearchSink> { IngestRuntime { settlements: Some(watch.clone()), analytics: None, ..fresh_runtime(CancellationToken::new()) } }
async fn run_stages(frame: HydrantFrame, rt: &IngestRuntime<NoopSearchSink>) { let staged = prep_stage(frame, rt.clone()).await; let staged = claim_stage(staged, rt.clone()).await; let staged = resolve_stage(staged, rt.clone(), None).await; commit_stage(staged, rt.clone(), 1, None).await; }
fn published(watch: &Arc<Settlements>, uri: &str) -> Option<RecordOutcome> { match watch.watch(&AtUri::new_owned(uri).unwrap(), None) { bobbin_edge_index::Watch::Settled(s) => Some(s.outcome), bobbin_edge_index::Watch::Waiting(..) => None, } }
fn star_frame(collection: &str, action: &str, body: serde_json::Value) -> HydrantFrame { parse_frame(json!({ "id": 1, "type": "record", "record": { "live": true, "did": "did:plc:nel", "rev": fresh_tid().as_str(), "collection": collection, "rkey": "abcabcabcabcz", "action": action, "cid": VALID_CID, "record": body, } })) }
const STAR_URI: &str = "at://did:plc:nel/org.tangled.feed.star/abcabcabcabcz";
#[tokio::test] async fn an_indexed_record_is_published_as_indexed() { let watch = Arc::new(Settlements::new()); let rt = settling_runtime(&watch); let frame = star_frame( "org.tangled.feed.star", "create", json!({ "$type": "org.tangled.feed.star", "createdAt": "2026-05-01T00:00:00Z", "subject": {"$type": "org.tangled.feed.star#repo", "identity": "did:plc:abalone"}, }), ); run_stages(frame, &rt).await; assert_eq!(published(&watch, STAR_URI), Some(RecordOutcome::Indexed)); }
#[tokio::test] async fn a_deleted_record_is_published_as_deleted() { let watch = Arc::new(Settlements::new()); let rt = settling_runtime(&watch); let frame = star_frame("org.tangled.feed.star", "delete", json!(null)); run_stages(frame, &rt).await; assert_eq!(published(&watch, STAR_URI), Some(RecordOutcome::Deleted)); }
#[tokio::test] async fn a_collection_bobbin_does_not_index_is_published_as_rejected() { let watch = Arc::new(Settlements::new()); let rt = settling_runtime(&watch); let frame = star_frame( "sh.tangled.nonsense", "create", json!({"$type": "sh.tangled.nonsense"}), ); run_stages(frame, &rt).await; let got = published(&watch, "at://did:plc:nel/sh.tangled.nonsense/abcabcabcabcz") .expect("published"); assert_eq!(got.rejection(), Some(Rejection::UnknownCollection)); assert_eq!(got.detail(), Some("sh.tangled.nonsense")); }
#[tokio::test] async fn a_record_that_will_not_decode_is_published_as_rejected() { let watch = Arc::new(Settlements::new()); let rt = settling_runtime(&watch); let frame = star_frame( "org.tangled.feed.star", "create", json!({"$type": "org.tangled.feed.star", "subject": 17}), ); run_stages(frame, &rt).await; let got = published(&watch, STAR_URI).expect("published"); assert_eq!(got.rejection(), Some(Rejection::Undecodable)); // the writer gets the bad value back and what was expected there let detail = got.detail().unwrap_or_default(); assert!(detail.contains("17"), "detail was {detail:?}"); assert!(detail.contains("StarSubject"), "detail was {detail:?}"); }
// bobbin never sees these, so a writer waiting on one has to be told no by // the endpoint rather than left hanging #[tokio::test] async fn a_foreign_collection_publishes_nothing() { let watch = Arc::new(Settlements::new()); let rt = settling_runtime(&watch); let frame = star_frame( "app.bsky.feed.post", "create", json!({"$type": "app.bsky.feed.post", "text": "hi"}), ); run_stages(frame, &rt).await; assert_eq!( published(&watch, "at://did:plc:nel/app.bsky.feed.post/abcabcabcabcz"), None, ); }
// a parked record is not visible yet, so it must not claim to be indexed // until its deps land #[tokio::test] async fn a_parked_record_publishes_nothing_until_it_drains() { let watch = Arc::new(Settlements::new()); let buffer = Arc::new(WarmingBuffer::new(RuntimeHasher::default())); let rt = IngestRuntime { warming_buffer: Some(buffer.clone()), ..settling_runtime(&watch) }; let frame = star_frame( "org.tangled.feed.star", "create", json!({ "$type": "org.tangled.feed.star", "createdAt": "2026-05-01T00:00:00Z", "subject": { "$type": "org.tangled.feed.star#record", "record": { "uri": "at://did:plc:olaren/sh.tangled.repo/reporkeyabcde", "cid": VALID_CID, }, }, }), ); run_stages(frame, &rt).await; assert_eq!(published(&watch, STAR_URI), None);
let repo = parse_frame(json!({ "id": 2, "type": "record", "record": { "live": true, "did": "did:plc:hydrant", "rev": fresh_tid().as_str(), "collection": "org.tangled.repo.manifest", "rkey": "self", "action": "create", "cid": VALID_CID, "record": { "$type": "org.tangled.repo.manifest", "createdAt": "2026-05-01T00:00:00Z", "declaration": { "owner": "did:plc:olaren", "name": "reporkeyabcde" }, } } })); run_stages(repo, &rt).await; assert_eq!(published(&watch, STAR_URI), Some(RecordOutcome::Indexed)); }}