Something went wrong. Try again.
A downstream fork of Tranquil PDS for use in margin.cafe
Something went wrong. Try again.
12345678910111213141516171819202122232425262728293031323334353637383940414243444546474849505152535455565758596061626364656667686970717273747576777879808182838485868788899091929394959697989910010110210310410510610710810911011111211311411511611711811912012112212312412512612712812913013113213313413513613713813914014114214314414514614714814915015115215315415515615715815916016116216316416516616716816917017117217317417517617717817918018118218318418518618718818919019119219319419519619719819920020120220320420520620720820921021121221321421521621721821922022122222322422522622722822923023123223323423523623723823924024124224324424524624724824925025125225325425525625725825926026126226326426526626726826927027127227327427527627727827928028128228328428528628728828929029129229329429529629729829930030130230330430530630730830931031131231331431531631731831932032132232332432532632732832933033133233333433533633733833934034134234334434534634734834935035135235335435535635735835936036136236336436536636736836937037137237337437537637737837938038138238338438538638738838939039139239339439539639739839940040140240340440540640740840941041141241341441541641741841942042142242342442542642742842943043143243343443543643743843944044144244344444544644744844945045145245345445545645745845946046146246346446546646746846947047147247347447547647747847948048148248348448548648748848949049149249349449549649749849950050150250350450550650750850951051151251351451551651751851952052152252352452552652752852953053153253353453553653753853954054154254354454554654754854955055155255355455555655755855956056156256356456556656756856957057157257357457557657757857958058158258358458558658758858959059159259359459559659759859960060160260360460560660760860961061161261361461561661761861962062162262362462562662762862963063163263363463563663763863964064164264364464564664764864965065165265365465565665765865966066166266366466566666766866967067167267367467567667767867968068168268368468568668768868969069169269369469569669769869970070170270370470570670770870971071171271371471571671771871972072172272372472572672772872973073173273373473573673773873974074174274374474574674774874975075175275375475575675775875976076176276376476576676776876977077177277377477577677777877978078178278378478578678778878979079179279379479579679779879980080180280380480580680780880981081181281381481581681781881982082182282382482582682782882983083183283383483583683783883984084184284384484584684784884985085185285385485585685785885986086186286386486586686786886987087187287387487587687787887988088188288388488588688788888989089189289389489589689789889990090190290390490590690790890991091191291391491591691791891992092192292392492592692792892993093193293393493593693793893994094194294394494594694794894995095195295395495595695795895996096196296396496596696796896997097197297397497597697797897998098198298398498598698798898999099199299399499599699799899910001001100210031004100510061007100810091010101110121013101410151016101710181019102010211022102310241025102610271028102910301031103210331034103510361037103810391040104110421043104410451046104710481049105010511052105310541055105610571058105910601061106210631064106510661067106810691070107110721073107410751076107710781079108010811082108310841085108610871088108910901091109210931094109510961097109810991100110111021103110411051106110711081109111011111112111311141115111611171118111911201121112211231124112511261127112811291130113111321133113411351136113711381139114011411142114311441145114611471148114911501151115211531154115511561157115811591160116111621163116411651166116711681169117011711172117311741175117611771178117911801181118211831184118511861187118811891190119111921193119411951196119711981199120012011202120312041205120612071208120912101211121212131214121512161217121812191220122112221223122412251226122712281229123012311232123312341235123612371238123912401241124212431244124512461247124812491250125112521253125412551256125712581259126012611262126312641265126612671268126912701271127212731274127512761277127812791280128112821283128412851286128712881289129012911292129312941295129612971298129913001301130213031304130513061307130813091310131113121313131413151316131713181319132013211322132313241325132613271328132913301331133213331334133513361337133813391340134113421343134413451346134713481349135013511352135313541355135613571358135913601361136213631364136513661367136813691370137113721373137413751376137713781379138013811382138313841385138613871388138913901391139213931394139513961397139813991400140114021403140414051406140714081409141014111412141314141415141614171418141914201421142214231424142514261427142814291430143114321433143414351436143714381439144014411442144314441445144614471448144914501451145214531454145514561457145814591460146114621463146414651466146714681469147014711472147314741475147614771478147914801481148214831484148514861487148814891490149114921493149414951496149714981499150015011502150315041505150615071508150915101511151215131514151515161517151815191520152115221523152415251526152715281529153015311532153315341535153615371538153915401541use std::sync::Arc;
use fjall::{Database, Keyspace};use smallvec::SmallVec;use uuid::Uuid;
use super::MetastoreError;use super::backlink_ops::BacklinkOps;use super::backlinks::path_to_discriminant;use super::encoding::KeyBuilder;use super::event_ops::EventOps;use super::keys::{KeyTag, UserHash};use super::record_ops::{RecordDelete, RecordOps, RecordWrite};use super::recovery::{ BacklinkMutation, CommitMutationSet, RecordMutationDelete, RecordMutationUpsert,};use super::repo_meta::{RepoMetaValue, repo_meta_key, repo_meta_prefix};use super::repo_ops::{RepoOps, bytes_to_cid_link, cid_link_to_bytes};use super::user_block_ops::UserBlockOps;use super::user_blocks::user_block_user_prefix;use super::user_hash::UserHashMap;use crate::blockstore::TranquilBlockStore;use crate::clock::SystemClock;use crate::eventlog::EventLogBridge;use crate::io::{RealIO, StorageIO};
use tranquil_db_traits::{ ApplyCommitError, ApplyCommitInput, ApplyCommitResult, ImportBlock, ImportRecord, ImportRepoError, UserNeedingRecordBlobsBackfill, UserWithoutBlocks,};use tranquil_types::{AtUri, CidLink, Did, Tid};
use serde::{Deserialize, Serialize};
pub(crate) const RECORD_BLOBS_SCHEMA_VERSION: u8 = 1;
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]pub(crate) struct RecordBlobsValue { pub(crate) blob_cid_bytes: Vec<Vec<u8>>,}
impl RecordBlobsValue { pub(crate) fn serialize(&self) -> Vec<u8> { let payload = postcard::to_allocvec(self).expect("RecordBlobsValue serialization cannot fail"); let mut buf = Vec::with_capacity(1 + payload.len()); buf.push(RECORD_BLOBS_SCHEMA_VERSION); buf.extend_from_slice(&payload); buf }
pub(crate) fn deserialize(bytes: &[u8]) -> Option<Self> { let (&version, payload) = bytes.split_first()?; match version { RECORD_BLOBS_SCHEMA_VERSION => postcard::from_bytes(payload).ok(), _ => None, } }}
fn record_blobs_key(user_hash: UserHash, uri: &AtUri) -> SmallVec<[u8; 128]> { KeyBuilder::new() .tag(KeyTag::RECORD_BLOBS) .u64(user_hash.raw()) .string(uri.as_str()) .build()}
pub(crate) fn record_blobs_user_prefix(user_hash: UserHash) -> SmallVec<[u8; 128]> { KeyBuilder::new() .tag(KeyTag::RECORD_BLOBS) .u64(user_hash.raw()) .build()}
pub(crate) fn record_blobs_backfilled_key(user_hash: UserHash) -> SmallVec<[u8; 128]> { KeyBuilder::new() .tag(KeyTag::RECORD_BLOBS_BACKFILLED) .u64(user_hash.raw()) .build()}
pub struct CommitOps<S: StorageIO> { db: Database, repo_data: Keyspace, user_hashes: Arc<UserHashMap>, repo_ops: RepoOps, record_ops: RecordOps, user_block_ops: UserBlockOps, backlink_ops: BacklinkOps, event_ops: EventOps<S>, blockstore: Option<TranquilBlockStore<RealIO, SystemClock>>,}
impl<S: StorageIO + 'static> CommitOps<S> { pub fn new( db: Database, repo_data: Keyspace, indexes: Keyspace, user_hashes: Arc<UserHashMap>, bridge: Arc<EventLogBridge<S>>, ) -> Self { let repo_ops = RepoOps::new(repo_data.clone(), Arc::clone(&user_hashes)); let record_ops = RecordOps::new(repo_data.clone(), Arc::clone(&user_hashes)); let user_block_ops = UserBlockOps::new(repo_data.clone(), Arc::clone(&user_hashes)); let backlink_ops = BacklinkOps::new(indexes, Arc::clone(&user_hashes)); let event_ops = EventOps::new(db.clone(), repo_data.clone(), bridge); Self { db, repo_data, user_hashes, repo_ops, record_ops, user_block_ops, backlink_ops, event_ops, blockstore: None, } }
pub fn with_blockstore(mut self, blockstore: TranquilBlockStore<RealIO, SystemClock>) -> Self { self.blockstore = Some(blockstore); self }
pub fn apply_commit( &self, input: ApplyCommitInput, ) -> Result<ApplyCommitResult, ApplyCommitError> { let user_hash = self .user_hashes .get(&input.user_id) .ok_or(ApplyCommitError::RepoNotFound)?;
let key = repo_meta_key(user_hash); let meta = self .repo_data .get(key.as_slice()) .map_err(|e| ApplyCommitError::Database(e.to_string()))? .and_then(|raw| RepoMetaValue::deserialize(&raw)) .ok_or(ApplyCommitError::RepoNotFound)?;
if let Some(expected) = &input.expected_root_cid { let current = bytes_to_cid_link(&meta.repo_root_cid) .map_err(|e| ApplyCommitError::Database(e.to_string()))?; if current != *expected { return Err(ApplyCommitError::ConcurrentModification); } }
let new_cid_bytes = cid_link_to_bytes(&input.new_root_cid) .map_err(|e| ApplyCommitError::Database(e.to_string()))?;
let is_active = meta.status.is_active();
let updated_meta = RepoMetaValue { repo_root_cid: new_cid_bytes.clone(), repo_rev: input.new_rev.to_string(), ..meta };
let mut batch = self.db.batch();
self.repo_ops .write_repo_meta(&mut batch, user_hash, &updated_meta);
let upserts: Vec<RecordWrite<'_>> = input .record_upserts .iter() .map(|u| RecordWrite { collection: &u.collection, rkey: &u.rkey, cid: &u.cid, }) .collect();
self.record_ops .upsert_records(&mut batch, user_hash, &upserts) .map_err(|e| ApplyCommitError::Database(e.to_string()))?;
let deletes: Vec<RecordDelete<'_>> = input .record_deletes .iter() .map(|d| RecordDelete { collection: &d.collection, rkey: &d.rkey, }) .collect();
self.record_ops .delete_records(&mut batch, user_hash, &deletes) .map_err(|e| ApplyCommitError::Database(e.to_string()))?;
self.user_block_ops .insert_user_blocks(&mut batch, user_hash, &input.new_block_cids, &input.new_rev) .map_err(|e| ApplyCommitError::Database(e.to_string()))?;
self.user_block_ops .delete_user_blocks_by_cid(&mut batch, user_hash, &input.obsolete_block_cids) .map_err(|e| ApplyCommitError::Database(e.to_string()))?;
input.backlinks_to_remove.iter().try_for_each(|uri| { self.backlink_ops .remove_backlinks_by_uri(&mut batch, user_hash, uri) .map_err(|e| ApplyCommitError::Database(e.to_string())) })?;
self.backlink_ops .add_backlinks(&mut batch, user_hash, &input.backlinks_to_add) .map_err(|e| ApplyCommitError::Database(e.to_string()))?;
let mutation_set = CommitMutationSet { new_root_cid: new_cid_bytes.clone(), new_rev: input.new_rev.clone(), record_upserts: input .record_upserts .iter() .map(|u| { let cid_bytes = cid_link_to_bytes(&u.cid) .map_err(|e| ApplyCommitError::Database(e.to_string()))?; Ok(RecordMutationUpsert { collection: u.collection.clone(), rkey: u.rkey.clone(), cid_bytes, }) }) .collect::<Result<Vec<_>, ApplyCommitError>>()?, record_deletes: input .record_deletes .iter() .map(|d| RecordMutationDelete { collection: d.collection.clone(), rkey: d.rkey.clone(), }) .collect(), block_inserts: input.new_block_cids.clone(), block_deletes: input.obsolete_block_cids.clone(), backlink_adds: input .backlinks_to_add .iter() .map(|bl| BacklinkMutation { uri: bl.uri.clone(), path: path_to_discriminant(bl.path), link_to: bl.link_to.clone(), }) .collect(), backlink_remove_uris: input.backlinks_to_remove.clone(), }; let mutation_set_bytes = mutation_set .serialize() .map_err(|e| ApplyCommitError::Database(e.to_string()))?;
let (_seq, deferred) = self .event_ops .append_commit_event_into_batch( &mut batch, &input.commit_event, Some(&mutation_set_bytes), ) .map_err(|e| ApplyCommitError::Database(e.to_string()))?;
batch .commit() .map_err(|e| ApplyCommitError::Database(e.to_string()))?;
self.event_ops.complete_broadcast(deferred);
Ok(ApplyCommitResult { is_account_active: is_active, }) }
pub fn import_repo_data( &self, user_id: Uuid, blocks: &[ImportBlock], records: &[ImportRecord], expected_root_cid: Option<&CidLink>, ) -> Result<(), ImportRepoError> { let user_hash = self .user_hashes .get(&user_id) .ok_or(ImportRepoError::RepoNotFound)?;
let key = repo_meta_key(user_hash); let meta = self .repo_data .get(key.as_slice()) .map_err(|e| ImportRepoError::Database(e.to_string()))? .and_then(|raw| RepoMetaValue::deserialize(&raw)) .ok_or(ImportRepoError::RepoNotFound)?;
if let Some(expected) = expected_root_cid { let current = bytes_to_cid_link(&meta.repo_root_cid) .map_err(|e| ImportRepoError::Database(e.to_string()))?; if current != *expected { return Err(ImportRepoError::ConcurrentModification); } }
if let Some(bs) = &self.blockstore && !blocks.is_empty() { let block_pairs: Vec<([u8; 36], Vec<u8>)> = blocks .iter() .map(|b| { let cid: [u8; 36] = b.cid_bytes.as_slice().try_into().map_err(|_| { ImportRepoError::Database(format!( "block CID has invalid length: {} (expected 36)", b.cid_bytes.len() )) })?; Ok((cid, b.data.clone())) }) .collect::<Result<Vec<_>, ImportRepoError>>()?; bs.put_blocks_blocking(block_pairs) .map_err(|e| ImportRepoError::Database(e.to_string()))?; }
let mut batch = self.db.batch();
let upserts: Vec<RecordWrite<'_>> = records .iter() .map(|r| RecordWrite { collection: &r.collection, rkey: &r.rkey, cid: &r.record_cid, }) .collect();
self.record_ops .upsert_records(&mut batch, user_hash, &upserts) .map_err(|e| ImportRepoError::Database(e.to_string()))?;
batch .commit() .map_err(|e| ImportRepoError::Database(e.to_string())) }
pub fn insert_record_blobs( &self, repo_id: Uuid, record_uris: &[AtUri], blob_cids: &[CidLink], ) -> Result<(), MetastoreError> { let user_hash = self .user_hashes .get(&repo_id) .ok_or(MetastoreError::InvalidInput("unknown user_id"))?;
let blob_bytes: Vec<Vec<u8>> = blob_cids .iter() .map(cid_link_to_bytes) .collect::<Result<_, _>>()?;
let serialized = RecordBlobsValue { blob_cid_bytes: blob_bytes, } .serialize();
let mut batch = self.db.batch(); record_uris.iter().for_each(|uri| { let key = record_blobs_key(user_hash, uri); batch.insert(&self.repo_data, key.as_slice(), serialized.as_slice()); }); batch.commit().map_err(MetastoreError::Fjall) }
pub fn get_users_needing_record_blobs_backfill( &self, limit: i64, ) -> Result<Vec<UserNeedingRecordBlobsBackfill>, MetastoreError> { let limit_usize = usize::try_from(limit).unwrap_or(0);
self.scan_users_missing_prefix( record_blobs_backfilled_key, |meta, user_id| { let did = match meta.did { None => Err(MetastoreError::CorruptData("repo_meta missing DID field")), Some(d) => Did::new(d) .map_err(|_| MetastoreError::CorruptData("corrupt repo_meta did")), }?; Ok(UserNeedingRecordBlobsBackfill { user_id, did }) }, limit_usize, ) }
pub fn mark_record_blobs_backfill_complete(&self, repo_id: Uuid) -> Result<(), MetastoreError> { let user_hash = self .user_hashes .get(&repo_id) .ok_or(MetastoreError::InvalidInput("unknown user_id"))?; let key = record_blobs_backfilled_key(user_hash); self.repo_data .insert(key.as_slice(), &[]) .map_err(MetastoreError::Fjall) }
pub fn get_users_without_blocks(&self) -> Result<Vec<UserWithoutBlocks>, MetastoreError> { const MAX_RESULTS: usize = 10_000;
self.scan_users_missing_prefix( user_block_user_prefix, |meta, user_id| { let root_cid = bytes_to_cid_link(&meta.repo_root_cid)?; Ok(UserWithoutBlocks { user_id, repo_root_cid: root_cid, repo_rev: match meta.repo_rev.is_empty() { true => None, false => Some(Tid::new(meta.repo_rev).map_err(|_| { MetastoreError::CorruptData("corrupt repo_meta repo_rev") })?), }, }) }, MAX_RESULTS, ) }
fn scan_users_missing_prefix<T, F, P>( &self, make_prefix: P, build_result: F, limit: usize, ) -> Result<Vec<T>, MetastoreError> where F: Fn(RepoMetaValue, Uuid) -> Result<T, MetastoreError>, P: Fn(UserHash) -> SmallVec<[u8; 128]>, { let prefix = repo_meta_prefix();
self.repo_data .prefix(prefix.as_slice()) .filter_map(|guard| { let (key_bytes, val_bytes) = match guard.into_inner() { Ok(pair) => pair, Err(e) => return Some(Err(MetastoreError::Fjall(e))), };
let user_hash = match parse_user_hash_from_key(&key_bytes) { Some(h) => h, None => { tracing::warn!("skipping a repo_meta row whose key doesn't parse"); return None; } };
let check_prefix = make_prefix(user_hash); let has_entries = match self.repo_data.prefix(check_prefix.as_slice()).next() { Some(guard) => match guard.into_inner() { Ok(_) => true, Err(e) => return Some(Err(MetastoreError::Fjall(e))), }, None => false, };
match has_entries { true => None, false => { let meta = match RepoMetaValue::deserialize(&val_bytes) { Some(v) => v, None => { tracing::warn!( user_hash = user_hash.raw(), "skipping a repo_meta row whose value doesn't decode" ); return None; } }; let user_id = match self.user_hashes.get_uuid(&user_hash) { Some(id) => id, None => { tracing::warn!( user_hash = user_hash.raw(), "skipping a repo_meta row whose user_hash has no reverse mapping" ); return None; } }; match build_result(meta, user_id) { Ok(v) => Some(Ok(v)), Err(e) => { tracing::warn!( user_id = %user_id, error = %e, "skipping a repo_meta row the current validators reject" ); None } } } } }) .take(limit) .collect() }}
fn parse_user_hash_from_key(key_bytes: &[u8]) -> Option<UserHash> { use super::encoding::KeyReader; let mut reader = KeyReader::new(key_bytes); let _tag = reader.tag()?; let hash = reader.u64()?; Some(UserHash::from_raw(hash))}
#[cfg(test)]mod tests { use super::*; use crate::eventlog::{EventLog, EventLogConfig}; use crate::metastore::{Metastore, MetastoreConfig}; use tranquil_db_traits::{CommitEventData, RepoEventType}; use tranquil_types::{Handle, Nsid, Rkey};
struct TestHarness { _metastore_dir: tempfile::TempDir, _eventlog_dir: tempfile::TempDir, metastore: Metastore, bridge: Arc<EventLogBridge<RealIO>>, }
fn setup() -> TestHarness { let metastore_dir = tempfile::TempDir::new().unwrap(); let eventlog_dir = tempfile::TempDir::new().unwrap(); let segments_dir = eventlog_dir.path().join("segments"); std::fs::create_dir_all(&segments_dir).unwrap();
let metastore = Metastore::open( metastore_dir.path(), MetastoreConfig { cache_size_bytes: 64 * 1024 * 1024, }, ) .unwrap();
let event_log = EventLog::open( EventLogConfig { segments_dir, ..EventLogConfig::default() }, RealIO::new(), ) .unwrap();
let bridge = Arc::new(EventLogBridge::new(Arc::new(event_log)));
TestHarness { _metastore_dir: metastore_dir, _eventlog_dir: eventlog_dir, metastore, bridge, } }
fn test_cid_link(seed: u8) -> CidLink { let digest: [u8; 32] = std::array::from_fn(|i| seed.wrapping_add(i as u8)); let mh = multihash::Multihash::<64>::wrap(0x12, &digest).unwrap(); let c = cid::Cid::new_v1(0x71, mh); CidLink::from_cid(&c) }
fn test_did(name: &str) -> Did { Did::new(format!("did:plc:{name}")).expect("test DID is well-formed") }
fn test_handle(name: &str) -> Handle { Handle::new(format!("{name}.oyster.cafe")).expect("test handle is valid") }
fn test_nsid(name: impl Into<String>) -> Nsid { Nsid::new(name).expect("test collection is a valid NSID") }
fn test_rkey(key: impl Into<String>) -> Rkey { Rkey::new(key).expect("test record key is a valid rkey") }
fn test_rev(seq: u64) -> Tid { const ALPHABET: &[u8] = b"234567abcdefghijklmnopqrstuvwxyz"; let s: String = (0..13) .rev() .map(|i| ALPHABET[((seq >> (i * 5)) & 0x1F) as usize] as char) .collect(); Tid::new(s).expect("generated TID is valid") }
fn make_commit_ops(h: &TestHarness) -> CommitOps<RealIO> { use crate::metastore::partitions::Partition; CommitOps::new( h.metastore.database().clone(), h.metastore.partition(Partition::RepoData).clone(), h.metastore.partition(Partition::Indexes).clone(), Arc::clone(h.metastore.user_hashes()), Arc::clone(&h.bridge), ) }
fn create_test_repo(h: &TestHarness, name: &str, seed: u8) -> (Uuid, Did, CidLink) { let user_id = Uuid::new_v4(); let did = test_did(name); let handle = test_handle(name); let cid = test_cid_link(seed); h.metastore .repo_ops() .create_repo( h.metastore.database(), user_id, &did, &handle, &cid, &test_rev(0), ) .unwrap(); (user_id, did, cid) }
#[test] fn apply_commit_updates_records_and_meta() { let h = setup(); let ops = make_commit_ops(&h); let (user_id, did, root_cid) = create_test_repo(&h, "olaren", 1);
let new_root = test_cid_link(2); let record_cid = test_cid_link(3); let collection = test_nsid("app.bsky.feed.post"); let rkey = test_rkey("3k2abc");
let input = ApplyCommitInput { user_id, did: did.clone(), expected_root_cid: Some(root_cid.clone()), new_root_cid: new_root.clone(), new_rev: Tid::new("3k2aaaaaaaaab").unwrap(), new_block_cids: vec![vec![0x01, 0x02]], obsolete_block_cids: vec![], record_upserts: vec![tranquil_db_traits::RecordUpsert { collection: collection.clone(), rkey: rkey.clone(), cid: record_cid.clone(), }], record_deletes: vec![], backlinks_to_add: vec![], backlinks_to_remove: vec![], commit_event: CommitEventData { did: did.clone(), event_type: RepoEventType::Commit, commit_cid: Some(new_root.clone()), prev_cid: Some(root_cid.clone()), ops: None, blobs: None, blocks: None, prev_data_cid: None, rev: Some(Tid::new("3k2aaaaaaaaab").unwrap()), }, };
let result = ops.apply_commit(input).unwrap(); assert!(result.is_account_active);
let repo = h.metastore.repo_ops().get_repo(user_id).unwrap().unwrap(); assert_eq!(repo.repo_root_cid, new_root); assert_eq!(repo.repo_rev.as_deref(), Some("3k2aaaaaaaaab"));
let found_cid = h .metastore .record_ops() .get_record_cid(user_id, &collection, &rkey) .unwrap() .unwrap(); assert_eq!(found_cid, record_cid); }
#[test] fn apply_commit_cas_rejects_stale_root() { let h = setup(); let ops = make_commit_ops(&h); let (user_id, did, _root_cid) = create_test_repo(&h, "teq", 10);
let stale_root = test_cid_link(99); let new_root = test_cid_link(11);
let input = ApplyCommitInput { user_id, did, expected_root_cid: Some(stale_root), new_root_cid: new_root, new_rev: Tid::new("3k2aaaaaaaaab").unwrap(), new_block_cids: vec![], obsolete_block_cids: vec![], record_upserts: vec![], record_deletes: vec![], backlinks_to_add: vec![], backlinks_to_remove: vec![], commit_event: CommitEventData { did: test_did("teq"), event_type: RepoEventType::Commit, commit_cid: None, prev_cid: None, ops: None, blobs: None, blocks: None, prev_data_cid: None, rev: Some(Tid::new("3k2aaaaaaaaab").unwrap()), }, };
let result = ops.apply_commit(input); assert_eq!( result.unwrap_err(), ApplyCommitError::ConcurrentModification ); }
#[test] fn apply_commit_returns_repo_not_found_for_unknown_user() { let h = setup(); let ops = make_commit_ops(&h);
let input = ApplyCommitInput { user_id: Uuid::new_v4(), did: test_did("nonexistent"), expected_root_cid: None, new_root_cid: test_cid_link(1), new_rev: Tid::new("3k2aaaaaaaaab").unwrap(), new_block_cids: vec![], obsolete_block_cids: vec![], record_upserts: vec![], record_deletes: vec![], backlinks_to_add: vec![], backlinks_to_remove: vec![], commit_event: CommitEventData { did: test_did("nonexistent"), event_type: RepoEventType::Commit, commit_cid: None, prev_cid: None, ops: None, blobs: None, blocks: None, prev_data_cid: None, rev: None, }, };
assert_eq!( ops.apply_commit(input).unwrap_err(), ApplyCommitError::RepoNotFound ); }
#[test] fn apply_commit_record_deletes() { let h = setup(); let ops = make_commit_ops(&h); let (user_id, did, root_cid) = create_test_repo(&h, "nel", 20);
let mid_root = test_cid_link(21); let record_cid = test_cid_link(22); let collection = test_nsid("app.bsky.feed.post"); let rkey = test_rkey("3k2del");
let insert_input = ApplyCommitInput { user_id, did: did.clone(), expected_root_cid: Some(root_cid.clone()), new_root_cid: mid_root.clone(), new_rev: Tid::new("3k2aaaaaaaaab").unwrap(), new_block_cids: vec![], obsolete_block_cids: vec![], record_upserts: vec![tranquil_db_traits::RecordUpsert { collection: collection.clone(), rkey: rkey.clone(), cid: record_cid, }], record_deletes: vec![], backlinks_to_add: vec![], backlinks_to_remove: vec![], commit_event: CommitEventData { did: did.clone(), event_type: RepoEventType::Commit, commit_cid: Some(mid_root.clone()), prev_cid: None, ops: None, blobs: None, blocks: None, prev_data_cid: None, rev: Some(Tid::new("3k2aaaaaaaaab").unwrap()), }, }; ops.apply_commit(insert_input).unwrap();
assert!( h.metastore .record_ops() .get_record_cid(user_id, &collection, &rkey) .unwrap() .is_some() );
let final_root = test_cid_link(23); let delete_input = ApplyCommitInput { user_id, did: did.clone(), expected_root_cid: Some(mid_root.clone()), new_root_cid: final_root.clone(), new_rev: Tid::new("3k2aaaaaaaaac").unwrap(), new_block_cids: vec![], obsolete_block_cids: vec![], record_upserts: vec![], record_deletes: vec![tranquil_db_traits::RecordDelete { collection: collection.clone(), rkey: rkey.clone(), }], backlinks_to_add: vec![], backlinks_to_remove: vec![], commit_event: CommitEventData { did: did.clone(), event_type: RepoEventType::Commit, commit_cid: Some(final_root.clone()), prev_cid: Some(mid_root.clone()), ops: None, blobs: None, blocks: None, prev_data_cid: None, rev: Some(Tid::new("3k2aaaaaaaaac").unwrap()), }, }; ops.apply_commit(delete_input).unwrap();
assert!( h.metastore .record_ops() .get_record_cid(user_id, &collection, &rkey) .unwrap() .is_none() ); }
#[test] fn apply_commit_event_visible_after_commit() { let h = setup(); let ops = make_commit_ops(&h); let (user_id, did, root_cid) = create_test_repo(&h, "lyna", 30);
let new_root = test_cid_link(31); let input = ApplyCommitInput { user_id, did: did.clone(), expected_root_cid: Some(root_cid.clone()), new_root_cid: new_root.clone(), new_rev: Tid::new("3k2aaaaaaaaab").unwrap(), new_block_cids: vec![], obsolete_block_cids: vec![], record_upserts: vec![], record_deletes: vec![], backlinks_to_add: vec![], backlinks_to_remove: vec![], commit_event: CommitEventData { did: did.clone(), event_type: RepoEventType::Commit, commit_cid: Some(new_root.clone()), prev_cid: Some(root_cid.clone()), ops: None, blobs: None, blocks: None, prev_data_cid: None, rev: Some(Tid::new("3k2aaaaaaaaab").unwrap()), }, };
ops.apply_commit(input).unwrap(); let seq = ops.event_ops.get_max_seq();
let event = ops.event_ops.get_event_by_seq(seq).unwrap().unwrap(); assert_eq!(event.did, did); assert_eq!(event.event_type, RepoEventType::Commit); assert_eq!(event.rev.as_deref(), Some("3k2aaaaaaaaab")); }
#[test] fn import_repo_data_inserts_records() { let h = setup(); let ops = make_commit_ops(&h); let (user_id, _did, root_cid) = create_test_repo(&h, "bailey", 40);
let collection = test_nsid("app.bsky.feed.post"); let rkey = test_rkey("3k2import"); let record_cid = test_cid_link(41);
ops.import_repo_data( user_id, &[], &[ImportRecord { collection: collection.clone(), rkey: rkey.clone(), record_cid: record_cid.clone(), }], Some(&root_cid), ) .unwrap();
let found = h .metastore .record_ops() .get_record_cid(user_id, &collection, &rkey) .unwrap() .unwrap(); assert_eq!(found, record_cid); }
#[test] fn import_repo_data_cas_rejects_stale_root() { let h = setup(); let ops = make_commit_ops(&h); let (user_id, _did, _root_cid) = create_test_repo(&h, "olaren", 50);
let stale = test_cid_link(99); let result = ops.import_repo_data(user_id, &[], &[], Some(&stale)); assert_eq!(result.unwrap_err(), ImportRepoError::ConcurrentModification); }
#[test] fn insert_record_blobs_and_backfill_query() { let h = setup(); let ops = make_commit_ops(&h); let (user_id_a, did_a, _) = create_test_repo(&h, "teq", 60); let (user_id_b, _did_b, _) = create_test_repo(&h, "nel", 61);
let mut batch = h.metastore.database().batch(); [user_id_a, user_id_b].iter().for_each(|user_id| { let user_hash = ops.user_hashes.get(user_id).unwrap(); batch.remove( &ops.repo_data, record_blobs_backfilled_key(user_hash).as_slice(), ); }); batch.commit().unwrap();
let needing = ops.get_users_needing_record_blobs_backfill(100).unwrap(); assert_eq!(needing.len(), 2);
let uri = AtUri::from_parts( &did_a, &"app.bsky.feed.post".parse().unwrap(), &"3k2abc".parse().unwrap(), ); let blob_cid = test_cid_link(62); ops.insert_record_blobs(user_id_a, &[uri], &[blob_cid]) .unwrap(); ops.mark_record_blobs_backfill_complete(user_id_a).unwrap();
let needing_after = ops.get_users_needing_record_blobs_backfill(100).unwrap(); assert_eq!(needing_after.len(), 1); assert_eq!(needing_after[0].user_id, user_id_b);
ops.mark_record_blobs_backfill_complete(user_id_b).unwrap(); assert!( ops.get_users_needing_record_blobs_backfill(100) .unwrap() .is_empty() ); }
#[test] fn get_users_without_blocks_returns_users_with_no_blocks() { let h = setup(); let ops = make_commit_ops(&h); let (user_id_a, did_a, root_a) = create_test_repo(&h, "lyna", 70); let (user_id_b, _did_b, _root_b) = create_test_repo(&h, "bailey", 71);
let new_root = test_cid_link(72); let input = ApplyCommitInput { user_id: user_id_a, did: did_a.clone(), expected_root_cid: Some(root_a), new_root_cid: new_root.clone(), new_rev: Tid::new("3k2aaaaaaaaab").unwrap(), new_block_cids: vec![vec![0x01, 0x02, 0x03]], obsolete_block_cids: vec![], record_upserts: vec![], record_deletes: vec![], backlinks_to_add: vec![], backlinks_to_remove: vec![], commit_event: CommitEventData { did: did_a.clone(), event_type: RepoEventType::Commit, commit_cid: Some(new_root), prev_cid: None, ops: None, blobs: None, blocks: None, prev_data_cid: None, rev: Some(Tid::new("3k2aaaaaaaaab").unwrap()), }, }; ops.apply_commit(input).unwrap();
let without = ops.get_users_without_blocks().unwrap(); assert_eq!(without.len(), 1); assert_eq!(without[0].user_id, user_id_b); }
#[test] fn apply_commit_without_expected_root_skips_cas() { let h = setup(); let ops = make_commit_ops(&h); let (user_id, did, _root_cid) = create_test_repo(&h, "kate", 80);
let new_root = test_cid_link(81); let input = ApplyCommitInput { user_id, did: did.clone(), expected_root_cid: None, new_root_cid: new_root.clone(), new_rev: Tid::new("3k2aaaaaaaaaz").unwrap(), new_block_cids: vec![], obsolete_block_cids: vec![], record_upserts: vec![], record_deletes: vec![], backlinks_to_add: vec![], backlinks_to_remove: vec![], commit_event: CommitEventData { did, event_type: RepoEventType::Commit, commit_cid: None, prev_cid: None, ops: None, blobs: None, blocks: None, prev_data_cid: None, rev: Some(Tid::new("3k2aaaaaaaaaz").unwrap()), }, };
ops.apply_commit(input).unwrap(); }
#[test] fn apply_commit_update_preserves_new_backlinks() { use crate::metastore::backlinks::backlink_target_prefix; use crate::metastore::partitions::Partition;
let h = setup(); let ops = make_commit_ops(&h); let (user_id, did, root_cid) = create_test_repo(&h, "backlink-upd", 90);
let collection = test_nsid("app.bsky.feed.like"); let rkey = test_rkey("3k2like1"); let record_cid = test_cid_link(91); let record_uri = AtUri::from_parts(&did, &collection, &rkey);
let mid_root = test_cid_link(92); let create_input = ApplyCommitInput { user_id, did: did.clone(), expected_root_cid: Some(root_cid.clone()), new_root_cid: mid_root.clone(), new_rev: Tid::new("3k2aaaaaaaaab").unwrap(), new_block_cids: vec![], obsolete_block_cids: vec![], record_upserts: vec![tranquil_db_traits::RecordUpsert { collection: collection.clone(), rkey: rkey.clone(), cid: record_cid.clone(), }], record_deletes: vec![], backlinks_to_add: vec![tranquil_db_traits::Backlink { uri: record_uri.clone(), path: tranquil_db_traits::BacklinkPath::SubjectUri, link_to: "at://did:plc:target_a/app.bsky.feed.post/p1".to_string(), }], backlinks_to_remove: vec![], commit_event: CommitEventData { did: did.clone(), event_type: RepoEventType::Commit, commit_cid: Some(mid_root.clone()), prev_cid: Some(root_cid.clone()), ops: None, blobs: None, blocks: None, prev_data_cid: None, rev: Some(Tid::new("3k2aaaaaaaaab").unwrap()), }, }; ops.apply_commit(create_input).unwrap();
let indexes = h.metastore.partition(Partition::Indexes); let target_a_prefix = backlink_target_prefix("at://did:plc:target_a/app.bsky.feed.post/p1"); let count_a_before = indexes .prefix(target_a_prefix.as_slice()) .map(|g| g.into_inner().expect("scan must not fail")) .fold(0, |acc, _| acc + 1); assert_eq!(count_a_before, 1);
let final_root = test_cid_link(93); let new_record_cid = test_cid_link(94); let update_input = ApplyCommitInput { user_id, did: did.clone(), expected_root_cid: Some(mid_root.clone()), new_root_cid: final_root.clone(), new_rev: Tid::new("3k2aaaaaaaaac").unwrap(), new_block_cids: vec![], obsolete_block_cids: vec![], record_upserts: vec![tranquil_db_traits::RecordUpsert { collection: collection.clone(), rkey: rkey.clone(), cid: new_record_cid.clone(), }], record_deletes: vec![], backlinks_to_add: vec![tranquil_db_traits::Backlink { uri: record_uri.clone(), path: tranquil_db_traits::BacklinkPath::SubjectUri, link_to: "at://did:plc:target_b/app.bsky.feed.post/p2".to_string(), }], backlinks_to_remove: vec![record_uri], commit_event: CommitEventData { did: did.clone(), event_type: RepoEventType::Commit, commit_cid: Some(final_root.clone()), prev_cid: Some(mid_root.clone()), ops: None, blobs: None, blocks: None, prev_data_cid: None, rev: Some(Tid::new("3k2aaaaaaaaac").unwrap()), }, }; ops.apply_commit(update_input).unwrap();
let count_a_after = indexes .prefix(target_a_prefix.as_slice()) .map(|g| g.into_inner().expect("scan must not fail")) .fold(0, |acc, _| acc + 1); assert_eq!(count_a_after, 0);
let target_b_prefix = backlink_target_prefix("at://did:plc:target_b/app.bsky.feed.post/p2"); let count_b = indexes .prefix(target_b_prefix.as_slice()) .map(|g| g.into_inner().expect("scan must not fail")) .fold(0, |acc, _| acc + 1); assert_eq!(count_b, 1); }
#[test] fn crash_recovery_replays_mutation_set() { let metastore_dir = tempfile::TempDir::new().unwrap(); let eventlog_dir = tempfile::TempDir::new().unwrap(); let segments_dir = eventlog_dir.path().join("segments"); std::fs::create_dir_all(&segments_dir).unwrap();
let user_id = Uuid::new_v4(); let did = test_did("crash-nel"); let handle = test_handle("crash-nel"); let initial_root = test_cid_link(200); let new_root = test_cid_link(201); let record_cid = test_cid_link(202); let collection = test_nsid("app.bsky.feed.post"); let rkey = test_rkey("3k2crash");
let event_log = EventLog::open( EventLogConfig { segments_dir: segments_dir.clone(), ..EventLogConfig::default() }, RealIO::new(), ) .unwrap(); let event_log = Arc::new(event_log); let bridge = Arc::new(EventLogBridge::new(Arc::clone(&event_log)));
{ let metastore = Metastore::open( metastore_dir.path(), MetastoreConfig { cache_size_bytes: 64 * 1024 * 1024, }, ) .unwrap();
metastore .repo_ops() .create_repo( metastore.database(), user_id, &did, &handle, &initial_root, &test_rev(0), ) .unwrap(); metastore.persist().unwrap();
let ops = make_commit_ops_from(&metastore, &bridge); let input = ApplyCommitInput { user_id, did: did.clone(), expected_root_cid: Some(initial_root.clone()), new_root_cid: new_root.clone(), new_rev: Tid::new("3k2aaaaaaaaab").unwrap(), new_block_cids: vec![vec![0xAA, 0xBB]], obsolete_block_cids: vec![], record_upserts: vec![tranquil_db_traits::RecordUpsert { collection: collection.clone(), rkey: rkey.clone(), cid: record_cid.clone(), }], record_deletes: vec![], backlinks_to_add: vec![], backlinks_to_remove: vec![], commit_event: CommitEventData { did: did.clone(), event_type: RepoEventType::Commit, commit_cid: Some(new_root.clone()), prev_cid: Some(initial_root.clone()), ops: None, blobs: None, blocks: None, prev_data_cid: None, rev: Some(Tid::new("3k2aaaaaaaaab").unwrap()), }, };
ops.apply_commit(input).unwrap(); metastore.persist().unwrap(); }
{ let metastore = Metastore::open( metastore_dir.path(), MetastoreConfig { cache_size_bytes: 64 * 1024 * 1024, }, ) .unwrap();
let event_ops = metastore.event_ops(Arc::clone(&bridge));
event_ops.write_last_applied_cursor_direct(0).unwrap(); metastore.persist().unwrap(); }
{ let metastore = Metastore::open( metastore_dir.path(), MetastoreConfig { cache_size_bytes: 64 * 1024 * 1024, }, ) .unwrap();
let repo_before = metastore.repo_ops().get_repo(user_id).unwrap().unwrap(); assert_eq!(repo_before.repo_root_cid, new_root);
let event_ops = metastore.event_ops(Arc::clone(&bridge)); let cursor_before = event_ops.read_last_applied_cursor().unwrap(); assert_eq!(cursor_before, Some(0));
let indexes = metastore .partition(crate::metastore::partitions::Partition::Indexes) .clone(); let recovered = event_ops.recover_metastore_mutations(&indexes).unwrap(); assert!(recovered > 0, "should replay at least one event");
let cursor_after = event_ops.read_last_applied_cursor().unwrap(); assert!(cursor_after.unwrap_or(0) > 0); } }
#[test] fn crash_recovery_with_uncommitted_batch() { let metastore_dir = tempfile::TempDir::new().unwrap(); let eventlog_dir = tempfile::TempDir::new().unwrap(); let segments_dir = eventlog_dir.path().join("segments"); std::fs::create_dir_all(&segments_dir).unwrap();
let user_id = Uuid::new_v4(); let did = test_did("crash-teq"); let handle = test_handle("crash-teq"); let initial_root = test_cid_link(210); let new_root = test_cid_link(211); let record_cid = test_cid_link(212); let collection = test_nsid("app.bsky.feed.post"); let rkey = test_rkey("3k2bob");
let event_log = EventLog::open( EventLogConfig { segments_dir: segments_dir.clone(), ..EventLogConfig::default() }, RealIO::new(), ) .unwrap(); let event_log = Arc::new(event_log); let bridge = Arc::new(EventLogBridge::new(Arc::clone(&event_log)));
{ let metastore = Metastore::open( metastore_dir.path(), MetastoreConfig { cache_size_bytes: 64 * 1024 * 1024, }, ) .unwrap();
metastore .repo_ops() .create_repo( metastore.database(), user_id, &did, &handle, &initial_root, &test_rev(0), ) .unwrap(); metastore.persist().unwrap();
let event_ops = metastore.event_ops(Arc::clone(&bridge)); let mutation_set = super::CommitMutationSet { new_root_cid: super::cid_link_to_bytes(&new_root).unwrap(), new_rev: Tid::new("3k2abcdefghij").unwrap(), record_upserts: vec![super::RecordMutationUpsert { collection: collection.clone(), rkey: rkey.clone(), cid_bytes: super::cid_link_to_bytes(&record_cid).unwrap(), }], record_deletes: vec![], block_inserts: vec![vec![0xCC, 0xDD]], block_deletes: vec![], backlink_adds: vec![], backlink_remove_uris: vec![], }; let ms_bytes = mutation_set.serialize().unwrap();
let commit_data = CommitEventData { did: did.clone(), event_type: RepoEventType::Commit, commit_cid: Some(new_root.clone()), prev_cid: Some(initial_root.clone()), ops: None, blobs: None, blocks: None, prev_data_cid: None, rev: Some(Tid::new("3k2aaaaaaaaab").unwrap()), };
let mut batch = metastore.database().batch(); let (_seq, deferred) = event_ops .append_commit_event_into_batch(&mut batch, &commit_data, Some(&ms_bytes)) .unwrap();
event_ops.complete_broadcast(deferred);
drop(batch);
metastore.persist().unwrap(); }
{ let metastore = Metastore::open( metastore_dir.path(), MetastoreConfig { cache_size_bytes: 64 * 1024 * 1024, }, ) .unwrap();
let repo = metastore.repo_ops().get_repo(user_id).unwrap().unwrap(); assert_eq!(repo.repo_root_cid, initial_root);
let record = metastore .record_ops() .get_record_cid(user_id, &collection, &rkey) .unwrap(); assert!(record.is_none());
let event_ops = metastore.event_ops(Arc::clone(&bridge)); let indexes = metastore .partition(crate::metastore::partitions::Partition::Indexes) .clone(); let recovered = event_ops.recover_metastore_mutations(&indexes).unwrap(); assert_eq!(recovered, 1);
let repo_after = metastore.repo_ops().get_repo(user_id).unwrap().unwrap(); assert_eq!(repo_after.repo_root_cid, new_root); assert_eq!(repo_after.repo_rev.as_deref(), Some("3k2abcdefghij"));
let record_after = metastore .record_ops() .get_record_cid(user_id, &collection, &rkey) .unwrap(); assert_eq!(record_after, Some(record_cid)); } }
fn make_commit_ops_from( metastore: &Metastore, bridge: &Arc<EventLogBridge<RealIO>>, ) -> CommitOps<RealIO> { use crate::metastore::partitions::Partition; CommitOps::new( metastore.database().clone(), metastore.partition(Partition::RepoData).clone(), metastore.partition(Partition::Indexes).clone(), Arc::clone(metastore.user_hashes()), Arc::clone(bridge), ) }
#[test] fn apply_commit_backlinks_isolated_by_collection() { use crate::metastore::backlinks::{backlink_by_user_prefix, backlink_target_prefix}; use crate::metastore::partitions::Partition;
let h = setup(); let ops = make_commit_ops(&h); let (user_id, did, root_cid) = create_test_repo(&h, "col-iso", 95);
let col_like = test_nsid("app.bsky.feed.like"); let col_repost = test_nsid("app.bsky.feed.repost"); let rkey = test_rkey("same_rkey"); let target = "at://did:plc:someone/app.bsky.feed.post/p1";
let mid_root = test_cid_link(96); let uri_like = AtUri::from_parts(&did, &col_like, &rkey); let uri_repost = AtUri::from_parts(&did, &col_repost, &rkey);
let input = ApplyCommitInput { user_id, did: did.clone(), expected_root_cid: Some(root_cid.clone()), new_root_cid: mid_root.clone(), new_rev: Tid::new("3k2aaaaaaaaab").unwrap(), new_block_cids: vec![], obsolete_block_cids: vec![], record_upserts: vec![ tranquil_db_traits::RecordUpsert { collection: col_like.clone(), rkey: rkey.clone(), cid: test_cid_link(97), }, tranquil_db_traits::RecordUpsert { collection: col_repost.clone(), rkey: rkey.clone(), cid: test_cid_link(98), }, ], record_deletes: vec![], backlinks_to_add: vec![ tranquil_db_traits::Backlink { uri: uri_like.clone(), path: tranquil_db_traits::BacklinkPath::SubjectUri, link_to: target.to_string(), }, tranquil_db_traits::Backlink { uri: uri_repost.clone(), path: tranquil_db_traits::BacklinkPath::SubjectUri, link_to: target.to_string(), }, ], backlinks_to_remove: vec![], commit_event: CommitEventData { did: did.clone(), event_type: RepoEventType::Commit, commit_cid: Some(mid_root.clone()), prev_cid: Some(root_cid.clone()), ops: None, blobs: None, blocks: None, prev_data_cid: None, rev: Some(Tid::new("3k2aaaaaaaaab").unwrap()), }, }; ops.apply_commit(input).unwrap();
let indexes = h.metastore.partition(Partition::Indexes); let user_hash = h.metastore.user_hashes().get(&user_id).unwrap(); let target_prefix = backlink_target_prefix(target); let user_prefix = backlink_by_user_prefix(user_hash);
assert_eq!( indexes .prefix(target_prefix.as_slice()) .map(|g| g.into_inner().expect("scan must not fail")) .fold(0, |acc, _| acc + 1), 2 ); assert_eq!( indexes .prefix(user_prefix.as_slice()) .map(|g| g.into_inner().expect("scan must not fail")) .fold(0, |acc, _| acc + 1), 2 );
let final_root = test_cid_link(99); let remove_like = ApplyCommitInput { user_id, did: did.clone(), expected_root_cid: Some(mid_root.clone()), new_root_cid: final_root.clone(), new_rev: Tid::new("3k2aaaaaaaaac").unwrap(), new_block_cids: vec![], obsolete_block_cids: vec![], record_upserts: vec![], record_deletes: vec![], backlinks_to_add: vec![], backlinks_to_remove: vec![uri_like], commit_event: CommitEventData { did: did.clone(), event_type: RepoEventType::Commit, commit_cid: Some(final_root.clone()), prev_cid: Some(mid_root.clone()), ops: None, blobs: None, blocks: None, prev_data_cid: None, rev: Some(Tid::new("3k2aaaaaaaaac").unwrap()), }, }; ops.apply_commit(remove_like).unwrap();
assert_eq!( indexes .prefix(target_prefix.as_slice()) .map(|g| g.into_inner().expect("scan must not fail")) .fold(0, |acc, _| acc + 1), 1 ); assert_eq!( indexes .prefix(user_prefix.as_slice()) .map(|g| g.into_inner().expect("scan must not fail")) .fold(0, |acc, _| acc + 1), 1 ); }}