a seal reads the chain back through the blocks of the last seal
Signed-off-by: dawn <90008@klbr.net>
123456789910111213141592 unmodified lines1081091101111101121131141151161171189 unmodified lines12812913013112813213313013113213313413513413513613713813914014114214327 unmodified lines17117217317417517617717817918018118218318418518618718818919019119219319419519619719819920020120220320420520620720820921021121221321421521621721821922022122222322422522622722822923023123223323423523623723823924024124224324455 unmodified linesuse std::collections::BTreeMap;use std::future::Future;use std::path::Path;use std::sync::Arc;
use anyhow::Context as _;use bytes::Bytes;use cid::Cid;use ipld_core::ipld::Ipld;use jacquard_repo::storage::BlockStore as _;use jacquard_repo::mst::Mst;use jacquard_repo::storage::BlockStore;use knot_git::Repo;use knot_record::blocks::GitBlocks;use knot_record::chain::{self, ChainTip};use knot_record::manifest;92 unmodified lines) -> anyhow::Result<ChainState> { let tip = chain::tip(chain_repo)?.context("chain has no tip after settling")?; let commit_bytes = chain::commit_bytes(chain_repo, &tip)?; let commit = chain::commit_cid(&commit_bytes)?; let known = prior.map(|prior| &prior.blocks); let known = prior .map(|prior| Arc::clone(&prior.blocks)) .unwrap_or_default(); let (mut blocks, collections, records) = block_on(whole_tree(chain_repo, &tip, did, known))?; blocks.insert(commit, Bytes::from(commit_bytes)); Ok(ChainState { did: did.clone(),9 unmodified linesasync fn whole_tree( repo: &Repo, tip: &ChainTip, did: &RepoDid, known: Option<&BTreeMap<Cid, Bytes>>, known: Arc<BTreeMap<Cid, Bytes>>,) -> anyhow::Result<(BTreeMap<Cid, Bytes>, Vec<String>, Vec<ServedRecord>)> { let tree = chain::load_tree(repo, tip); let store = GitBlocks::at(repo, tip.blocks); let get = async |cid: &Cid| match known.and_then(|known| known.get(cid)) { Some(block) => Ok(Some(block.clone())), None => store.get(cid).await, }; let store = Arc::new(KnownFirst { known, chain: Arc::new(GitBlocks::at(repo, tip.blocks)), }); let tree = Mst::load(Arc::clone(&store), tip.data, None); let get = async |cid: &Cid| store.get(cid).await; let mut nodes = tree.collect_node_cids().await?; nodes.push(tip.data); let leaves = tree.leaves().await?;
27 unmodified lines } Ok((blocks, collections, records))}
/// the prior seal's blocks, then the chain's. a known block is used only when it hashes to its cid#[derive(Clone)]struct KnownFirst { known: Arc<BTreeMap<Cid, Bytes>>, chain: Arc<GitBlocks>,}
impl KnownFirst { fn known(&self, cid: &Cid) -> Option<Bytes> { use sha2::{Digest as _, Sha256}; const SHA2_256: u64 = 0x12; let block = self.known.get(cid)?; let hash = cid.hash(); let named = hash.code() == SHA2_256 && hash.digest() == Sha256::digest(block).as_slice(); #[cfg(test)] if named { KNOWN_READS.set(KNOWN_READS.get() + 1); } named.then(|| block.clone()) }}
#[cfg(test)]thread_local! { pub(crate) static KNOWN_READS: std::cell::Cell<usize> = const { std::cell::Cell::new(0) };}
impl BlockStore for KnownFirst { async fn get(&self, cid: &Cid) -> jacquard_repo::error::Result<Option<Bytes>> { match self.known(cid) { Some(block) => Ok(Some(block)), None => self.chain.get(cid).await, } }
async fn put(&self, data: &[u8]) -> jacquard_repo::error::Result<Cid> { self.chain.put(data).await }
async fn has(&self, cid: &Cid) -> jacquard_repo::error::Result<bool> { Ok(self.get(cid).await?.is_some()) }
async fn put_many( &self, blocks: impl IntoIterator<Item = (Cid, Bytes)> + Send, ) -> jacquard_repo::error::Result<()> { self.chain.put_many(blocks).await }
async fn get_many(&self, cids: &[Cid]) -> jacquard_repo::error::Result<Vec<Option<Bytes>>> { let mut blocks = Vec::with_capacity(cids.len()); for cid in cids { blocks.push(self.get(cid).await?); } Ok(blocks) }
async fn apply_commit( &self, commit: jacquard_repo::repo::CommitData, ) -> jacquard_repo::error::Result<()> { self.chain.apply_commit(commit).await }}
fn record_value(bytes: &[u8]) -> anyhow::Result<serde_json::Value> { let decoded: Ipld = serde_ipld_dagcbor::from_slice(bytes)?; json_of(&decoded)}55 unmodified lines414 unmodified linesExpand all4154164174184194194204214224234 unmodified lines42842943043143243243343443543613 unmodified lines414 unmodified linesExpand all crate::xrpc::seal_repo(pool, state, prior)}
pub(crate) struct Prior { pub(crate) blocks: BTreeMap<Cid, Bytes>, pub(crate) blocks: std::sync::Arc<BTreeMap<Cid, Bytes>>, pub(crate) versions: BTreeMap<String, Vec<Version>>, pub(crate) version_rows: std::collections::HashMap<String, Vec<(String, String)>>, pub(crate) files: BTreeMap<String, u64>,}4 unmodified lines let blocks = std::fs::read(dir.join(&car)) .ok() .filter(|bytes| sealed.files.get(&car) == Some(&(bytes.len() as u64))) .and_then(|bytes| knot_record::car_blocks(&bytes).ok()) .map(|car| car.blocks) .map(|car| std::sync::Arc::new(car.blocks)) .unwrap_or_default(); let mut version_rows = std::collections::HashMap::<String, Vec<_>>::new(); for (row, file) in sealed.rows.version { let uri = row.split_once('|').map_or(row.as_str(), |(uri, _)| uri);13 unmodified lines454 unmodified linesExpand all455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531454 unmodified linesExpand all settled.dir_name() ))); }}
#[test]fn a_read_back_through_the_prior_seal_is_the_read_back_from_the_chain() { let mut fixture = fixture("sha1"); let source = CobSource::new(&fixture); let alice = account("alice"); let open = |fixture: &Fixture, from: usize, count: usize| { let charset = b"234567abcdefghijklmnopqrstuvwxyz"; let pick = |at: usize| charset[at % charset.len()] as char; let cobs = source.cobs(&fixture.signer); for n in from..from + count { let key = format!("3lubrptx5{}{}{}2", pick(n / 1024), pick(n / 32), pick(n)); cobs.open_ticket(&key, &alice, &format!("ticket {n}")); } from + count }; let prior_of = |car: &[u8], tamper: bool| { let mut blocks = knot_record::car_blocks(car).unwrap().blocks; if tamper { for block in blocks.values_mut() { *block = bytes::Bytes::from_static(b"\xa1\x61x\x01"); } } Prior { blocks: std::sync::Arc::new(blocks), versions: BTreeMap::new(), version_rows: Default::default(), files: BTreeMap::new(), } }; let mut opened = open(&fixture, 40, 80); let mut car = source.settle(&mut fixture).car(); for round in 0..3 { opened = open(&fixture, opened, 5 + round * 20); let settled = source.settle(&mut fixture); let chain_repo = open_chain(&fixture.chain_root, &source.did).unwrap(); let read = |prior: Option<&Prior>| { let state = crate::atproto::chain::read_back(&chain_repo, &source.did, BTreeMap::new(), prior) .unwrap(); let records: Vec<_> = state .records .iter() .map(|record| record.answer.clone()) .collect(); (state.commit, state.collections, records, state.car) }; let whole = read(None); assert_eq!( whole.3, settled.car(), "round {round}: the seal holds what the chain does" ); let before = crate::atproto::chain::KNOWN_READS.get(); assert!( read(Some(&prior_of(&car, false))) == whole, "round {round}: through the prior" ); let used = crate::atproto::chain::KNOWN_READS.get() - before; assert!(used > 10, "round {round}: the prior served {used} blocks"); let before = crate::atproto::chain::KNOWN_READS.get(); assert!( read(Some(&prior_of(&car, true))) == whole, "round {round}: a tampered prior" ); assert_eq!( crate::atproto::chain::KNOWN_READS.get(), before, "round {round}: a tampered block is never used" ); car = settled.car(); }}