a write's build reuses the social index the write kept current
Signed-off-by: dawn <90008@klbr.net>
22 unmodified lines23242526272829303132333435363738394041424344454647484950511 unmodified line5354555657585960616263646566676869707172737475424344454647484950515276777879804 unmodified lines858687888965909192939495264 unmodified linesExpand all22 unmodified lines
pub(crate) const COB_KIND: &str = "atproto-cob";const COBS_KIND: &str = "atproto-cobs";
/// the index a write kept current, and the cob tips it was current atpub(crate) struct SyncedIndex { pub index: Index, pub repo: RepoDid, pub tips: BTreeMap<String, String>,}
pub(crate) fn cob_tips(source: &Repo) -> anyhow::Result<BTreeMap<String, String>> { Ok(source .references()? .into_iter() .filter(|reference| reference.name.as_str().starts_with(knot_cob::COBS_PREFIX)) .map(|reference| { ( reference.name.as_str().to_string(), reference.target.to_hex(), ) }) .collect())}
/// a cob ref moving after the tips are read would cache a state the knot never had.pub fn cob_state( index: &Index, source: &Repo,1 unmodified line cache: &RepoStore, records: &Datastore, lock: &RepoLock,) -> anyhow::Result<CobState> { cob_state_synced(index, source, did, cache, records, lock, None)}
/// an index synced at the very tips read here already holds the repo's social state.pub(crate) fn cob_state_synced( index: &Index, source: &Repo, did: &RepoDid, cache: &RepoStore, records: &Datastore, lock: &RepoLock, synced: Option<&BTreeMap<String, String>>,) -> anyhow::Result<CobState> { let repo = RepoKey::of(did); anyhow::ensure!( lock.key() == repo.as_str(), "cob state of {did} read under the lock of {}", lock.key() ); let tips: BTreeMap<String, String> = source .references()? .into_iter() .filter(|reference| reference.name.as_str().starts_with(knot_cob::COBS_PREFIX)) .map(|reference| { ( reference.name.as_str().to_string(), reference.target.to_hex(), ) }) .collect(); let tips = cob_tips(source)?; let key = CobStateKey::new(&format!( "{COB_STATE_VERSION}-{}", short_hash(&serde_json::to_vec(&tips)?) ))?;4 unmodified lines record_cob_state(records, &repo, &key)?; return Ok(state); }
if synced != Some(&tips) { index.refresh_social(did)?; index.refresh_social(did)?; } index.refresh_collaborators(did)?; let cobs = CobStore::new(source); let mut state = CobState::default(); let mut cob_keys = BTreeSet::new();264 unmodified linesExpand all21 unmodified lines22232425262728293056 unmodified lines21 unmodified linespub(crate) use chain::{record_file, short_hash, write_json};pub(crate) use cob_state::Version;#[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};pub(crate) use pool::{POOLED, Pool, Prior};pub use pool::{kept_served, last_seal, served_dirs, settle};
56 unmodified lines1123451489 unmodified linesExpand all149514961497149814991500150115021503150415051506150715081509151015111512151315141515151615171518151915201521152215231524152515261527152815291530153115321533153415351536153715381539154015411542154315441545154615471548154915501551155215531554155515561557155815591560156115621563156415651566156715681569157015711572157315741575157615771578157915801581158215831584158515861587158815891590159115921593159415951596172 unmodified linesExpand alluse std::collections::BTreeSet;use std::collections::{BTreeMap, BTreeSet};
use knot_cob::{CobHome, CobStore};use knot_index::{Index, Resolved};use knot_types::{1489 unmodified linesExpand all "accepted {accepted}, refused {refused}" );}
/// what generate reads for the repo, from `index` as the write left it and from a fresh index.fn social_states( hosted: &Hosted, index: &Index, tips: &BTreeMap<String, String>,) -> [serde_json::Value; 2] { let lock = crate::atproto::tests::test_lock(&hosted.root, &hosted.repo); let records = crate::datastore::Datastore::at(&hosted.root); let state = |index: &Index, synced: Option<&BTreeMap<String, String>>| { let cache = tempfile::tempdir().unwrap(); let state = crate::atproto::cob_state_synced( index, &hosted.git, &hosted.repo, &crate::store::RepoStore::open(cache.path()), &records, &lock, synced, ) .unwrap(); crate::atproto::tests::state_json(&state) }; let fresh = Index::new(hosted.meta.path(), knot_git::Layout::new(&hosted.root)); fresh.rebuild().unwrap(); [state(index, Some(tips)), state(&fresh, None)]}
#[test]fn an_index_kept_current_by_writes_reads_the_same_cob_state_as_a_fresh_one() { 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 mut written = 0; for seed in 0..300 { let rng = &mut crate::test_rng::Rng(seed); let actor = *rng.pick(&[&hosted.owner, &hosted.collaborator, &hosted.stranger]); let Some((method, body, subject)) = plausible_body(rng, &hosted, &live) else { continue; }; // another process's write moves a tip this index never saw, so it must refresh let elsewhere = rng.odds(10); let stale = elsewhere.then(|| crate::atproto::cob_tips(&hosted.git).unwrap()); let done = if elsewhere { let other = Index::new(hosted.meta.path(), knot_git::Layout::new(&hosted.root)); other.rebuild().unwrap(); let _guard = other.guard_social(&hosted.repo).unwrap(); let ledger = Ledger { index: &other, ..hosted.ledger() }; session( &other, &BTreeSet::from([hosted.owner.clone()]), actor.clone(), hosted.repo.clone(), ) .and_then(|session| ledger.apply(&session, method, body.to_string().as_bytes())) } else { hosted.apply(actor, method, &body) }; if let Ok(Done::Written(record)) = &done { written += 1; let rkey = rkey_of(record); if !live.iter().any(|known| known.rkey == rkey) { live.push(Live { collection: collection_of(record), rkey, subject, }); } } let tips = match stale { Some(stale) => stale, None => crate::atproto::cob_tips(&hosted.git).unwrap(), }; let [synced, fresh] = social_states(&hosted, &hosted.index, &tips); assert_eq!(synced, fresh, "seed {seed}, {actor} {method:?} {body}"); } assert!(written > 100, "only {written} writes landed");}
#[test]fn ledger_fuzz_keeps_outsiders_out_on_an_open_repo() { ledger_fuzz(ContributionPolicy::Anyone, 0..1_500);}
172 unmodified linesExpand all135 unmodified linesExpand all13613713813914014114214314414514614714814915015115215315415515615715815916014816116216316416516616716876 unmodified lines2452462472482332492502512522532542552402562572582592603 unmodified lines264265266267252268269270271272257 unmodified linesExpand all530531532533518519534535536537538539540541542543544545546547197 unmodified linesExpand all135 unmodified linesExpand allpub async fn run( config: Arc<ValidatedConfig>, repo_hint: Option<&str>,) -> anyhow::Result<BuildOutcome> { run_with(config, repo_hint, None).await}
/// a write hands over the index it kept current, which saves replaying every cob of its repo againpub(crate) async fn run_with( config: Arc<ValidatedConfig>, repo_hint: Option<&str>, synced: Option<&crate::atproto::SyncedIndex>,) -> anyhow::Result<BuildOutcome> { let started = std::time::Instant::now(); let t_plan = started; let trace = Tracer::new(); let layout = trace.time("bootstrap: layout", || config.layout())?; let meta_repo = layout .bootstrap_meta(&config.knot_did) .context("bootstrap meta repo")?; let meta_path = layout.meta_path(&config.knot_did)?; let fresh; let index = match synced { Some(synced) => &synced.index, None => { let index = Index::new(meta_path, layout.clone()); fresh = Index::new(meta_path, layout.clone()); &fresh } }; trace.time("bootstrap: index rebuild", || { index.rebuild().context("rebuild registry index") })?;
76 unmodified lines if did.as_str() == hint { hint_found = true; captured.push(crate::snapshot::snapshot_repo( &layout, &index, index, registry_state.as_ref(), did, )?); } else if !carried_scope.contains(did.as_str()) { captured.push(crate::snapshot::snapshot_repo( &layout, &index, index, registry_state.as_ref(), did, )?); }3 unmodified lines "repo hint {hint} does not resolve to a hosted repo" ); captured } None => crate::snapshot::snapshot_all(&layout, &index, registry_state.as_ref())?, None => crate::snapshot::snapshot_all(&layout, index, registry_state.as_ref())?, }; trace.record("plan: snapshot capture", t_snapshots);
// a removed repo must force a rebuild here, not pass as unchanged.257 unmodified linesExpand all let key = plan::repo_key(&entry.repo_did); let lock = _source_locks.get(&key).with_context(|| { format!("the source lifetime lock of {} is not held", entry.repo_did) })?; crate::atproto::cob_state( &index, crate::atproto::cob_state_synced( index, &layout.open(&did)?, &did, &store, &crate::datastore::Datastore::at(&output_root), lock, synced .filter(|synced| synced.repo == did) .map(|synced| &synced.tips), ) }) .transpose()?; Ok(crate::atproto::RepoSource {197 unmodified linesExpand all135 unmodified linesExpand all13613713813914014014114214314498 unmodified lines135 unmodified linesExpand all Ok(output.status) }) .map_err(internal)?; if !moved.is_empty() { let changed = match crate::writes::publish_stored(config, &repo).await { let changed = match crate::writes::publish_stored(config, &repo, None).await { Ok(changed) => changed, Err(error) => { crate::spindle::defer(config, &repo, &actor, &moved).await; return Err(error);98 unmodified lines348 unmodified linesExpand all3493503513523533533543553563571 unmodified line348 unmodified linesExpand all .map_err(internal)}
async fn republish(config: &Arc<ValidatedConfig>, repo: &RepoDid) -> Result<(), WriteError> { let changed = writes::publish_stored(config, repo).await?; let changed = writes::publish_stored(config, repo, None).await?; if !changed.is_empty() { crate::crawl::announce(config, &changed).await; } Ok(())1 unmodified line5 unmodified lines6789101112131464 unmodified lines798081828283848586878888899091929338 unmodified lines13213313413513513613713813914015 unmodified lines1561571581591591601611621631647 unmodified lines1721731741751761771761781791801811821831841851861871881891901911921931831841851861941951961971981992002016 unmodified lines20820921021120121221321421521621721821922022122222322437 unmodified lines5 unmodified linesuse knot_git::{Layout, Repo};use knot_index::{Index, Resolved};use knot_types::{AccountDid, RepoDid};
use crate::atproto::SyncedIndex;use crate::blobs::Store;use crate::cob_write::{Done, Ledger, Method, Session, WriteError};use crate::config::ValidatedConfig;use crate::repolock::{self, LockMode};64 unmodified lines repo: &RepoDid, actor: Actor, write: impl FnOnce(&Ledger<'_>, &Session) -> Result<Done, WriteError>,) -> Result<Published, WriteError> { let done = write_locked(config, repo, actor, write)?; let (done, synced) = write_locked(config, repo, actor, write)?; if matches!(done, Done::Granted) { if let Err(error) = crate::git_ssh::spawn_detached(config, ["keys", "sync", "--new"]) { tracing::warn!("couldn't refresh the ssh keys after a grant: {error:#}"); } } let changed = publish_stored(config, repo).await?; let changed = publish_stored(config, repo, Some(&synced)).await?; let (commit, rev) = crate::atproto::chain_head(&config.output_root.join("atproto"), repo).map_err(internal)?; Ok(Published { done,38 unmodified lines config: &ValidatedConfig, repo: &RepoDid, actor: Actor, write: impl FnOnce(&Ledger<'_>, &Session) -> Result<Done, WriteError>,) -> Result<Done, WriteError> {) -> Result<(Done, SyncedIndex), WriteError> { let (layout, meta_path, index) = rebuilt_index(config)?; let _lock = repolock::acquire( &config.output_root, &crate::plan::repo_key(repo.as_str()),15 unmodified lines }, }; let now = crate::atproto::unix_now(); let session = crate::cob_write::session(&index, &admins(config), actor, repo.clone())?; let _guard = index.guard_social(repo)?; let guard = index.guard_social(repo)?; let home = CobHome::from(&config.knot_did); let chain_root = config.output_root.join("atproto"); let ledger = Ledger { index: &index,7 unmodified lines cited: Default::default(), now, }; let done = write(&ledger, &session)?; let tips = crate::atproto::cob_tips(&git).map_err(internal)?; crate::commit_graph::refresh(git.path()); Ok(done) drop((ledger, guard)); let synced = SyncedIndex { index, repo: repo.clone(), tips, }; Ok((done, synced))}
pub(crate) async fn publish_stored( config: &Arc<ValidatedConfig>, repo: &RepoDid, synced: Option<&SyncedIndex>,) -> Result<Vec<String>, WriteError> { publish_scoped_with(config, Some(repo.as_str()), synced) .await publish(config, repo).await.map_err(|error| { tracing::error!(%repo, "stored changes weren't published: {error:#}"); WriteError::internal_notice(UNPUBLISHED) }) .map_err(|error| { tracing::error!(%repo, "stored changes weren't published: {error:#}"); WriteError::internal_notice(UNPUBLISHED) })}
pub(crate) async fn publish( config: &Arc<ValidatedConfig>,6 unmodified linespub(crate) async fn publish_scoped( config: &Arc<ValidatedConfig>, repo_hint: Option<&str>,) -> anyhow::Result<Vec<String>> { match crate::generate::run(config.clone(), repo_hint).await? { publish_scoped_with(config, repo_hint, None).await}
async fn publish_scoped_with( config: &Arc<ValidatedConfig>, repo_hint: Option<&str>, synced: Option<&SyncedIndex>,) -> anyhow::Result<Vec<String>> { match crate::generate::run_with(config.clone(), repo_hint, synced).await? { crate::generate::BuildOutcome::Unchanged { .. } => Ok(Vec::new()), crate::generate::BuildOutcome::Built { generation_id, pin: _pin,37 unmodified lines