a settle writes only the records that changed since the last one
Signed-off-by: dawn <90008@klbr.net>
22 unmodified lines232425262728293031323334353637383940414243444546474849505152535455565758596061562 unmodified linesExpand all22 unmodified lines
pub(crate) struct Framing<'a> { ticker: Ticker, did: &'a RepoDid, #[cfg(test)] at_micros: Option<u64>,}
impl<'a> Framing<'a> { pub(crate) fn new(did: &'a RepoDid) -> Self { Self { ticker: Ticker::new(), did, #[cfg(test)] at_micros: None, } }
/// revs from a fixed clock, so two settles of the same records make the same commits #[cfg(test)] pub(crate) fn at(did: &'a RepoDid, micros: u64) -> Self { Self { at_micros: Some(micros), ..Self::new(did) } }}
impl Emit for Framing<'_> { fn rev(&mut self, prev: Option<&Tid>) -> Tid { #[cfg(test)] if let Some(micros) = &mut self.at_micros { *micros += 1; return Tid::from_time(*micros, 0); } self.ticker.next(prev.cloned()) }
fn frame(&self, chunk: &CommitChunk) -> Option<Bytes> {562 unmodified linesExpand all23 unmodified lines24252627282928303132333454 unmodified lines23 unmodified lines#[cfg(test)]pub use cob_state::record_cob_state;pub(crate) use cob_state::{SyncedIndex, cob_state_synced, cob_tips};pub use cob_state::{cob_state, cob_state_dirs, kept_cob_states};#[cfg(test)]pub(crate) use pool::APPLIED;pub(crate) use pool::{POOLED, Pool, Prior};pub(crate) use pool::{POOLED, Pool, Prior, settle_at};pub use pool::{kept_served, last_seal, served_dirs, settle};
pub struct RepoSource { pub did: RepoDid,54 unmodified lines8 unmodified lines9101112131314151617181913 unmodified lines3334353637383940417 unmodified lines49505152535455565758596061626364656667686970717248 unmodified lines121122123124125126127128129130131132133134135136137138139140141142143144145146106107147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263109110264265112113266267268115269270271272273274275276277278279280281282167 unmodified linesExpand all8 unmodified linesuse knot_record::git_refs;use knot_record::grant::GrantCollection;use knot_record::manifest::{ManifestMeta, ManifestRecord};use knot_runtime::Signer;use knot_types::{AccountDid, MstKey, RecordCid};use knot_types::{ AccountDid, MstKey, RecordAddress, RecordCid, RecordCollection, RecordRkey, UnixSeconds,};
use crate::cas::atomic_write;use crate::plan::repo_key;use crate::store::RepoStore;13 unmodified linesconst POOL_KIND: &str = "atproto";pub(crate) const SEAL_KIND: &str = "atproto-seal";
const LAST_SEALED: &str = "sealed";const SETTLED: &str = "settled.json";
#[derive(serde::Serialize, serde::Deserialize)]pub(crate) struct Sealed { commit: String,7 unmodified lines chain_root: &Path, signer: &dyn Signer, source: &RepoSource,) -> anyhow::Result<SettledRepo> { settle_at( chain_root, signer, source, unix_now(), &mut crate::firehose::Framing::new(&source.did), )}
pub(crate) fn settle_at( chain_root: &Path, signer: &dyn Signer, source: &RepoSource, now: UnixSeconds, framing: &mut crate::firehose::Framing,) -> anyhow::Result<SettledRepo> { let chain_repo = open_chain(chain_root, &source.did)?; let versions_path = chain_root .join(repo_key(source.did.as_str())) .join("versions.json");48 unmodified lines .chain(knot_cobs::social_collections()) .chain([GrantCollection::Collaborator.records()]) .collect();
let settled_path = chain_root.join(repo_key(source.did.as_str())).join(SETTLED); let written = Settled::written(&due); let applied = match &written { Some(written) => { let clock = (now, &mut *framing); apply_changes( &chain_repo, source, signer, clock, &settled_path, written, &due, ) .unwrap_or_else(|error| { tracing::warn!(repo = %source.did, "settling only the changed records: {error:#}"); false }) } None => false, }; if !applied { block_on(chain::settle( &chain_repo, block_on(chain::settle( &chain_repo, &source.did, &collections, &DueRecords(due), signer, framing, now, ))?; } match (written, chain::tip(&chain_repo)?) { (Some(records), Some(tip)) => { let settled = Settled { tip: tip.git.to_hex(), records, }; atomic_write(&settled_path, &serde_json::to_vec(&settled)?) .with_context(|| format!("write {}", settled_path.display()))?; } _ => remove_settled(&settled_path)?, } sealed(chain_root, &chain_repo, source, cobs.versions.clone())}
/// the records the last settle left on the chain, and the chain tip they're on#[derive(serde::Serialize, serde::Deserialize)]struct Settled { tip: String, records: BTreeMap<String, String>,}
impl Settled { /// a kept record's body is unreadable, so only the full settle knows what's on the chain fn written(due: &BTreeMap<MstKey, chain::Due>) -> Option<BTreeMap<String, String>> { due.iter() .map(|(key, due)| match due { chain::Due::Write(body) => { Some((key.as_str().to_string(), body.cid().cid().to_string())) } chain::Due::Keep(_) => None, }) .collect() }}
fn remove_settled(path: &Path) -> anyhow::Result<()> { match std::fs::remove_file(path) { Err(error) if error.kind() != std::io::ErrorKind::NotFound => { Err(error).with_context(|| format!("remove {}", path.display())) } _ => Ok(()), }}
/// writes only what changed since the last settle, when the chain is still where that one left it./// false leaves the chain alone for the full settle.fn apply_changes( chain_repo: &Repo, source: &RepoSource, signer: &dyn Signer, (now, framing): (UnixSeconds, &mut crate::firehose::Framing), settled_path: &Path, written: &BTreeMap<String, String>, due: &BTreeMap<MstKey, chain::Due>,) -> anyhow::Result<bool> { let Some(last) = std::fs::read(settled_path) .ok() .and_then(|bytes| serde_json::from_slice::<Settled>(&bytes).ok()) else { return Ok(false); }; let Some(tip) = chain::tip(chain_repo)? else { return Ok(false); }; if last.tip != tip.git.to_hex() { return Ok(false); } let address = |key: &str| -> anyhow::Result<RecordAddress> { let (collection, rkey) = key .split_once('/') .context("a record key without a collection")?; Ok(RecordAddress::new( RecordCollection::new(collection)?, RecordRkey::new(rkey)?, )) }; let from = |cid: &str| -> anyhow::Result<RecordCid> { RecordCid::new(cid.parse()?).context("a settled record cid isn't a record cid") }; let mut ops = Vec::new(); for (key, due) in due { let chain::Due::Write(record) = due else { return Ok(false); }; let op = match last.records.get(key.as_str()) { None => chain::RecordOp::Create { record: record.clone(), }, Some(cid) if *cid == written[key.as_str()] => continue, Some(cid) => chain::RecordOp::Update { record: record.clone(), from: from(cid)?, }, }; ops.push((address(key.as_str())?, op)); } for (key, cid) in &last.records { if !written.contains_key(key) { ops.push((address(key)?, chain::RecordOp::Delete { from: from(cid)? })); } } if ops.len() > chain::MAX_OPS_PER_APPLY { return Ok(false); } block_on(chain::apply( chain_repo, &source.did, &collections, &DueRecords(due), &ops, signer, &mut crate::firehose::Framing::new(&source.did), unix_now(), framing, now, ))?; sealed(chain_root, &chain_repo, source, cobs.versions.clone()) #[cfg(test)] APPLIED.set(APPLIED.get() + 1); Ok(true)}
#[cfg(test)]thread_local! { pub(crate) static APPLIED: std::cell::Cell<usize> = const { std::cell::Cell::new(0) };}
fn sealed( chain_root: &Path, chain_repo: &Repo, source: &RepoSource,167 unmodified linesExpand all1587 unmodified linesExpand all1588158915901591159215931594159515961597159815991600160116021603160416051606160716081609161016111612161316141615161616171618161916201621162216231624162516261627162816291630163116321633163416351636163716381639164016411642164316441645164616471648164916501651165216531654165516561657165816591660166116621663166416651666166716681669167016711672167316741675167616771678167916801681168216831684168516861687168816891690169116921693169416951696169716981699170017011702170317041705170617071708170917101711171217131714171517161717171817191720172117221723172417251726172717281729173017311732173317341735173617371738173917401741174217431744174517461747174817491750175117521753175417551756175717581759176017611762176317641765176617671768176917701771177217731774177517761777177817791780178117821783178417851786178717881789179017911792179317941795179617971798179918001801180218031804180518061807180818091810181118121813172 unmodified linesExpand all1587 unmodified linesExpand all } assert!(written > 100, "only {written} writes landed");}
fn files_under(root: &std::path::Path) -> BTreeMap<String, Vec<u8>> { let mut files = BTreeMap::new(); let mut dirs = vec![root.to_path_buf()]; while let Some(dir) = dirs.pop() { let Ok(entries) = std::fs::read_dir(&dir) else { continue; }; for entry in entries.flatten() { let path = entry.path(); let relative = path.strip_prefix(root).unwrap().display().to_string(); if relative.ends_with("chain.git") { continue; } let meta = std::fs::symlink_metadata(&path).unwrap(); if meta.is_dir() { dirs.push(path); } else if meta.is_symlink() { let target = std::fs::read_link(&path).unwrap(); files.insert(relative, target.display().to_string().into_bytes()); } else { files.insert(relative, std::fs::read(&path).unwrap()); } } } files}
/// one chain settled from what changed, one settled whole every time, each with its own sealsstruct Settling { dir: tempfile::TempDir, whole: bool,}
impl Settling { fn new(did: &RepoDid, whole: bool) -> Self { let dir = tempfile::tempdir().unwrap(); let key = crate::plan::repo_key(did.as_str()); std::fs::create_dir_all(dir.path().join("atproto").join(&key)).unwrap(); crate::cas::atomic_write( &dir.path().join("atproto").join(&key).join("created_at"), b"1700000000", ) .unwrap(); Self { dir, whole } }
fn settle( &self, hosted: &Hosted, cobs: crate::atproto::CobState, now: i64, ) -> crate::atproto::SettledRepo { let chain_root = self.dir.path().join("atproto"); if self.whole { let settled = chain_root .join(crate::plan::repo_key(hosted.repo.as_str())) .join("settled.json"); let _ = std::fs::remove_file(settled); } let source = crate::atproto::RepoSource { did: hosted.repo.clone(), owner: OwnerDid::from(hosted.owner.clone()), name: RepoName::new("sample").unwrap(), export_dir: hosted.git.path().to_path_buf(), cobs: Some(cobs), served: crate::store::RepoStore::served(&self.dir.path().join("repo")), verify: false, }; crate::atproto::settle_at( &chain_root, &hosted.signer, &source, UnixSeconds::new(now), &mut crate::firehose::Framing::at(&hosted.repo, now as u64 * 1_000_000), ) .unwrap() }
fn tip(&self, hosted: &Hosted) -> String { let chain = crate::atproto::tests::open_chain(&self.dir.path().join("atproto"), &hosted.repo) .unwrap(); knot_record::chain::tip(&chain) .unwrap() .unwrap() .git .to_hex() }}
#[test]fn settling_only_the_changed_records_matches_settling_them_all() { let hosted = Hosted::new(ContributionPolicy::Anyone); let ticket = hosted.ticket(&hosted.owner, "owner's"); let label = hosted.written( &hosted.owner, Method::CreateRecord, serde_json::json!({ "repo": hosted.repo, "collection": LABEL, "rkey": "bug", "record": { "name": "bug", "createdAt": CREATED_AT } }), ); let mut live: Vec<Live> = [(TICKET, &ticket), (LABEL, &label)] .into_iter() .map(|(collection, written)| Live { collection, rkey: rkey_of(written), subject: None, }) .collect(); let changed = Settling::new(&hosted.repo, false); let whole = Settling::new(&hosted.repo, true); let lock = crate::atproto::tests::test_lock(&hosted.root, &hosted.repo); let records = crate::datastore::Datastore::at(&hosted.root); let cache = crate::store::RepoStore::open(&hosted.root.join("cob-cache")); let settled = changed .dir .path() .join("atproto") .join(crate::plan::repo_key(hosted.repo.as_str())) .join("settled.json"); let applied = crate::atproto::APPLIED.get(); let mut settles = 0; let mut settle_both = |case: &str| { settles += 1; let now = 1_700_000_000 + settles; let state = || { crate::atproto::cob_state( &hosted.index, &hosted.git, &hosted.repo, &cache, &records, &lock, ) .unwrap() }; let ours = changed.settle(&hosted, state(), now); let theirs = whole.settle(&hosted, state(), now); assert_eq!( changed.tip(&hosted), whole.tip(&hosted), "{case}: chain tips" ); assert_eq!(ours.commit, theirs.commit, "{case}"); assert_eq!(ours.entry_key, theirs.entry_key, "{case}"); let ours = files_under(changed.dir.path()); let theirs = files_under(whole.dir.path()); assert_eq!( ours.keys().collect::<Vec<_>>(), theirs.keys().collect::<Vec<_>>(), "{case}: file names" ); for (name, bytes) in &ours { if !name.ends_with("settled.json") { assert!(theirs[name] == *bytes, "{case}: {name} differs"); } } }; let mut stale: Option<Vec<u8>> = None; for seed in 0..200 { let rng = &mut crate::test_rng::Rng(seed); let actor = *rng.pick(&[ &hosted.owner, &hosted.owner, &hosted.collaborator, &hosted.stranger, ]); if let Some((method, body, subject)) = plausible_body(rng, &hosted, &live) && let Ok(Done::Written(record)) = hosted.apply(actor, method, &body) { let rkey = rkey_of(&record); if !live.iter().any(|known| known.rkey == rkey) { live.push(Live { collection: collection_of(&record), rkey, subject, }); } } // a settle that crashed before recording, or a record older than the chain match rng.below(12) { 0 => { let _ = std::fs::remove_file(&settled); } 1 => { if let Some(bytes) = &stale { crate::cas::atomic_write(&settled, bytes).unwrap(); } } _ => {} } settle_both(&format!("seed {seed}")); stale = std::fs::read(&settled) .ok() .filter(|_| rng.odds(30)) .or(stale); }
// a record that came and went after the stale record, which no op of that record removes let before = std::fs::read(&settled).unwrap(); let ticket = hosted.ticket(&hosted.owner, "passing"); settle_both("the passing ticket written"); hosted .apply( &hosted.owner, Method::DeleteRecord, &serde_json::json!({ "repo": hosted.repo, "collection": TICKET, "rkey": rkey_of(&ticket), "swapRecord": ticket.cid.to_string() }), ) .unwrap(); crate::cas::atomic_write(&settled, &before).unwrap(); settle_both("the passing ticket erased");
let applied = crate::atproto::APPLIED.get() - applied; assert!( applied * 2 > settles as usize, "only {applied} of {settles} settles took the changed records" );}
#[test]fn ledger_fuzz_keeps_outsiders_out_on_an_open_repo() { ledger_fuzz(ContributionPolicy::Anyone, 0..1_500);}
172 unmodified linesExpand all