chadtmiller.com/knotless
git that lives in your Atmosphere account
1
10
A private repository's space syncs past a record that cannot be read
crates/knotless/src/space_sync.rs+116−38
| @@ -237,10 +237,13 @@ pub struct ReadOn { | ||
/// `theirs`'s writes since `since`, every page of them, `hash` (its set hash | ||
/// at `since`) moved by each and checked against the commit they come to, | ||
/// which their server vouches for. `kem` is the member's `#atproto_kem` key. | ||
/// With no `since`, the oplog is read from its start, and `hash` is the | ||
/// empty set's: it comes to the commit only if the oplog still holds every | ||
/// write the repo was made of. | ||
pub async fn read_on( | ||
theirs: &Remote, | ||
kem: &PublicKey, | ||
since: &str, | ||
since: Option<&str>, | ||
mut hash: SetHash, | ||
) -> Result<ReadOn> { | ||
#[derive(Deserialize)] | ||
| @@ -255,10 +258,8 @@ pub async fn read_on( | ||
while last.is_none() || cursor.is_some() { | ||
// From where the last page left off, after the first. | ||
let from = std::iter::repeat("cursor").zip(cursor.as_deref()); | ||
let params: Vec<_> = [("since", since), ("limit", PAGE)] | ||
.into_iter() | ||
.chain(from) | ||
.collect(); | ||
let since = std::iter::repeat("since").zip(since); | ||
let params: Vec<_> = since.chain([("limit", PAGE)]).chain(from).collect(); | ||
let (agreed, mac, body) = vouched(theirs, kem, "listRepoOps", ¶ms).await?; | ||
let page: Page = serde_json::from_slice(&body)?; | ||
page.ops.iter().for_each(|op| hash.apply(op)); | ||
| @@ -288,20 +289,34 @@ pub struct Change { | ||
pub value: Option<Value>, | ||
} | ||
| ||
/// A record a repo holds that cannot be read, and why: one whose value is | ||
/// not DRISL, or does not come to the CID it was written with. It is left | ||
/// out of what is read, and the rest read all the same. | ||
#[derive(Debug, Clone, PartialEq)] | ||
pub struct Unreadable { | ||
pub collection: String, | ||
pub rkey: String, | ||
pub why: String, | ||
} | ||
| ||
impl ReadOn { | ||
/// Each record of a collection `wanted` that the writes read touched, | ||
/// as the last of them left it. An oplog gives a record's value with | ||
/// the last write to it, as it is when the page is read, so each is | ||
/// checked against that write's CID. One with no value, or another, | ||
/// was written again meanwhile, and does not check out: the repo is | ||
/// read whole instead. | ||
pub fn changes(&self, wanted: impl Fn(&str) -> bool) -> Result<Vec<Change>> { | ||
/// as the last of them left it, and those of them that cannot be read. | ||
/// An oplog gives a record's value with the last write to it, as it is | ||
/// when the page is read, so each is checked against that write's CID, | ||
/// which the commit vouches for. One that does not come to it is | ||
/// unreadable, and given with no value, as though deleted, so that | ||
/// nothing unvouched for is kept of it. One with no value was written | ||
/// again meanwhile, and does not check out: the repo is read whole | ||
/// instead. | ||
pub fn changes(&self, wanted: impl Fn(&str) -> bool) -> Result<(Vec<Change>, Vec<Unreadable>)> { | ||
let mut last = std::collections::BTreeMap::new(); | ||
for op in self.ops.iter().filter(|op| wanted(&op.collection)) { | ||
last.insert((op.collection.as_str(), op.rkey.as_str()), op); | ||
} | ||
let mut changes = Vec::with_capacity(last.len()); | ||
let (mut changes, mut unreadable) = (Vec::with_capacity(last.len()), Vec::new()); | ||
for ((collection, rkey), op) in last { | ||
let (collection, rkey) = (collection.to_string(), rkey.to_string()); | ||
let value = match &op.cid { | ||
None => None, | ||
Some(cid) => { | ||
| @@ -309,16 +324,24 @@ impl ReadOn { | ||
.value | ||
.as_ref() | ||
.context("a record written with no value")?; | ||
Some(checked(value, cid)?) | ||
let checked = checked(value, cid); | ||
if let Err(why) = &checked { | ||
unreadable.push(Unreadable { | ||
collection: collection.clone(), | ||
rkey: rkey.clone(), | ||
why: format!("{why:#}"), | ||
}); | ||
} | ||
checked.ok() | ||
} | ||
}; | ||
changes.push(Change { | ||
collection: collection.to_string(), | ||
rkey: rkey.to_string(), | ||
collection, | ||
rkey, | ||
value, | ||
}); | ||
} | ||
Ok(changes) | ||
Ok((changes, unreadable)) | ||
} | ||
} | ||
| ||
| @@ -341,9 +364,11 @@ pub struct Record { | ||
pub value: Value, | ||
} | ||
| ||
/// A member's repo read whole. | ||
/// A member's repo read whole: the records it holds that can be read, and | ||
/// those that cannot, which count towards its set hash all the same. | ||
pub struct Whole { | ||
pub records: Vec<Record>, | ||
pub unreadable: Vec<Unreadable>, | ||
pub commit: Commit, | ||
pub hash: SetHash, | ||
} | ||
| @@ -389,7 +414,9 @@ async fn vouched( | ||
} | ||
| ||
/// Reads a repo's STAR archive: its commit, then each record by path, in | ||
/// order, hashed as it is read. | ||
/// order, hashed as it is read. A record that is not a DRISL map is hashed | ||
/// as its bytes are, and set aside as unreadable: whether the archive | ||
/// comes to its commit is for the caller to check, with it. | ||
fn star(bytes: &[u8]) -> Result<Whole> { | ||
let rest = bytes | ||
.strip_prefix(&STAR_MAGIC[..]) | ||
| @@ -399,26 +426,34 @@ fn star(bytes: &[u8]) -> Result<Whole> { | ||
let meta = drisl::record(meta).context("STAR metadata that is not DRISL")?; | ||
ensure!(meta["$type"] == COMMIT_MAC, "not a repo's STAR archive"); | ||
let commit: Commit = serde_json::from_value(meta)?; | ||
let (mut records, mut hash, mut last) = (Vec::new(), SetHash::default(), None); | ||
let (mut records, mut unreadable) = (Vec::new(), Vec::new()); | ||
let (mut hash, mut last) = (SetHash::default(), None); | ||
while let Some(key) = star.next()? { | ||
let value = star.next()?.context("a STAR key with no value")?; | ||
ensure!(last < Some(key), "a STAR archive's records out of order"); | ||
last = Some(key); | ||
let path = std::str::from_utf8(key)?; | ||
let (collection, rkey) = path.split_once('/').context("a record path with no key")?; | ||
let record = drisl::record(value).filter(Value::is_object); | ||
let record = record.context("a record that is not a DRISL map")?; | ||
let cid = cid(value); | ||
hash.add(collection, rkey, &cid); | ||
records.push(Record { | ||
collection: collection.to_string(), | ||
rkey: rkey.to_string(), | ||
cid, | ||
value: record, | ||
}); | ||
let (collection, rkey) = (collection.to_string(), rkey.to_string()); | ||
match drisl::record(value).filter(Value::is_object) { | ||
Some(value) => records.push(Record { | ||
collection, | ||
rkey, | ||
cid, | ||
value, | ||
}), | ||
None => unreadable.push(Unreadable { | ||
collection, | ||
rkey, | ||
why: "a record that is not a DRISL map".to_string(), | ||
}), | ||
} | ||
} | ||
Ok(Whole { | ||
records, | ||
unreadable, | ||
commit, | ||
hash, | ||
}) | ||
| @@ -535,8 +570,9 @@ mod tests { | ||
op("a.b.c", "2", None, None), | ||
op("x.y.z", "1", Some("bafy".into()), None), | ||
]; | ||
let changes = read(ops).changes(wanted).unwrap(); | ||
let (changes, unreadable) = read(ops).changes(wanted).unwrap(); | ||
assert_eq!(changes, [change("1", Some(&two)), change("2", None)]); | ||
assert!(unreadable.is_empty()); | ||
| ||
let fails = |ops| read(ops).changes(wanted).err().map(|e| e.to_string()); | ||
let last = op("a.b.c", "1", Some(cid_of(&one)), None); | ||
| @@ -544,13 +580,33 @@ mod tests { | ||
fails(vec![last]).as_deref(), | ||
Some("a record written with no value") | ||
); | ||
let other = op("a.b.c", "1", Some(cid_of(&one)), Some(&two)); | ||
let said = Some("a record that does not come to its CID"); | ||
assert_eq!(fails(vec![other]).as_deref(), said); | ||
// A record whose value does not come to the CID its write gave | ||
// it, as one its server keeps cut short, or that is not DRISL, is | ||
// left out as though deleted, said to be unreadable, and the rest | ||
// read all the same. | ||
let unreadable = |rkey: &str, why: &str| Unreadable { | ||
collection: "a.b.c".into(), | ||
rkey: rkey.into(), | ||
why: why.into(), | ||
}; | ||
let float = serde_json::json!({"n": 1.5}); | ||
let unencodable = op("a.b.c", "1", Some(cid_of(&one)), Some(&float)); | ||
let said = Some("a record that is not DRISL"); | ||
assert_eq!(fails(vec![unencodable]).as_deref(), said); | ||
let ops = vec![ | ||
op("a.b.c", "1", Some(cid_of(&one)), Some(&two)), | ||
op("a.b.c", "2", Some(cid_of(&one)), Some(&float)), | ||
op("a.b.c", "3", Some(cid_of(&one)), Some(&one)), | ||
]; | ||
let (changes, skipped) = read(ops).changes(wanted).unwrap(); | ||
let left = [ | ||
change("1", None), | ||
change("2", None), | ||
change("3", Some(&one)), | ||
]; | ||
assert_eq!(changes, left); | ||
let said = [ | ||
unreadable("1", "a record that does not come to its CID"), | ||
unreadable("2", "a record that is not DRISL"), | ||
]; | ||
assert_eq!(skipped, said); | ||
} | ||
| ||
#[test] | ||
| @@ -741,10 +797,6 @@ mod tests { | ||
archive(&good, &[("nokey", one.clone())]), | ||
"a record path with no key", | ||
), | ||
( | ||
archive(&good, &[("a.b.c/1", cbor(serde_json::json!([1])))]), | ||
"a record that is not a DRISL map", | ||
), | ||
( | ||
[archive(&good, &[]), vec![5, 1]].concat(), | ||
"a STAR archive ends inside a frame", | ||
| @@ -768,5 +820,31 @@ mod tests { | ||
let unreadable = archive(&good, &[]); | ||
let unreadable = [&unreadable[..], &[1, 0xff, 1, 0xa0]].concat(); | ||
assert!(star(&unreadable).is_err()); | ||
| ||
// A record that is not a DRISL map, as one cut short, or not a map | ||
// at all, is set aside, hashed as its bytes are, and the rest read. | ||
let cut = cbor(serde_json::json!({"n": 5_000_000_000u64})); | ||
let cut = cut[..cut.len() - 4].to_vec(); | ||
let list = cbor(serde_json::json!([1])); | ||
let pairs = [ | ||
("a.b.c/1", cut.clone()), | ||
("a.b.c/2", list.clone()), | ||
("a.b.c/3", one.clone()), | ||
]; | ||
let read = star(&archive(&good, &pairs)).unwrap(); | ||
assert_eq!(read.records.len(), 1); | ||
assert_eq!(read.records[0].rkey, "3"); | ||
let why = |rkey: &str| Unreadable { | ||
collection: "a.b.c".into(), | ||
rkey: rkey.into(), | ||
why: "a record that is not a DRISL map".into(), | ||
}; | ||
assert_eq!(read.unreadable, [why("1"), why("2")]); | ||
let mut hash = SetHash::default(); | ||
for (key, bytes) in &pairs { | ||
let (collection, rkey) = key.split_once('/').unwrap(); | ||
hash.add(collection, rkey, &cid(bytes)); | ||
} | ||
assert_eq!(read.hash, hash); | ||
} | ||
} | ||
crates/knotless/tests/common/mod.rs+3−0
| @@ -26,6 +26,9 @@ pub struct Net { | ||
/// Whether the PDS is on the permissioned spaces alpha. | ||
#[serde(default)] | ||
pub spaces: bool, | ||
/// Where the PDS keeps each account's store. | ||
#[serde(default)] | ||
pub actors: Option<String>, | ||
pub gateway: GatewayTls, | ||
pub accounts: Vec<Account>, | ||
} | ||
crates/knotless-serve/src/spaces.rs+59−8
| @@ -402,7 +402,9 @@ pub async fn sync_with(ctx: &Ctx, index: &Index, joined: &Space) -> Result<bool> | ||
} | ||
| ||
/// Takes in `did`'s repo in `joined`: read on from where it was last | ||
/// synced to, or whole. | ||
/// synced to, or whole, or, should it not come to its commit whole, from | ||
/// the start of its oplog, where each record's CID is vouched for apart | ||
/// from its bytes. | ||
async fn member(ctx: &Ctx, index: &Index, joined: &Space, did: &str) -> Result<()> { | ||
let theirs = remote(ctx, joined, did).await?; | ||
let kem = identity::kem_key(&identity::document(ctx, did).await?)?; | ||
| @@ -416,7 +418,11 @@ async fn member(ctx: &Ctx, index: &Index, joined: &Space, did: &str) -> Result<( | ||
None => false, | ||
}; | ||
if !read { | ||
whole(index, uri, &theirs, &kem).await?; | ||
let read = whole(index, uri, &theirs, &kem).await; | ||
crate::telemetry::read_from_start(uri, did, &read); | ||
if read.is_err() { | ||
from_start(index, uri, &theirs, &kem).await?; | ||
} | ||
} | ||
Ok(()) | ||
} | ||
| @@ -452,9 +458,11 @@ async fn read_on( | ||
state: &[u8], | ||
) -> Result<()> { | ||
let hash = SetHash::from_state(state)?; | ||
let read = space_sync::read_on(theirs, kem, rev, hash).await?; | ||
let read = space_sync::read_on(theirs, kem, Some(rev), hash).await?; | ||
let did = theirs.did(); | ||
for change in read.changes(indexed)? { | ||
let (changes, unreadable) = read.changes(indexed)?; | ||
crate::telemetry::unreadable(uri, did, &unreadable); | ||
for change in changes { | ||
let (collection, rkey) = (&change.collection, &change.rkey); | ||
match &change.value { | ||
Some(value) => index.put_from_space(uri, did, collection, rkey, value)?, | ||
| @@ -470,15 +478,58 @@ async fn read_on( | ||
async fn whole(index: &Index, uri: &str, theirs: &Remote, kem: &PublicKey) -> Result<()> { | ||
let whole = space_sync::whole(theirs, kem).await?; | ||
let did = theirs.did(); | ||
crate::telemetry::unreadable(uri, did, &whole.unreadable); | ||
let records = whole.records.iter().filter(|r| indexed(&r.collection)); | ||
let records = records.map(|r| (&r.collection, &r.rkey, &r.value)); | ||
let commit = (whole.commit.rev.as_str(), &whole.hash); | ||
replace(index, uri, did, records, commit) | ||
} | ||
| ||
/// Reads `theirs` from the start of its oplog, in place of all it gave | ||
/// before, as [`whole`] does: for a repo whose archive does not come to | ||
/// its commit, as one holding a record its server keeps cut short, whose | ||
/// bytes no longer come to the CID it was written with. The oplog gives | ||
/// each record's CID apart from its bytes, which the commit vouches for | ||
/// all together, and each record's value is checked against its own: | ||
/// one that does not come to it is left out. It comes to the commit only | ||
/// if the oplog still holds every write the repo was made of. | ||
async fn from_start(index: &Index, uri: &str, theirs: &Remote, kem: &PublicKey) -> Result<()> { | ||
let read = space_sync::read_on(theirs, kem, None, SetHash::default()).await?; | ||
let did = theirs.did(); | ||
let (changes, unreadable) = read.changes(indexed)?; | ||
crate::telemetry::unreadable(uri, did, &unreadable); | ||
let records = changes.iter().filter_map(|c| { | ||
let value = c.value.as_ref()?; | ||
Some((&c.collection, &c.rkey, value)) | ||
}); | ||
replace( | ||
index, | ||
uri, | ||
did, | ||
records, | ||
(read.commit.rev.as_str(), &read.hash), | ||
) | ||
} | ||
| ||
/// Puts `records` in the index as all `did`'s repo in the space `uri` | ||
/// holds, at the revision and set hash its commit came to: what it gave | ||
/// before and no longer has is taken out. | ||
fn replace<'a>( | ||
index: &Index, | ||
uri: &str, | ||
did: &str, | ||
records: impl Iterator<Item = (&'a String, &'a String, &'a Value)>, | ||
(rev, hash): (&str, &SetHash), | ||
) -> Result<()> { | ||
let mut gone: HashSet<(String, String)> = index.space_records(uri, did)?.into_iter().collect(); | ||
for r in whole.records.iter().filter(|r| indexed(&r.collection)) { | ||
index.put_from_space(uri, did, &r.collection, &r.rkey, &r.value)?; | ||
gone.remove(&(r.collection.clone(), r.rkey.clone())); | ||
for (collection, rkey, value) in records { | ||
index.put_from_space(uri, did, collection, rkey, value)?; | ||
gone.remove(&(collection.clone(), rkey.clone())); | ||
} | ||
for (collection, rkey) in gone { | ||
index.delete_from_space(uri, did, &collection, &rkey)?; | ||
} | ||
index.set_space_writer(uri, did, &whole.commit.rev, &whole.hash.state())?; | ||
index.set_space_writer(uri, did, rev, &hash.state())?; | ||
Ok(()) | ||
} | ||
| ||
crates/knotless-serve/src/telemetry.rs+42−0
| @@ -357,6 +357,37 @@ pub fn read_whole(uri: &str, did: &str, read: &anyhow::Result<()>) { | ||
} | ||
} | ||
| ||
/// Logs, at warn, why `did`'s repo in the space `uri` could not be read | ||
/// whole, if it could not: it is read from the start of its oplog instead. | ||
pub fn read_from_start(uri: &str, did: &str, read: &anyhow::Result<()>) { | ||
if let Err(err) = read { | ||
let fields = Fields { | ||
source: Some(uri), | ||
did: Some(did), | ||
error: Some(format!("{err:#}")), | ||
..Fields::default() | ||
}; | ||
telemetry::note(Level::WARN, "space repo read from its oplog", &fields); | ||
} | ||
} | ||
| ||
/// Logs, at warn, each record of `did`'s repo in the space `uri` that | ||
/// could not be read, by its path, and why: each is left out of the | ||
/// index, and the rest of the repo read all the same. | ||
pub fn unreadable(uri: &str, did: &str, records: &[knotless::space_sync::Unreadable]) { | ||
for record in records { | ||
let path = format!("{}/{}", record.collection, record.rkey); | ||
let fields = Fields { | ||
source: Some(uri), | ||
did: Some(did), | ||
key: Some(&path), | ||
error: Some(record.why.clone()), | ||
..Fields::default() | ||
}; | ||
telemetry::note(Level::WARN, "space record unreadable", &fields); | ||
} | ||
} | ||
| ||
/// Logs, at warn, why following the labeler `did` stopped, if it did | ||
/// for a reason: it is followed again at once. | ||
pub fn labeler<T>(did: &str, done: anyhow::Result<T>) { | ||
| @@ -476,6 +507,17 @@ mod tests { | ||
"did:plc:y", | ||
&Err(anyhow::anyhow!("no match")), | ||
); | ||
read_from_start( | ||
"at://did:plc:x/space/t/k", | ||
"did:plc:y", | ||
&Err(anyhow::anyhow!("no match")), | ||
); | ||
let cut = knotless::space_sync::Unreadable { | ||
collection: "a.b.c".into(), | ||
rkey: "1".into(), | ||
why: "a record that does not come to its CID".into(), | ||
}; | ||
unreadable("at://did:plc:x/space/t/k", "did:plc:y", &[cut]); | ||
opened(None, Duration::ZERO); | ||
let res = failed(StatusCode::BAD_GATEWAY, "down".into()); | ||
assert_eq!(res.extensions().get::<Cause>().unwrap().0, "down"); | ||
crates/knotless-serve/tests/gateway.rs+61−0
| @@ -3019,6 +3019,32 @@ fn private_repositories_on_the_pages() { | ||
wait_until(|| matches!(reads(), 404 | 502)); | ||
} | ||
| ||
/// Spoils `did`'s record at `collection`/`rkey` in `space` where the PDS | ||
/// keeps it, as a fault of a server's might, leaving its CID as it was: | ||
/// its bytes are another record's, which no longer come to it. | ||
fn cut_short(did: &str, space: &str, collection: &str, rkey: &str) { | ||
use sha2::Digest; | ||
let actors = net() | ||
.actors | ||
.as_ref() | ||
.expect("a harness that names its store"); | ||
let hash: String = sha2::Sha256::digest(did.as_bytes()) | ||
.iter() | ||
.map(|b| format!("{b:02x}")) | ||
.collect(); | ||
let store = std::path::Path::new(actors).join(&hash[..2]).join(did); | ||
let db = rusqlite::Connection::open(store.join("store.sqlite")).unwrap(); | ||
db.busy_timeout(Duration::from_secs(10)).unwrap(); | ||
let other = knotless::drisl::encode(&json!({"$type": collection, "cut": 1})).unwrap(); | ||
let spoiled = db | ||
.execute( | ||
"UPDATE space_record SET value = ?1 WHERE space = ?2 AND collection = ?3 AND rkey = ?4", | ||
rusqlite::params![other, space, collection, rkey], | ||
) | ||
.unwrap(); | ||
assert_eq!(spoiled, 1); | ||
} | ||
| ||
/// The reference PDS's administrator, as dev-env makes it, for basic auth. | ||
fn admin() -> String { | ||
use base64::Engine; | ||
| @@ -3165,6 +3191,41 @@ fn private_repositories_indexed_here() { | ||
run(&owner, &["delete-comment", &url, &said]); | ||
wait_until(|| comments() == 0); | ||
assert_eq!(read("getRepo", "")["openIssues"], 2); | ||
// A record the member's server keeps cut short, so that it no longer | ||
// comes to the CID it was written with, is left out, and the rest of | ||
// their repo read all the same. Read whole, the repo does not come to | ||
// its commit, so it is read from the start of its oplog instead, which | ||
// gives each record's CID apart from its bytes. | ||
let theirs_did = member.account.did.clone(); | ||
let checks = || { | ||
let records = index.space_records(&space, &theirs_did).unwrap(); | ||
let checks = records | ||
.into_iter() | ||
.filter(|(c, _)| c == "dev.knotless.check"); | ||
let mut rkeys: Vec<String> = checks.map(|(_, rkey)| rkey).collect(); | ||
rkeys.sort(); | ||
rkeys | ||
}; | ||
let file = |rkey: &str| { | ||
let check = json!({ | ||
"$type": "dev.knotless.check", "repo": reference, "ref": "refs/heads/main", | ||
"commit": owner.git(&work, &["rev-parse", "HEAD"]), "workflow": rkey, | ||
"status": "success", "startedAt": "2026-10-24T00:00:00.000Z", | ||
}); | ||
member.put_space_record(&space, "dev.knotless.check", rkey, check); | ||
}; | ||
file("p2"); | ||
wait_until(|| checks() == ["p1", "p2"]); | ||
std::thread::sleep(Duration::from_secs(1)); | ||
cut_short(&theirs_did, &space, "dev.knotless.check", "p2"); | ||
let (rev, _) = index.space_writer(&space, &theirs_did).unwrap().unwrap(); | ||
index | ||
.set_space_writer(&space, &theirs_did, &rev, &wrong) | ||
.unwrap(); | ||
file("p3"); | ||
wait_until(|| checks() == ["p1", "p3"]); | ||
let p3 = format!("&uri=at://{theirs_did}/dev.knotless.check/p3"); | ||
assert_eq!(read("getCheck", &p3)["check"]["workflow"], "p3"); | ||
| ||
// A member's reaction is in the space, and counted from here at once. | ||
let react = format!("{}/xrpc/dev.knotless.addReaction", server.base); | ||
docs/design/access.md+20−2
| @@ -403,14 +403,32 @@ checks what it reads: | ||
must be vouched for. Each record the writes touched is taken as the | ||
last of them left it: one written, with the value the oplog gives it, | ||
which must encode, as DRISL, to that write's CID; one deleted (an op | ||
with no CID) taken out. A last write with no value, or another, was | ||
written again as the page was read, and does not check out. | ||
with no CID) taken out. A last write with no value was written again | ||
as the page was read, and does not check out. | ||
- A repo never synced, or whose writes do not check out (its oplog no | ||
longer reaching back that far, say), is read whole with | ||
`com.atproto.space.getRepo`: a STAR archive, its commit then each | ||
record's DRISL encoding by path, in order, hashed afresh and checked the | ||
same way. Its records replace those it gave the index before, so one | ||
deleted meanwhile goes too. | ||
- A repo whose archive does not come to its commit is read from the | ||
start of its oplog instead, `listRepoOps` with no `since`, from the | ||
empty set's hash, and its records replace those it gave before as | ||
the archive's would. The archive gives each record's bytes, its CID | ||
only as their hash, so a record whose server keeps its bytes spoiled | ||
(cut short, say) spoils the whole archive's hash; the oplog gives each | ||
CID apart from the bytes, all of them vouched for together by the | ||
commit. This comes to the commit only if the oplog still holds every | ||
write the repo was made of; otherwise the repo is not read. | ||
| ||
A record that cannot be read, one whose value does not encode to the | ||
CID its write gave it, or that is not a DRISL map, is left out of the | ||
index (taken out, if an earlier read put it there), logged at warn as | ||
`space record unreadable` with the repo, its path and why, and the rest | ||
of the repo read all the same. Nothing is kept of it that its CID does | ||
not vouch for: the set hash is checked over the CIDs the commit vouches | ||
for, whatever the bytes, so leaving one out loosens nothing about the | ||
rest. | ||
| ||
The set hash and revision are kept per repo with the records, and a repo | ||
is moved on only once what was read checks out. A repo the listing names | ||
harness/pds.mjs+3−0
| @@ -159,6 +159,9 @@ console.log( | ||
pds: net.pds.url, | ||
plc: net.plc.url, | ||
spaces: described.spacesAlpha === true, | ||
// Where the PDS keeps each account's store, for a test to spoil a | ||
// record in it as a server's fault might. | ||
actors: net.pds.ctx.cfg.actorStore.directory, | ||
gateway: { host: GATEWAY_HOST, cert, key }, | ||
accounts, | ||
}), | ||