diff --git a/bobbin/crates/xrpc/src/profile_activity.rs b/bobbin/crates/xrpc/src/profile_activity.rs index 80e34174b..263467910 100644 --- a/bobbin/crates/xrpc/src/profile_activity.rs +++ b/bobbin/crates/xrpc/src/profile_activity.rs @@ -4,12 +4,12 @@ use axum::{extract::State, response::Json}; use bobbin_edge_index::{EdgeItem, EdgeStore, PageCursor, PageLimit, SortDir}; use bobbin_types::edges::{TICKET_PULL_KIND, ticket_kind}; use bobbin_types::ids::{EdgeKey, SubjectRef, nsid_static, owner_did_from_aturi}; -use bobbin_types::org_tangled::repo::manifest::{Manifest, ManifestRecord}; -use bobbin_types::org_tangled::track::ticket::{Ticket, TicketRecord}; use bobbin_types::org_tangled::actor::get_profile_activity::{ ActivityMonth, GetProfileActivityOutput, IssueActivity, PullActivity, RepoActivity, RepoActivityKind, RepoRef, }; +use bobbin_types::org_tangled::repo::manifest::{Manifest, ManifestRecord}; +use bobbin_types::org_tangled::track::ticket::{Ticket, TicketRecord}; use chrono::{DateTime, TimeZone, Utc}; use futures::{StreamExt, stream}; use jacquard_common::DefaultStr; diff --git a/bobbin/crates/xrpc/tests/coverage.rs b/bobbin/crates/xrpc/tests/coverage.rs index 9e01830a3..d91e6695b 100644 --- a/bobbin/crates/xrpc/tests/coverage.rs +++ b/bobbin/crates/xrpc/tests/coverage.rs @@ -319,7 +319,10 @@ async fn any_origin_allows_preflight_and_get() { } fn ready_request() -> Request { - Request::builder().uri("/_ready").body(Body::empty()).unwrap() + Request::builder() + .uri("/_ready") + .body(Body::empty()) + .unwrap() } #[tokio::test] diff --git a/crates/gix-chunk/src/file/decode.rs b/crates/gix-chunk/src/file/decode.rs index 498a66872..1ad3f952b 100644 --- a/crates/gix-chunk/src/file/decode.rs +++ b/crates/gix-chunk/src/file/decode.rs @@ -7,7 +7,11 @@ use crate::{file, file::index}; impl file::Index { /// Provided a mapped file at the beginning via `data`, starting at `toc_offset` decode all chunk information to return /// an index with `num_chunks` chunks. - pub fn from_bytes(data: &[u8], toc_offset: usize, num_chunks: u32) -> Result { + pub fn from_bytes( + data: &[u8], + toc_offset: usize, + num_chunks: u32, + ) -> Result { if num_chunks == 0 { return Err(ValidationError::new( "Empty chunk indices are not allowed as the point of chunked files is to have chunks.", @@ -67,7 +71,10 @@ impl file::Index { let sentinel = to_kind(&toc_entry[..4]); if sentinel != crate::SENTINEL { - Err(format!("Sentinel value wasn't found, saw '{}'", sentinel.as_bstr()))?; + Err(format!( + "Sentinel value wasn't found, saw '{}'", + sentinel.as_bstr() + ))?; } Ok(file::Index { diff --git a/crates/gix-chunk/src/file/index.rs b/crates/gix-chunk/src/file/index.rs index f2ead63d1..1f017fb95 100644 --- a/crates/gix-chunk/src/file/index.rs +++ b/crates/gix-chunk/src/file/index.rs @@ -42,7 +42,9 @@ impl Index { pub fn usize_offset_by_id(&self, kind: Id) -> Result, Message> { self.chunks .iter() - .find_map(|c| (c.kind == kind).then(|| crate::range::into_usize_or_panic(c.offset.clone()))) + .find_map(|c| { + (c.kind == kind).then(|| crate::range::into_usize_or_panic(c.offset.clone())) + }) .ok_or_else(make_message(kind)) } @@ -54,7 +56,9 @@ impl Index { ) -> Result { self.chunks .iter() - .find_map(|c| (c.kind == kind).then(|| crate::range::into_usize_or_panic(c.offset.clone()))) + .find_map(|c| { + (c.kind == kind).then(|| crate::range::into_usize_or_panic(c.offset.clone())) + }) .map(validate) .ok_or_else(make_message(kind)) } @@ -62,8 +66,9 @@ impl Index { /// Find a chunk of `kind` and return its data slice based on its offset. pub fn data_by_id<'a>(&self, data: &'a [u8], kind: Id) -> Result<&'a [u8], Message> { let offset = self.offset_by_id(kind)?; - Ok(&data[crate::range::into_usize(offset) - .ok_or_else(|| message("The offsets into the file couldn't be represented by usize"))?]) + Ok(&data[crate::range::into_usize(offset).ok_or_else(|| { + message("The offsets into the file couldn't be represented by usize") + })?]) } /// Return the end offset of the last chunk, which is the highest offset as well. @@ -74,5 +79,10 @@ impl Index { } fn make_message(kind: Id) -> impl FnOnce() -> Message { - move || message!("Chunk named '{}' was not found in chunk file index", kind.as_bstr()) + move || { + message!( + "Chunk named '{}' was not found in chunk file index", + kind.as_bstr() + ) + } } diff --git a/crates/gix-chunk/src/file/write.rs b/crates/gix-chunk/src/file/write.rs index 42dbac0dc..ccbbf5621 100644 --- a/crates/gix-chunk/src/file/write.rs +++ b/crates/gix-chunk/src/file/write.rs @@ -80,7 +80,10 @@ impl Index { } /// Plan to write a new chunk as part of the index when [`into_write()`][Index::into_write()] is called. pub fn plan_chunk(&mut self, chunk: crate::Id, exact_size_on_disk: u64) { - assert!(self.will_write, "BUG: create the index with `for_writing()`"); + assert!( + self.will_write, + "BUG: create the index with `for_writing()`" + ); assert!( !self.chunks.iter().any(|e| e.kind == chunk), "BUG: must not add chunk of same kind twice: {chunk:?}" @@ -93,7 +96,10 @@ impl Index { /// Return the total size of all planned chunks thus far. pub fn planned_storage_size(&self) -> u64 { - assert!(self.will_write, "BUG: create the index with `for_writing()`"); + assert!( + self.will_write, + "BUG: create the index with `for_writing()`" + ); self.chunks.iter().map(|e| e.offset.end).sum() } @@ -114,7 +120,8 @@ impl Index { "BUG: create the index with `for_writing()`, cannot write decoded indices" ); // First chunk starts past the table of contents - let mut current_offset = (current_offset + Self::size_for_entries(self.num_chunks())) as u64; + let mut current_offset = + (current_offset + Self::size_for_entries(self.num_chunks())) as u64; for entry in &self.chunks { out.write_all(&entry.kind)?; diff --git a/crates/gix-commitgraph/src/access.rs b/crates/gix-commitgraph/src/access.rs index cf30e3f64..5f4bdd3fd 100644 --- a/crates/gix-commitgraph/src/access.rs +++ b/crates/gix-commitgraph/src/access.rs @@ -55,7 +55,10 @@ impl Graph { /// Return changed-path Bloom filter settings used by the top-most compatible graph layer, if available. pub fn bloom_filter_settings(&self) -> Option { - self.files.iter().rev().find_map(File::bloom_filter_settings) + self.files + .iter() + .rev() + .find_map(File::bloom_filter_settings) } } diff --git a/crates/gix-commitgraph/src/bloom.rs b/crates/gix-commitgraph/src/bloom.rs index df707a43e..f14e7e82c 100644 --- a/crates/gix-commitgraph/src/bloom.rs +++ b/crates/gix-commitgraph/src/bloom.rs @@ -46,7 +46,9 @@ impl BloomKey { let (h0, h1) = match settings.hash_version { 1 => (murmur3_v1(SEED0, path), murmur3_v1(SEED1, path)), 2 => (murmur3_v2(SEED0, path), murmur3_v2(SEED1, path)), - version => panic!("BUG: unsupported Bloom hash version {version} should have been filtered earlier"), + version => panic!( + "BUG: unsupported Bloom hash version {version} should have been filtered earlier" + ), }; Self { h0, @@ -148,7 +150,11 @@ impl File { impl Graph { /// Query by commit id if `path` may be present in changed-path Bloom filters. - pub fn maybe_contains_path_by_id(&self, id: impl AsRef, path: &BStr) -> Option { + pub fn maybe_contains_path_by_id( + &self, + id: impl AsRef, + path: &BStr, + ) -> Option { let pos = self.lookup(id)?; self.maybe_contains_path(pos, path) } diff --git a/crates/gix-commitgraph/src/file/access.rs b/crates/gix-commitgraph/src/file/access.rs index cf5b580e9..94fbd2b6f 100644 --- a/crates/gix-commitgraph/src/file/access.rs +++ b/crates/gix-commitgraph/src/file/access.rs @@ -54,7 +54,8 @@ impl File { /// Return an iterator over all object hashes stored in the base graph. pub fn iter_base_graph_ids(&self) -> impl Iterator { let start = self.base_graphs_list_offset.unwrap_or(0); - let base_graphs_list = &self.data[start..][..self.hash_len * usize::from(self.base_graph_count)]; + let base_graphs_list = + &self.data[start..][..self.hash_len * usize::from(self.base_graph_count)]; base_graphs_list .chunks_exact(self.hash_len) .map(gix_hash::oid::from_bytes_unchecked) @@ -79,7 +80,11 @@ impl File { fn lookup_inner(&self, id: &gix_hash::oid) -> Option { let first_byte = usize::from(id.first_byte()); let mut upper_bound = self.fan[first_byte]; - let mut lower_bound = if first_byte != 0 { self.fan[first_byte - 1] } else { 0 }; + let mut lower_bound = if first_byte != 0 { + self.fan[first_byte - 1] + } else { + 0 + }; while lower_bound < upper_bound { let mid = u32::midpoint(lower_bound, upper_bound); diff --git a/crates/gix-commitgraph/src/file/commit.rs b/crates/gix-commitgraph/src/file/commit.rs index f20fdf5f5..e3db867b0 100644 --- a/crates/gix-commitgraph/src/file/commit.rs +++ b/crates/gix-commitgraph/src/file/commit.rs @@ -2,7 +2,10 @@ use crate::{ File, Position, bloom::BloomKey, - file::{self, CORRECTED_COMMIT_DATE_OFFSET_OVERFLOW, EXTENDED_EDGES_MASK, LAST_EXTENDED_EDGE_MASK, NO_PARENT}, + file::{ + self, CORRECTED_COMMIT_DATE_OFFSET_OVERFLOW, EXTENDED_EDGES_MASK, LAST_EXTENDED_EDGE_MASK, + NO_PARENT, + }, from_be_u32, }; use bstr::BStr; @@ -35,8 +38,9 @@ impl<'a> Commit<'a> { parent1: ParentEdge::from_raw(from_be_u32(&bytes[file.hash_len..][..4])), parent2: ParentEdge::from_raw(from_be_u32(&bytes[file.hash_len + 4..][..4])), generation: from_be_u32(&bytes[file.hash_len + 8..][..4]) >> 2, - commit_timestamp: u64::from_be_bytes(bytes[file.hash_len + 8..][..8].try_into().unwrap()) - & 0x0003_ffff_ffff, + commit_timestamp: u64::from_be_bytes( + bytes[file.hash_len + 8..][..8].try_into().unwrap(), + ) & 0x0003_ffff_ffff, } } @@ -61,7 +65,10 @@ impl<'a> Commit<'a> { /// It is the larger of this commit's committer timestamp and its parents' corrected commit dates plus one. pub fn corrected_commit_date(&self) -> Result, Message> { let file = self.file; - let Some(start) = file.generation_data_offset.filter(|_| file.read_generation_data) else { + let Some(start) = file + .generation_data_offset + .filter(|_| file.read_generation_data) + else { return Ok(None); }; let offset = from_be_u32(&file.data[start + self.pos.0 as usize * 4..][..4]); @@ -89,7 +96,12 @@ impl<'a> Commit<'a> { self.commit_timestamp .checked_add(offset) .map(Some) - .ok_or_else(|| message!("Commit {} has a corrected commit date that overflows", self.id())) + .ok_or_else(|| { + message!( + "Commit {} has a corrected commit date that overflows", + self.id() + ) + }) } /// Returns an iterator over the parent positions for lookup in the owning [Graph][crate::Graph]. @@ -243,12 +255,26 @@ impl Iterator for Parents<'_> { } fn size_hint(&self) -> (usize, Option) { - match (&self.state, self.commit_data.parent1, self.commit_data.parent2) { + match ( + &self.state, + self.commit_data.parent1, + self.commit_data.parent2, + ) { (ParentIteratorState::First, ParentEdge::None, ParentEdge::None) => (0, Some(0)), (ParentIteratorState::First, ParentEdge::None, _) => (1, Some(1)), - (ParentIteratorState::First, ParentEdge::GraphPosition(_), ParentEdge::None) => (1, Some(1)), - (ParentIteratorState::First, ParentEdge::GraphPosition(_), ParentEdge::GraphPosition(_)) => (2, Some(2)), - (ParentIteratorState::First, ParentEdge::GraphPosition(_), ParentEdge::ExtraEdgeIndex(_)) => (3, None), + (ParentIteratorState::First, ParentEdge::GraphPosition(_), ParentEdge::None) => { + (1, Some(1)) + } + ( + ParentIteratorState::First, + ParentEdge::GraphPosition(_), + ParentEdge::GraphPosition(_), + ) => (2, Some(2)), + ( + ParentIteratorState::First, + ParentEdge::GraphPosition(_), + ParentEdge::ExtraEdgeIndex(_), + ) => (3, None), (ParentIteratorState::First, ParentEdge::ExtraEdgeIndex(_), _) => (1, Some(1)), (ParentIteratorState::Second, _, ParentEdge::None) => (0, Some(0)), (ParentIteratorState::Second, _, ParentEdge::GraphPosition(_)) => (1, Some(1)), diff --git a/crates/gix-commitgraph/src/file/init.rs b/crates/gix-commitgraph/src/file/init.rs index 6af0449a9..3e743f1f3 100644 --- a/crates/gix-commitgraph/src/file/init.rs +++ b/crates/gix-commitgraph/src/file/init.rs @@ -5,9 +5,10 @@ use gix_error::{ErrorExt, Exn, Message, ResultExt, message}; use crate::{ BloomFilterSettings, File, file::{ - BASE_GRAPHS_LIST_CHUNK_ID, BLOOM_FILTER_DATA_CHUNK_ID, BLOOM_FILTER_HEADER_SIZE, BLOOM_FILTER_INDEX_CHUNK_ID, - COMMIT_DATA_CHUNK_ID, COMMIT_DATA_ENTRY_SIZE_SANS_HASH, EXTENDED_EDGES_LIST_CHUNK_ID, FAN_LEN, - GENERATION_DATA_CHUNK_ID, GENERATION_DATA_OVERFLOW_CHUNK_ID, HEADER_LEN, OID_FAN_CHUNK_ID, OID_LOOKUP_CHUNK_ID, + BASE_GRAPHS_LIST_CHUNK_ID, BLOOM_FILTER_DATA_CHUNK_ID, BLOOM_FILTER_HEADER_SIZE, + BLOOM_FILTER_INDEX_CHUNK_ID, COMMIT_DATA_CHUNK_ID, COMMIT_DATA_ENTRY_SIZE_SANS_HASH, + EXTENDED_EDGES_LIST_CHUNK_ID, FAN_LEN, GENERATION_DATA_CHUNK_ID, + GENERATION_DATA_OVERFLOW_CHUNK_ID, HEADER_LEN, OID_FAN_CHUNK_ID, OID_LOOKUP_CHUNK_ID, SIGNATURE, }, from_be_u32, @@ -37,7 +38,9 @@ impl File { let mut ofs = 0; if &data[ofs..ofs + SIGNATURE.len()] != SIGNATURE { - return Err(message("Commit-graph file does not start with expected signature").raise()); + return Err( + message("Commit-graph file does not start with expected signature").raise(), + ); } ofs += SIGNATURE.len(); @@ -49,8 +52,9 @@ impl File { } ofs += 1; - let object_hash = gix_hash::Kind::try_from(data[ofs]) - .map_err(|v| message!("Commit-graph file uses unsupported hash version: {v}").raise())?; + let object_hash = gix_hash::Kind::try_from(data[ofs]).map_err(|v| { + message!("Commit-graph file uses unsupported hash version: {v}").raise() + })?; ofs += 1; let chunk_count = data[ofs]; @@ -62,7 +66,11 @@ impl File { ofs += 1; let chunks = gix_chunk::file::Index::from_bytes(&data, ofs, u32::from(chunk_count)) - .or_raise(|| message!("Couldn't read commit-graph file with {chunk_count} chunks at offset {ofs}"))?; + .or_raise(|| { + message!( + "Couldn't read commit-graph file with {chunk_count} chunks at offset {ofs}" + ) + })?; let base_graphs_list_offset = chunks .validated_usize_offset_by_id(BASE_GRAPHS_LIST_CHUNK_ID, |chunk_range| { @@ -139,7 +147,9 @@ impl File { .filter(|range| range.len() / 4 == commit_data_count as usize) .map(|range| range.start); // The overflow chunk is only checked when reading from it. - let generation_data_overflow_range = chunks.usize_offset_by_id(GENERATION_DATA_OVERFLOW_CHUNK_ID).ok(); + let generation_data_overflow_range = chunks + .usize_offset_by_id(GENERATION_DATA_OVERFLOW_CHUNK_ID) + .ok(); let bloom_filter_index_range = chunks.usize_offset_by_id(BLOOM_FILTER_INDEX_CHUNK_ID).ok(); let bloom_filter_data_range = chunks.usize_offset_by_id(BLOOM_FILTER_DATA_CHUNK_ID).ok(); @@ -147,11 +157,14 @@ impl File { let mut bloom_filter_data_offset = None; let mut bloom_filter_index_offset = None; let mut bloom_filter_settings = None; - if let (Some(index_range), Some(data_range)) = - (bloom_filter_index_range.as_ref(), bloom_filter_data_range.as_ref()) - { + if let (Some(index_range), Some(data_range)) = ( + bloom_filter_index_range.as_ref(), + bloom_filter_data_range.as_ref(), + ) { let expected_index_chunk_size = commit_data_count as usize * std::mem::size_of::(); - if index_range.len() == expected_index_chunk_size && data_range.len() >= BLOOM_FILTER_HEADER_SIZE { + if index_range.len() == expected_index_chunk_size + && data_range.len() >= BLOOM_FILTER_HEADER_SIZE + { let settings = BloomFilterSettings { hash_version: from_be_u32(&data[data_range.start..][..4]), num_hashes: from_be_u32(&data[data_range.start + 4..][..4]), @@ -178,7 +191,10 @@ impl File { } if base_graph_count > 0 && base_graphs_list_offset.is_none() { - return Err(message!("Chunk named {BASE_GRAPHS_LIST_CHUNK_ID:?} was not found in chunk file index").into()); + return Err(message!( + "Chunk named {BASE_GRAPHS_LIST_CHUNK_ID:?} was not found in chunk file index" + ) + .into()); } let (fan, _) = read_fan(&data[fan_offset..]); @@ -230,7 +246,12 @@ impl TryFrom<&Path> for File { memmap2::MmapOptions::new().map_copy_read_only(&file) } }) - .or_raise(|| message!("Could not open commit-graph file at '{path}'", path = path.display()))?; + .or_raise(|| { + message!( + "Could not open commit-graph file at '{path}'", + path = path.display() + ) + })?; Self::new(data, path.to_owned()) } } diff --git a/crates/gix-commitgraph/src/file/verify.rs b/crates/gix-commitgraph/src/file/verify.rs index 735719463..b1c5c6a95 100644 --- a/crates/gix-commitgraph/src/file/verify.rs +++ b/crates/gix-commitgraph/src/file/verify.rs @@ -35,7 +35,10 @@ impl File { /// Traverse all [commits][file::Commit] stored in this file and call `processor(commit) -> Result<(), Error>` on it. /// /// If the `processor` fails, the iteration will be stopped and the entire call results in the respective error. - pub fn traverse<'a, Processor>(&'a self, mut processor: Processor) -> Result> + pub fn traverse<'a, Processor>( + &'a self, + mut processor: Processor, + ) -> Result> where Processor: FnMut(&file::Commit<'a>) -> Result<(), Exn>, { @@ -80,14 +83,22 @@ impl File { .raise()); } if commit.generation() > GENERATION_NUMBER_MAX { - return Err(message!("commit {} has invalid generation {}", commit.id(), commit.generation()).raise()); + return Err(message!( + "commit {} has invalid generation {}", + commit.id(), + commit.generation() + ) + .raise()); } - processor(&commit).or_raise(|| message!("processor failed on commit {}", commit.id()))?; + processor(&commit) + .or_raise(|| message!("processor failed on commit {}", commit.id()))?; stats.max_generation = max(stats.max_generation, commit.generation()); stats.min_generation = min(stats.min_generation, commit.generation()); - let parent_count = commit.iter_parents().try_fold(0u32, |acc, pos| pos.map(|_| acc + 1))?; + let parent_count = commit + .iter_parents() + .try_fold(0u32, |acc, pos| pos.map(|_| acc + 1))?; *stats.parent_counts.entry(parent_count).or_insert(0) += 1; prev_id = commit.id(); } @@ -114,20 +125,31 @@ impl File { let actual = hasher .try_finalize() .map_err(|e| message!("failed to hash commit graph file: {e}").raise())?; - actual.verify(self.checksum()).map_err(|e| message!("{e}").raise())?; + actual + .verify(self.checksum()) + .map_err(|e| message!("{e}").raise())?; Ok(actual) } } /// If the given path's filename matches "graph-{hash}.graph", check that `hash` matches the /// expected hash. -fn verify_split_chain_filename_hash(path: &Path, expected: &gix_hash::oid) -> Result<(), Exn> { +fn verify_split_chain_filename_hash( + path: &Path, + expected: &gix_hash::oid, +) -> Result<(), Exn> { path.file_name() .and_then(std::ffi::OsStr::to_str) .and_then(|filename| filename.strip_suffix(".graph")) .and_then(|stem| stem.strip_prefix("graph-")) - .map_or(Ok(()), |hex| match gix_hash::ObjectId::from_hex(hex.as_bytes()) { - Ok(actual) if actual == expected => Ok(()), - _ => Err(message!("commit-graph filename should be graph-{}.graph", expected.to_hex()).raise()), + .map_or(Ok(()), |hex| { + match gix_hash::ObjectId::from_hex(hex.as_bytes()) { + Ok(actual) if actual == expected => Ok(()), + _ => Err(message!( + "commit-graph filename should be graph-{}.graph", + expected.to_hex() + ) + .raise()), + } }) } diff --git a/crates/gix-commitgraph/src/init.rs b/crates/gix-commitgraph/src/init.rs index d822a4b6f..74f62fc1d 100644 --- a/crates/gix-commitgraph/src/init.rs +++ b/crates/gix-commitgraph/src/init.rs @@ -52,7 +52,8 @@ impl Graph { /// Instantiate a commit graph from a `.git/objects/info/commit-graph` or /// `.git/objects/info/commit-graphs/graph-*.graph` file. pub fn from_file(path: &Path) -> Result> { - let file = File::at(path).or_raise(|| message!("Could not open commit-graph file at '{}'", path.display()))?; + let file = File::at(path) + .or_raise(|| message!("Could not open commit-graph file at '{}'", path.display()))?; Ok(Self::new(vec![file])?) } diff --git a/crates/gix-commitgraph/src/write/info_dir.rs b/crates/gix-commitgraph/src/write/info_dir.rs index 6f9cf3065..bccd58f70 100644 --- a/crates/gix-commitgraph/src/write/info_dir.rs +++ b/crates/gix-commitgraph/src/write/info_dir.rs @@ -66,7 +66,12 @@ pub fn write_to_info_dir( gix_tempfile::ContainingDirectory::Exists, gix_tempfile::AutoRemove::Tempfile, ) - .or_raise(|| message!("Could not create a temporary file in '{}'", graphs_dir.display()))?; + .or_raise(|| { + message!( + "Could not create a temporary file in '{}'", + graphs_dir.display() + ) + })?; let mut out = std::io::BufWriter::new(layer); let Some(outcome) = write(objects, graph.as_ref(), tips, options, &mut out)? else { return Ok(None); @@ -79,8 +84,9 @@ pub fn write_to_info_dir( match base_layers.last() { // A monolithic `commit-graph` kept below the new layer moves into the chain. Some((path, checksum)) if *path != layer_path(checksum) => { - std::fs::rename(path, layer_path(checksum)) - .or_raise(|| message!("Could not rename '{}' into the chain", path.display()))?; + std::fs::rename(path, layer_path(checksum)).or_raise(|| { + message!("Could not rename '{}' into the chain", path.display()) + })?; } Some(_) => {} None => remove(&info_dir.join("commit-graph"))?, @@ -98,7 +104,8 @@ pub fn write_to_info_dir( .chain(Some(outcome.checksum)) .collect(); for checksum in &chain { - writeln!(chain_lock, "{checksum}").or_raise(|| message!("Could not write commit-graph chain"))?; + writeln!(chain_lock, "{checksum}") + .or_raise(|| message!("Could not write commit-graph chain"))?; } commit(chain_lock)?; make_read_only(&chain_path); @@ -111,7 +118,11 @@ pub fn write_to_info_dir( } (outcome, chain.iter().map(layer_path).collect()) }; - expire(&graphs_dir, &chain, expire_time.unwrap_or_else(SystemTime::now))?; + expire( + &graphs_dir, + &chain, + expire_time.unwrap_or_else(SystemTime::now), + )?; Ok(Some(outcome)) } @@ -130,7 +141,11 @@ fn commit(lock: gix_lock::File) -> Result<(), Exn> { } /// Delete the `.graph` files in `graphs_dir` that aren't in `chain` and weren't modified after `expire_time`. -fn expire(graphs_dir: &Path, chain: &[PathBuf], expire_time: SystemTime) -> Result<(), Exn> { +fn expire( + graphs_dir: &Path, + chain: &[PathBuf], + expire_time: SystemTime, +) -> Result<(), Exn> { let Ok(entries) = std::fs::read_dir(graphs_dir) else { return Ok(()); }; diff --git a/crates/gix-commitgraph/src/write/mod.rs b/crates/gix-commitgraph/src/write/mod.rs index 036274179..a601cf40d 100644 --- a/crates/gix-commitgraph/src/write/mod.rs +++ b/crates/gix-commitgraph/src/write/mod.rs @@ -12,10 +12,10 @@ use crate::{ BloomFilterSettings, File, Graph, Position, bloom::BloomKey, file::{ - BASE_GRAPHS_LIST_CHUNK_ID, BLOOM_FILTER_DATA_CHUNK_ID, BLOOM_FILTER_INDEX_CHUNK_ID, COMMIT_DATA_CHUNK_ID, - CORRECTED_COMMIT_DATE_OFFSET_OVERFLOW, EXTENDED_EDGES_LIST_CHUNK_ID, EXTENDED_EDGES_MASK, - GENERATION_DATA_CHUNK_ID, GENERATION_DATA_OVERFLOW_CHUNK_ID, LAST_EXTENDED_EDGE_MASK, NO_PARENT, - OID_FAN_CHUNK_ID, OID_LOOKUP_CHUNK_ID, SIGNATURE, + BASE_GRAPHS_LIST_CHUNK_ID, BLOOM_FILTER_DATA_CHUNK_ID, BLOOM_FILTER_INDEX_CHUNK_ID, + COMMIT_DATA_CHUNK_ID, CORRECTED_COMMIT_DATE_OFFSET_OVERFLOW, EXTENDED_EDGES_LIST_CHUNK_ID, + EXTENDED_EDGES_MASK, GENERATION_DATA_CHUNK_ID, GENERATION_DATA_OVERFLOW_CHUNK_ID, + LAST_EXTENDED_EDGE_MASK, NO_PARENT, OID_FAN_CHUNK_ID, OID_LOOKUP_CHUNK_ID, SIGNATURE, }, }; @@ -175,7 +175,9 @@ pub fn write( return Ok(None); }; // Readers ignore generation data unless every layer has it. - let write_generation_data = base_layers.last().is_none_or(|file| file.has_generation_data()); + let write_generation_data = base_layers + .last() + .is_none_or(|file| file.has_generation_data()); let generation_data = generation_data(&commits, graph); let filters = match bloom_settings { Some(settings) => { @@ -275,7 +277,14 @@ fn collect<'a>( missing.insert(id); } } - walk.run(objects, Some(graph), merged, num_commits_in_base, &missing, object_hash)?; + walk.run( + objects, + Some(graph), + merged, + num_commits_in_base, + &missing, + object_hash, + )?; Ok(Some(( walk.into_commits(Some(graph), num_commits_in_base)?, base_layers, @@ -340,16 +349,20 @@ impl Walk { continue; } if missing.contains(&id) { - return Err(message!("Commit {id} is in the commit-graph, but it could not be found").raise()); + return Err(message!( + "Commit {id} is in the commit-graph, but it could not be found" + ) + .raise()); } let start = self.parent_ids.len(); - let mut commit = match graph.and_then(|graph| from_graph(graph, &id, &mut self.parent_ids)) { - Some(commit) => commit, - None => { - self.parent_ids.truncate(start); - decode_commit(objects, &id, &mut buf, &mut self.parent_ids, object_hash)? - } - }; + let mut commit = + match graph.and_then(|graph| from_graph(graph, &id, &mut self.parent_ids)) { + Some(commit) => commit, + None => { + self.parent_ids.truncate(start); + decode_commit(objects, &id, &mut buf, &mut self.parent_ids, object_hash)? + } + }; commit.parents = (start as u32, self.parent_ids.len() as u32); stack.extend_from_slice(&self.parent_ids[start..]); self.index.insert(id, self.ids.len() as u32); @@ -360,7 +373,11 @@ impl Walk { } /// Sort the commits by id, placing them after the `num_commits_in_base` commits of `graph`'s base layers. - fn into_commits(self, graph: Option<&Graph>, num_commits_in_base: u32) -> Result> { + fn into_commits( + self, + graph: Option<&Graph>, + num_commits_in_base: u32, + ) -> Result> { let mut order: Vec = (0..self.ids.len() as u32).collect(); order.sort_unstable_by_key(|&index| self.ids[index as usize]); let mut position = vec![0u32; self.ids.len()]; @@ -448,7 +465,11 @@ fn decode_commit( }) } -fn find<'a>(objects: &impl Find, id: &ObjectId, buf: &'a mut Vec) -> Result, Exn> { +fn find<'a>( + objects: &impl Find, + id: &ObjectId, + buf: &'a mut Vec, +) -> Result, Exn> { objects .try_find(id, buf) .map_err(|err| message!("Could not read object {id}: {err}"))? @@ -484,15 +505,17 @@ fn peel_to_commit( /// Return the generation and corrected commit date of each commit. /// Commits taken from the existing graph keep theirs, and so do parents in its base layers. fn generation_data(commits: &Commits, graph: Option<&Graph>) -> Vec<(u32, u64)> { - let mut out: Vec> = commits.commits.iter().map(|commit| commit.known).collect(); + let mut out: Vec> = + commits.commits.iter().map(|commit| commit.known).collect(); let num_commits_in_base = commits.num_commits_in_base; - let value = |out: &[Option<(u32, u64)>], parent: u32| match parent.checked_sub(num_commits_in_base) { - Some(pos) => out[pos as usize], - None => Some(base_generation_data( - graph.expect("parents in base layers are in the graph"), - parent, - )), - }; + let value = + |out: &[Option<(u32, u64)>], parent: u32| match parent.checked_sub(num_commits_in_base) { + Some(pos) => out[pos as usize], + None => Some(base_generation_data( + graph.expect("parents in base layers are in the graph"), + parent, + )), + }; let mut stack = Vec::new(); for root in 0..out.len() { stack.push(root); @@ -510,10 +533,12 @@ fn generation_data(commits: &Commits, graph: Option<&Graph>) -> Vec<(u32, u64)> .map(|&parent| (parent - num_commits_in_base) as usize), ); if stack.len() == before { - let (generation, date) = parents.iter().fold((0, 0), |(generation, date), &parent| { - let (parent_generation, parent_date) = value(&out, parent).expect("parents are done"); - (generation.max(parent_generation), date.max(parent_date)) - }); + let (generation, date) = + parents.iter().fold((0, 0), |(generation, date), &parent| { + let (parent_generation, parent_date) = + value(&out, parent).expect("parents are done"); + (generation.max(parent_generation), date.max(parent_date)) + }); let seconds = commits.commits[top].seconds.max(0) as u64; out[top] = Some(( generation.saturating_add(1).min(GENERATION_NUMBER_MAX), @@ -565,7 +590,9 @@ impl Filters<'_> { Filter::Missing => &[], Filter::Stored(filter) => filter, Filter::Computed(index) => { - let start = index.checked_sub(1).map_or(0, |previous| self.ends[previous]); + let start = index + .checked_sub(1) + .map_or(0, |previous| self.ends[previous]); &self.data[start..self.ends[index]] } } @@ -592,16 +619,21 @@ fn changed_path_filters<'a>( let mut jobs: Vec<(Option, ObjectId)> = Vec::new(); for &pos in order { let tree = commits.commits[pos].tree; - let parent_tree = commits.parents(pos).first().map(|&parent| match commits.at(parent) { - Some(commit) => commit.tree, - None => graph - .expect("parents in base layers are in the graph") - .commit_at(Position(parent)) - .root_tree_id() - .to_owned(), - }); + let parent_tree = commits + .parents(pos) + .first() + .map(|&parent| match commits.at(parent) { + Some(commit) => commit.tree, + None => graph + .expect("parents in base layers are in the graph") + .commit_at(Position(parent)) + .root_tree_id() + .to_owned(), + }); let may_compute = jobs.len() < max_new_filters; - if let Some((stored, version)) = graph.and_then(|graph| stored_filter(graph, &commits.ids[pos], settings)) { + if let Some((stored, version)) = + graph.and_then(|graph| stored_filter(graph, &commits.ids[pos], settings)) + { let reuse = version == settings.hash_version || (may_compute && !has_high_bit_paths(objects, Some(tree), &mut clean_trees, object_hash)? @@ -629,7 +661,8 @@ fn changed_path_filters<'a>( // Large chunks of consecutive commits keep each thread's trees close together, which matters more than // balancing the load: with 10000 the linux kernel's filters are computed 15% faster than with 1000. - let chunk_size = (jobs.len() / (gix_features::parallel::num_threads(thread_limit) * 8)).clamp(1, 10_000); + let chunk_size = + (jobs.len() / (gix_features::parallel::num_threads(thread_limit) * 8)).clamp(1, 10_000); let chunks = jobs.chunks(chunk_size); let num_chunks = chunks.len(); let computed = gix_features::parallel::in_parallel( @@ -661,7 +694,9 @@ fn changed_path_filters<'a>( for (data, ends) in computed { let offset = filters.data.len(); filters.data.extend_from_slice(&data); - filters.ends.extend(ends.into_iter().map(|end| offset + end)); + filters + .ends + .extend(ends.into_iter().map(|end| offset + end)); } Ok(filters) } @@ -691,7 +726,11 @@ impl gix_features::parallel::Reduce for InOrder { } /// Return the filter `graph` has for `id` and its hash version, unless it wasn't computed or used other parameters. -fn stored_filter<'a>(graph: &'a Graph, id: &ObjectId, settings: BloomFilterSettings) -> Option<(&'a [u8], u32)> { +fn stored_filter<'a>( + graph: &'a Graph, + id: &ObjectId, + settings: BloomFilterSettings, +) -> Option<(&'a [u8], u32)> { let commit = graph.commit_by_id(id)?; let stored = commit.file.bloom_filter_settings()?; let filter = commit.bloom_filter().filter(|filter| !filter.is_empty())?; @@ -767,7 +806,8 @@ impl TreeDiff { while let Some((parent, commit, prefix)) = changes.pending.pop() { let mut lhs = tree_iter(objects, parent, &mut self.parent, object_hash)?; let mut rhs = tree_iter(objects, commit, &mut self.commit, object_hash)?; - let (mut left, mut right) = (next_entry(&mut lhs, parent)?, next_entry(&mut rhs, commit)?); + let (mut left, mut right) = + (next_entry(&mut lhs, parent)?, next_entry(&mut rhs, commit)?); loop { let order = match (&left, &right) { (None, None) => break, @@ -787,7 +827,10 @@ impl TreeDiff { right = next_entry(&mut rhs, commit)?; } Ordering::Equal => { - let (old, new) = (left.take().expect("present"), right.take().expect("present")); + let (old, new) = ( + left.take().expect("present"), + right.take().expect("present"), + ); if old.oid != new.oid || old.mode.kind() != new.mode.kind() { if new.mode.is_tree() { changes.pending.push(( @@ -885,12 +928,19 @@ fn tree_iter<'a>( }; let data = find(objects, &id, buf)?; Ok(TreeRefIter::from_bytes( - if data.kind == Kind::Tree { data.data } else { &[] }, + if data.kind == Kind::Tree { + data.data + } else { + &[] + }, object_hash, )) } -fn next_entry<'a>(iter: &mut TreeRefIter<'a>, tree: Option) -> Result>, Exn> { +fn next_entry<'a>( + iter: &mut TreeRefIter<'a>, + tree: Option, +) -> Result>, Exn> { iter.next().transpose().or_raise(|| { message!( "Could not decode tree {}", @@ -928,7 +978,9 @@ fn serialize( .commits .iter() .zip(generation_data) - .map(|(commit, (_, corrected))| corrected.saturating_sub(commit.seconds.max(0) as u64 & COMMIT_DATE_MASK)) + .map(|(commit, (_, corrected))| { + corrected.saturating_sub(commit.seconds.max(0) as u64 & COMMIT_DATE_MASK) + }) }; let num_overflows = if write_generation_data { offsets() @@ -957,12 +1009,17 @@ fn serialize( index.plan_chunk(EXTENDED_EDGES_LIST_CHUNK_ID, num_edges * 4); } if let Some((_, filters)) = filters { - let filter_bytes: u64 = (0..commits.ids.len()).map(|pos| filters.get(pos).len() as u64).sum(); + let filter_bytes: u64 = (0..commits.ids.len()) + .map(|pos| filters.get(pos).len() as u64) + .sum(); index.plan_chunk(BLOOM_FILTER_INDEX_CHUNK_ID, num_commits * 4); index.plan_chunk(BLOOM_FILTER_DATA_CHUNK_ID, 12 + filter_bytes); } if !base_layers.is_empty() { - index.plan_chunk(BASE_GRAPHS_LIST_CHUNK_ID, base_layers.len() as u64 * hash_len); + index.plan_chunk( + BASE_GRAPHS_LIST_CHUNK_ID, + base_layers.len() as u64 * hash_len, + ); } let mut out = std::io::BufWriter::new(gix_hash::io::Write::new(out, object_hash)); @@ -971,7 +1028,8 @@ fn serialize( gix_hash::Kind::Sha1 => 1u8, _ => 2u8, }; - let num_base_layers = u8::try_from(base_layers.len()).map_err(|_| std::io::Error::other("Too many base layers"))?; + let num_base_layers = u8::try_from(base_layers.len()) + .map_err(|_| std::io::Error::other("Too many base layers"))?; out.write_all(&[1, hash_version, index.num_chunks() as u8, num_base_layers])?; let mut out = index.into_write(out, SIGNATURE.len() + 4)?; while let Some(chunk) = out.next_chunk() { @@ -994,7 +1052,9 @@ fn serialize( } COMMIT_DATA_CHUNK_ID => { let mut num_edges = 0; - for (pos, (commit, (generation, _))) in commits.commits.iter().zip(generation_data).enumerate() { + for (pos, (commit, (generation, _))) in + commits.commits.iter().zip(generation_data).enumerate() + { out.write_all(commit.tree.as_slice())?; let fields = match commits.parents(pos) { [] => [NO_PARENT, NO_PARENT], @@ -1025,7 +1085,8 @@ fn serialize( } } GENERATION_DATA_OVERFLOW_CHUNK_ID => { - for offset in offsets().filter(|&offset| offset > CORRECTED_COMMIT_DATE_OFFSET_MAX) { + for offset in offsets().filter(|&offset| offset > CORRECTED_COMMIT_DATE_OFFSET_MAX) + { out.write_all(&offset.to_be_bytes())?; } } @@ -1052,7 +1113,11 @@ fn serialize( } BLOOM_FILTER_DATA_CHUNK_ID => { let (settings, filters) = filters.expect("planned with filters"); - for value in [settings.hash_version, settings.num_hashes, settings.bits_per_entry] { + for value in [ + settings.hash_version, + settings.num_hashes, + settings.bits_per_entry, + ] { out.write_all(&value.to_be_bytes())?; } for pos in 0..commits.ids.len() { diff --git a/gitmirror/crates/gitmirror-sync/src/task.rs b/gitmirror/crates/gitmirror-sync/src/task.rs index 933aae207..7158c0d9b 100644 --- a/gitmirror/crates/gitmirror-sync/src/task.rs +++ b/gitmirror/crates/gitmirror-sync/src/task.rs @@ -12,8 +12,8 @@ use jacquard_identity::resolver::IdentityResolver as _; use thiserror::Error; use tracing::debug; -use crate::admission::BackfillAdmission; use crate::Directory; +use crate::admission::BackfillAdmission; const AGENT: &str = concat!("gitmirror/", env!("CARGO_PKG_VERSION")); const LISTED_PREFIXES: [&str; 2] = ["refs/cobs/", "refs/tngl/keep/"]; diff --git a/knot2/crates/knot-atproto/src/identity.rs b/knot2/crates/knot-atproto/src/identity.rs index b8e6f1450..944a158cf 100644 --- a/knot2/crates/knot-atproto/src/identity.rs +++ b/knot2/crates/knot-atproto/src/identity.rs @@ -8,7 +8,7 @@ use serde::Serialize; use sha2::{Digest, Sha256}; const PLC_OP_TYPE: &str = "plc_operation"; -pub(crate) const ATPROTO_METHOD: &str = "atproto"; +pub const ATPROTO_METHOD: &str = "atproto"; pub(crate) const KNOT_HOME_SERVICE: &str = "tangled_knot"; pub(crate) const KNOT_HOME_TYPE: &str = "TangledKnot"; const PDS_SERVICE_ID: &str = crate::plc::ATPROTO_PDS_SERVICE; diff --git a/knot2/crates/knot-atproto/src/lib.rs b/knot2/crates/knot-atproto/src/lib.rs index 828647eb6..6830b930b 100644 --- a/knot2/crates/knot-atproto/src/lib.rs +++ b/knot2/crates/knot-atproto/src/lib.rs @@ -10,7 +10,7 @@ mod test_support; pub use auth::{OauthAuthorizer, PointerAuth, PointerAuthorizer, ServiceAuth}; pub use identity::{ - IdentityError, MintNonce, PreparedRepoDid, knot_did_document, prepare_repo_did, + ATPROTO_METHOD, IdentityError, MintNonce, PreparedRepoDid, knot_did_document, prepare_repo_did, }; pub use jwt::{ JwtError, JwtNonce, MintExpiryError, RepoTokenExpiry, RepoTokenRequest, ServiceJwt, diff --git a/knot2/crates/knot-cob/src/lib.rs b/knot2/crates/knot-cob/src/lib.rs index 2a0d0ff45..bcaa424a8 100644 --- a/knot2/crates/knot-cob/src/lib.rs +++ b/knot2/crates/knot-cob/src/lib.rs @@ -706,19 +706,30 @@ impl<'r> CobStore<'r> { owner: &ActorId, ) -> Result<(), CobError> { let type_name = E::Change::type_name(); - let (graph, _tip) = backend::load_graph( - self.repo, - &type_name, - object, - E::HISTORY, - backend::MAX_GRAPH_CHANGES, - )?; - graph.into_ordered().into_iter().try_for_each(|change| { - if !accepts_change_type(&type_name, &change.type_name) { + let tip = ChangeId::new( + backend::resolve_tip(self.repo, &type_name, object)? + .ok_or(CobError::NoSuchObject(object))?, + ); + self.verify_type_from_tip(home, &type_name, object, tip, E::HISTORY, owner) + } + + pub fn verify_type_from_tip( + &self, + home: &CobHome, + type_name: &TypeName, + object: CobId, + tip: ChangeId, + history: HistoryModel, + owner: &ActorId, + ) -> Result<(), CobError> { + let changes = backend::collect(self.repo, tip, object, backend::MAX_GRAPH_CHANGES, None)?; + backend::check_full_shape(&changes, object, history)?; + changes.values().try_for_each(|change| { + if !accepts_change_type(type_name, &change.type_name) { return Err(CobError::UnexpectedChangeType { change: change.id, expected: type_name.clone(), - found: change.type_name, + found: change.type_name.clone(), }); } change diff --git a/knot2/crates/knot-cobs/src/lib.rs b/knot2/crates/knot-cobs/src/lib.rs index dc0013fa8..0f04fe6f5 100644 --- a/knot2/crates/knot-cobs/src/lib.rs +++ b/knot2/crates/knot-cobs/src/lib.rs @@ -31,7 +31,9 @@ pub use social::{ AnyBody, AnySocialChange, Creation, Editing, Emission, Erasure, Opening, Social, SocialBody, SocialChange, SocialCob, SocialObject, SocialState, decode_social, }; -pub use ticket::{Ticket, TicketChange, TicketCob, social_collections, social_state}; +pub use ticket::{ + Ticket, TicketChange, TicketCob, history_model_of, social_collections, social_state, +}; #[doc(hidden)] pub mod fuzz { diff --git a/knot2/crates/knot-cobs/src/ticket.rs b/knot2/crates/knot-cobs/src/ticket.rs index 1aaf6832e..c98dd076a 100644 --- a/knot2/crates/knot-cobs/src/ticket.rs +++ b/knot2/crates/knot-cobs/src/ticket.rs @@ -41,6 +41,16 @@ macro_rules! social_kinds { _ => Ok(None), } } + + pub fn history_model_of( + type_name: &knot_types::TypeName, + ) -> Option { + match type_name.as_str() { + $(<$body as $crate::SocialBody>::COLLECTION => + Some(<$crate::SocialCob<$body> as knot_cob::Evaluate>::HISTORY),)+ + _ => None, + } + } }; } diff --git a/knot2/crates/knot-git/src/repo.rs b/knot2/crates/knot-git/src/repo.rs index d7955684b..80550b6d4 100644 --- a/knot2/crates/knot-git/src/repo.rs +++ b/knot2/crates/knot-git/src/repo.rs @@ -691,15 +691,12 @@ impl Repo { return Ok(true); }; let in_graph = |oid: Oid| graph.lookup(oid.object_id()).is_some(); - Ok(self - .references()? - .into_iter() - .any(|record| { - !in_graph(record.target) - && self - .peel_to_commit(record.target) - .is_ok_and(|tip| !in_graph(tip)) - })) + Ok(self.references()?.into_iter().any(|record| { + !in_graph(record.target) + && self + .peel_to_commit(record.target) + .is_ok_and(|tip| !in_graph(tip)) + })) } pub fn with_ref_lock(&self, body: impl FnOnce() -> R) -> R { diff --git a/knot2/crates/knot-index/src/lib.rs b/knot2/crates/knot-index/src/lib.rs index d26fc5c71..668d89a12 100644 --- a/knot2/crates/knot-index/src/lib.rs +++ b/knot2/crates/knot-index/src/lib.rs @@ -751,6 +751,9 @@ impl Index { pub fn owner_of(&self, repo: &RepoDid) -> Resolved> { self.registry.owner_of(&self.interner, repo) } + pub fn is_fostered(&self, repo: &RepoDid) -> Resolved { + self.registry.fostered_of(&self.interner, repo) + } pub fn rkey_of(&self, repo: &RepoDid) -> Resolved> { self.registry.rkey_of(&self.interner, repo) diff --git a/knot2/crates/knot-index/src/projections.rs b/knot2/crates/knot-index/src/projections.rs index 4b5600200..b63c58276 100644 --- a/knot2/crates/knot-index/src/projections.rs +++ b/knot2/crates/knot-index/src/projections.rs @@ -635,6 +635,19 @@ impl RegistryProjection { }), ) } + pub(crate) fn fostered_of(&self, interner: &Interner, repo: &RepoDid) -> Resolved { + if self.coverage.get() == Coverage::Warming { + return Resolved::Warming; + } + let Some(target) = interner.repo(repo) else { + return Resolved::Ready(false); + }; + Resolved::Ready( + self.records + .read_sync(&target, |_, slot| slot.fostered) + .unwrap_or(false), + ) + } pub(crate) fn hosted_repos(&self, interner: &Interner) -> Vec { let mut repos = BTreeSet::new(); @@ -799,8 +812,12 @@ impl RegistryProjection { let record = self .records .read_sync(&repo, |_, slot| { - (slot.owner == owner) - .then_some((slot.created_at, slot.name, slot.policy, slot.fostered)) + (slot.owner == owner).then_some(( + slot.created_at, + slot.name, + slot.policy, + slot.fostered, + )) }) .flatten(); let Some((created_at, previous, policy, fostered)) = record else { diff --git a/knot2/crates/knot-maintenance/tests/commitgraph.rs b/knot2/crates/knot-maintenance/tests/commitgraph.rs index 48e24ae02..12b75676e 100644 --- a/knot2/crates/knot-maintenance/tests/commitgraph.rs +++ b/knot2/crates/knot-maintenance/tests/commitgraph.rs @@ -306,11 +306,19 @@ fn gate_skips_graphed_refs_and_annotated_tags() { let tmp = tempfile::tempdir().unwrap(); let root = tmp.path(); seed_history(root, "sha1"); - let misses = || Repo::open(root).unwrap().commit_graph_misses_a_tip().unwrap(); + let misses = || { + Repo::open(root) + .unwrap() + .commit_graph_misses_a_tip() + .unwrap() + }; assert!(misses(), "no graph yet"); git(root, &["commit-graph", "write", "--reachable"]); assert!(!misses(), "every tip is graphed"); - git(root, &["tag", "-a", "-m", "annotated", "v1-annotated", "v1"]); + git( + root, + &["tag", "-a", "-m", "annotated", "v1-annotated", "v1"], + ); assert!(!misses(), "an annotated tag peels to a graphed commit"); commit(root, "c.txt", "c3", "ungraphed"); assert!(misses(), "a new tip is missing from the graph"); diff --git a/knot2/crates/knot-record/src/manifest.rs b/knot2/crates/knot-record/src/manifest.rs index cf2190ce8..549c8d675 100644 --- a/knot2/crates/knot-record/src/manifest.rs +++ b/knot2/crates/knot-record/src/manifest.rs @@ -3,8 +3,8 @@ use std::str::FromStr; use jacquard_common::deps::codegen::unicode_segmentation::UnicodeSegmentation; use knot_types::{ - AccountDid, ContributionPolicy, OwnerDid, RecordBody, RecordBodyError, RepoDid, - RepoName, ServiceDid, UnixSeconds, + AccountDid, ContributionPolicy, OwnerDid, RecordBody, RecordBodyError, RepoDid, RepoName, + ServiceDid, UnixSeconds, }; use serde::{Deserialize, Serialize, Serializer}; use url::Url; diff --git a/knot2/crates/knot-server/src/main.rs b/knot2/crates/knot-server/src/main.rs index f54f04c10..400d4f266 100644 --- a/knot2/crates/knot-server/src/main.rs +++ b/knot2/crates/knot-server/src/main.rs @@ -649,6 +649,7 @@ async fn run(config: knot_config::Validated) -> anyhow::Result<()> { appview: appview_endpoint, slots: slots.clone(), imports, + handover_sessions: Arc::new(knot_xrpc::handover::HandoverSessions::default()), firehose: Arc::new( knot_events::EventLog::new(SystemClock, replay_bounds).with_boot_floor(firehose_floor), ), @@ -784,6 +785,10 @@ async fn run(config: knot_config::Validated) -> anyhow::Result<()> { )); tracing::info!("import worker running"); } + tokio::spawn(knot_xrpc::handover::run_boot( + Arc::clone(&xrpc_state), + shutdown.clone(), + )); tokio::spawn(allocator::govern_decay(shutdown.clone())); let keyfill_task = knot_keyfill::spawn( Arc::clone(&index), diff --git a/knot2/crates/knot-sim/src/harness.rs b/knot2/crates/knot-sim/src/harness.rs index 8982755fc..b6c3a6782 100644 --- a/knot2/crates/knot-sim/src/harness.rs +++ b/knot2/crates/knot-sim/src/harness.rs @@ -792,6 +792,7 @@ fn assemble_router(parts: StateParts) -> (Router, FakeState) { let outbox = knot_xrpc::Outbox::open(&meta_path, Default::default()); let state = Arc::new(XrpcState { imports: knot_xrpc::imports::ImportQueue::channel().0, + handover_sessions: Arc::new(knot_xrpc::handover::HandoverSessions::default()), layout: layout.clone(), index: Arc::clone(&index), atproto, diff --git a/knot2/crates/knot-sim/tests/lfs_roundtrip.rs b/knot2/crates/knot-sim/tests/lfs_roundtrip.rs index 3e022d7c9..518a9498b 100644 --- a/knot2/crates/knot-sim/tests/lfs_roundtrip.rs +++ b/knot2/crates/knot-sim/tests/lfs_roundtrip.rs @@ -365,6 +365,7 @@ async fn spawn(published_line: String, with_h3: bool) -> World { let outbox = knot_xrpc::Outbox::open(&meta_path, Default::default()); let xrpc_state = Arc::new(knot_xrpc::XrpcState { imports, + handover_sessions: Arc::new(knot_xrpc::handover::HandoverSessions::default()), ci_logs: None, layout: layout.clone(), index: Arc::clone(&index), diff --git a/knot2/crates/knot-ssh/tests/ssh_push.rs b/knot2/crates/knot-ssh/tests/ssh_push.rs index d3d20957a..0e0bfd7dc 100644 --- a/knot2/crates/knot-ssh/tests/ssh_push.rs +++ b/knot2/crates/knot-ssh/tests/ssh_push.rs @@ -420,6 +420,7 @@ async fn spawn_server_core( let outbox = knot_xrpc::Outbox::open(&scan.path().join("meta"), Default::default()); let state = Arc::new(knot_xrpc::XrpcState { imports: knot_xrpc::imports::ImportQueue::channel().0, + handover_sessions: Arc::new(knot_xrpc::handover::HandoverSessions::default()), layout: layout.clone(), index: Arc::clone(&index), atproto: Arc::clone(&atproto), diff --git a/knot2/crates/knot-xrpc/src/forks.rs b/knot2/crates/knot-xrpc/src/forks.rs index 99ad14396..39632e112 100644 --- a/knot2/crates/knot-xrpc/src/forks.rs +++ b/knot2/crates/knot-xrpc/src/forks.rs @@ -243,6 +243,35 @@ pub(crate) fn ingest_pack( .map_err(|error| XrpcError::internal(error.to_string())) } +pub(crate) async fn pull_objects( + state: &XrpcState, + upstream: &Upstream, + repo: &RepoDid, + wants: Vec, + haves: Vec, + object_format: ObjectFormat, +) -> Result<(), XrpcError> { + let scratch = state.layout.scratch_dir().to_path_buf(); + std::fs::create_dir_all(&scratch).map_err(|error| XrpcError::internal(error.to_string()))?; + let pack = tempfile::Builder::new() + .prefix("pull-pack.") + .tempfile_in(&scratch) + .map_err(|error| XrpcError::internal(error.to_string()))?; + fetch_pack_into( + state, + upstream, + WantOids::new(wants.clone()), + HaveOids::new(haves.clone()), + object_format, + state.byte_limits.import_pack.get(), + pack.path(), + ) + .await?; + let layout = state.layout.clone(); + let target = repo.clone(); + run_blocking(move || ingest_pack(&layout.open(&target)?, pack.path(), &wants, &haves)).await +} + pub(crate) async fn fetch_commit( state: &XrpcState, repo: &RepoDid, @@ -270,25 +299,7 @@ pub(crate) async fn fetch_commit( Ok((haves, repo.object_format())) }) .await?; - let scratch = state.layout.scratch_dir().to_path_buf(); - std::fs::create_dir_all(&scratch).map_err(|error| XrpcError::internal(error.to_string()))?; - let pack = tempfile::Builder::new() - .prefix("patch-pack.") - .tempfile_in(&scratch) - .map_err(|error| XrpcError::internal(error.to_string()))?; - fetch_pack_into( - state, - &upstream, - WantOids::new(vec![commit]), - HaveOids::new(haves.clone()), - object_format, - state.byte_limits.import_pack.get(), - pack.path(), - ) - .await?; - let layout = state.layout.clone(); - let target = repo.clone(); - run_blocking(move || ingest_pack(&layout.open(&target)?, pack.path(), &[commit], &haves)).await + pull_objects(state, &upstream, repo, vec![commit], haves, object_format).await } #[derive(Clone, Copy, PartialEq, Eq, Debug, Deserialize, Serialize)] diff --git a/knot2/crates/knot-xrpc/src/foster.rs b/knot2/crates/knot-xrpc/src/foster.rs index 6bdda9d34..764841a88 100644 --- a/knot2/crates/knot-xrpc/src/foster.rs +++ b/knot2/crates/knot-xrpc/src/foster.rs @@ -13,7 +13,8 @@ use http::HeaderMap; use serde::{Deserialize, Serialize}; use knot_cob::{ - ChangePayload, CobHome, CobId, CobStore, Emits, KnotAuthored, Materialized, Record, + ChangePayload, CobHome, CobId, CobStore, Emits, HistoryModel, KnotAuthored, Materialized, + Record, }; use knot_git::Repo; use knot_index::Resolved; @@ -169,6 +170,11 @@ async fn seed_repo( "Too many records (max {MAX_RECORDS_PER_REQUEST})" ))); } + if input.records.is_empty() { + return Err(XrpcError::invalid_request( + "a seed includes at least one record. an empty shadow could never hand its repo back", + )); + } let SeedInput { name, owner_did: owner, @@ -348,6 +354,39 @@ pub(crate) fn fostered_objects( Ok(objects) } +pub(crate) struct SeededVersion { + pub(crate) address: RecordAddress, + pub(crate) object: CobId, + pub(crate) produces: RecordCid, +} + +pub(crate) fn seeded_versions(store: &CobStore) -> Result, XrpcError> { + let seed_type = + knot_types::TypeName::new(Seeded::TYPE).expect("the seed type is a valid type name"); + fostered_objects(store)? + .into_iter() + .map(|(address, object)| { + let changes = store.changes_of(&seed_type, object, HistoryModel::Convergent, None)?; + let tip = changes.changes.last().ok_or_else(|| { + XrpcError::internal(format!("the seeded graph for {address:?} has no changes")) + })?; + let seeded = Seeded::decode(tip.payload()) + .map_err(|error| XrpcError::internal(error.to_string()))?; + Ok(SeededVersion { + address, + object, + produces: seeded.produces, + }) + }) + .collect() +} + +pub(crate) fn history_of(type_name: &knot_types::TypeName) -> Option { + (type_name.as_str() == Seeded::TYPE) + .then_some(HistoryModel::Convergent) + .or_else(|| knot_cobs::history_model_of(type_name)) +} + #[allow(clippy::too_many_arguments)] fn seed_one( state: &XrpcState, diff --git a/knot2/crates/knot-xrpc/src/handover.rs b/knot2/crates/knot-xrpc/src/handover.rs new file mode 100644 index 000000000..bcb845450 --- /dev/null +++ b/knot2/crates/knot-xrpc/src/handover.rs @@ -0,0 +1,682 @@ +use std::collections::BTreeMap; +use std::collections::BTreeSet; +use std::path::Path; +use std::sync::Arc; + +use axum::Json; +use axum::body::Bytes; +use axum::extract::State; +use axum::response::{IntoResponse, Response}; +use base64::Engine as _; +use base64::engine::general_purpose::URL_SAFE_NO_PAD; +use http::{HeaderMap, StatusCode}; +use knot_acl::{KnotAcl, can_admin_knot}; +use knot_capability::Capability; +use knot_cob::{ChangeId, CobHome, CobStore, parse_cob_ref}; +use knot_git::{RefUpdate, Repo}; +use knot_index::Resolved; +use knot_record::chain; +use knot_runtime::{Clock, HttpRequest, HttpTransport, Signer}; +use knot_types::{ActorId, CobId, ObjectFormat, Oid, RefName, RepoDid, TypeName, UnixSeconds}; +use serde::{Deserialize, Serialize}; +use tokio_util::sync::CancellationToken; +use url::Url; + +use crate::XrpcState; +use crate::error::XrpcError; +use crate::forks::{self, Upstream}; +use crate::{decode, run_blocking}; + +pub const TRIGGER_ROUTE: &str = "/xrpc/org.tangled.temp.server.handoverRepo"; +pub const ACK_ROUTE: &str = "/xrpc/org.tangled.temp.server.ackHandover"; +pub const CANCEL_ROUTE: &str = "/xrpc/org.tangled.temp.server.cancelHandover"; +pub(crate) const SIDECAR: &str = "tangled.handover.json"; + +const COBS_PREFIX: &str = "refs/cobs/"; +const ACK_CONTEXT: &str = "org.tangled.temp.handover.ack.v1"; + +#[derive(Default)] +pub struct HandoverSessions { + running: std::sync::Mutex>, +} + +impl HandoverSessions { + fn locked(&self) -> std::sync::MutexGuard<'_, BTreeMap> { + self.running + .lock() + .unwrap_or_else(|poisoned| poisoned.into_inner()) + } + + pub fn acquire(self: &Arc, repo: &RepoDid) -> Option { + let mut running = self.locked(); + (!running.contains_key(repo)).then(|| { + let token = CancellationToken::new(); + running.insert(repo.clone(), token.clone()); + HandoverLease { + sessions: Arc::clone(self), + repo: repo.clone(), + token, + } + }) + } + + fn stop(&self, repo: &RepoDid) -> Option { + self.locked().get(repo).cloned() + } +} + +pub struct HandoverLease { + sessions: Arc, + repo: RepoDid, + token: CancellationToken, +} + +impl Drop for HandoverLease { + fn drop(&mut self) { + self.sessions.locked().remove(&self.repo); + } +} +#[derive(Clone, Debug, Serialize, Deserialize, PartialEq, Eq)] +pub struct FosterKnot(Url); + +impl FosterKnot { + pub fn new(raw: &str) -> Result { + let url = Url::parse(raw).map_err(|_| format!("foster url {raw} isn't a url"))?; + let trusted = matches!(url.scheme(), "https") + || matches!(url.scheme(), "http") && host_is_loopback(&url); + if !trusted { + return Err(format!( + "foster url {raw} must be https, or http on loopback. a foster knot is \ + trusted over tls or not at all" + )); + } + Ok(Self(url)) + } + + fn url(&self, path: &str) -> Url { + let mut url = self.0.clone(); + url.path_segments_mut() + .expect("an http url always has segments") + .extend(path.split('/')); + url + } +} + +fn host_is_loopback(url: &Url) -> bool { + match url.host() { + Some(url::Host::Domain(name)) => name.eq_ignore_ascii_case("localhost"), + Some(url::Host::Ipv4(ip)) => ip.is_loopback(), + Some(url::Host::Ipv6(ip)) => ip.is_loopback(), + None => false, + } +} + +#[derive(Serialize, Deserialize, Clone, Debug)] +pub(crate) struct HandoverJob { + repo: RepoDid, + foster: FosterKnot, + triggered_at: UnixSeconds, + #[serde(default, skip_serializing_if = "Option::is_none")] + error: Option, +} + +impl HandoverJob { + fn read(repo_path: &Path) -> Option> { + std::fs::read(repo_path.join(SIDECAR)) + .ok() + .map(|bytes| serde_json::from_slice(&bytes)) + } + + fn write(&self, repo_path: &Path) -> std::io::Result<()> { + let staged = tempfile::Builder::new() + .prefix(&format!("{SIDECAR}.tmp.")) + .tempfile_in(repo_path)?; + std::fs::write(staged.path(), serde_json::to_vec(self)?)?; + std::fs::rename(staged.path(), repo_path.join(SIDECAR)) + } +} + +#[derive(Deserialize)] +struct HandoverInput { + #[serde(rename = "repoDid")] + repo_did: RepoDid, + #[serde(rename = "fosterUrl")] + foster_url: String, +} + +pub(crate) async fn trigger( + State(state): State>>, + headers: HeaderMap, + method: crate::Method, + body: Bytes, +) -> Result { + let actor = state.authenticate(&headers, &method).await?; + let acl = KnotAcl::new(&state.admins, state.admission, &state.index); + if !can_admin_knot(&acl, &actor).is_allowed() { + return Err(XrpcError::forbidden( + "only a knot admin may pull a repo back from its foster knot", + )); + } + let input: HandoverInput = decode(&body)?; + crate::reads::require_hosted(&state, input.repo_did.clone())?; + let foster = FosterKnot::new(&input.foster_url).map_err(XrpcError::invalid_request)?; + let lease = state + .handover_sessions + .acquire(&input.repo_did) + .ok_or_else(|| { + XrpcError::conflict("a handover for this repo is already running on this knot") + })?; + let repo_path = state + .layout + .repo_path(&input.repo_did) + .map_err(|error| XrpcError::internal(error.to_string()))?; + HandoverJob { + repo: input.repo_did.clone(), + foster: foster.clone(), + triggered_at: state.now(), + error: None, + } + .write(&repo_path) + .map_err(|error| XrpcError::internal(format!("writing the handover job: {error}")))?; + tokio::spawn(drive(Arc::clone(&state), input.repo_did.clone(), lease)); + Ok(( + StatusCode::OK, + Json(serde_json::json!({ + "repoDid": input.repo_did, + "state": "running", + })), + ) + .into_response()) +} + +#[derive(Deserialize)] +struct CancelInput { + #[serde(rename = "repoDid")] + repo_did: RepoDid, +} + +pub(crate) async fn cancel( + State(state): State>>, + headers: HeaderMap, + method: crate::Method, + body: Bytes, +) -> Result { + let actor = state.authenticate(&headers, &method).await?; + let acl = KnotAcl::new(&state.admins, state.admission, &state.index); + if !can_admin_knot(&acl, &actor).is_allowed() { + return Err(XrpcError::forbidden( + "only a knot admin may abandon a handover job", + )); + } + let input: CancelInput = decode(&body)?; + if let Some(running) = state.handover_sessions.stop(&input.repo_did) { + running.cancel(); + } + let repo_path = state + .layout + .repo_path(&input.repo_did) + .map_err(|error| XrpcError::internal(error.to_string()))?; + let sidecar = repo_path.join(SIDECAR); + if !sidecar.is_file() { + return Err(XrpcError::not_found(format!( + "{} has no handover job to abandon", + input.repo_did + ))); + } + std::fs::remove_file(&sidecar) + .map_err(|error| XrpcError::internal(format!("removing the handover job: {error}")))?; + tracing::info!( + repo = %input.repo_did, + "the handover job is abandoned. a live pull stops at its next step" + ); + Ok(( + StatusCode::OK, + Json(serde_json::json!({ "repoDid": input.repo_did })), + ) + .into_response()) +} + +async fn drive( + state: Arc>, + repo: RepoDid, + lease: HandoverLease, +) { + tokio::select! { + () = lease.token.cancelled() => tracing::info!(repo = %repo, "the handover was canceled. its job is abandoned"), + outcome = finish(&state, &repo) => match outcome { + Ok(records) => tracing::info!(repo = %repo, records, "the handover finished. every fostered record of this repo is home"), + Err(error) => tracing::error!(repo = %repo, %error, "the handover didn't complete. its job stays for a replay"), + }, + } +} + +pub(crate) async fn finish( + state: &Arc>, + repo: &RepoDid, +) -> Result { + let repo_path = state + .layout + .repo_path(repo) + .map_err(|error| XrpcError::internal(error.to_string()))?; + let job = match HandoverJob::read(&repo_path) { + Some(Ok(job)) => job, + Some(Err(error)) => { + return Err(XrpcError::internal(format!( + "the handover sidecar for {repo} wouldn't decode: {error}" + ))); + } + None => { + return Err(XrpcError::not_found(format!( + "{repo} has no handover pending" + ))); + } + }; + match handover(state, repo, &job.foster).await { + Ok(records) => { + std::fs::remove_file(repo_path.join(SIDECAR)).map_err(|error| { + XrpcError::internal(format!("retiring the finished handover job: {error}")) + })?; + Ok(records) + } + Err(error) => { + HandoverJob { + error: Some(error.to_string()), + ..job + } + .write(&repo_path) + .map_err(|write_error| { + XrpcError::internal(format!("recording the failed handover: {write_error}")) + })?; + Err(error) + } + } +} + +async fn handover( + state: &Arc>, + repo: &RepoDid, + foster: &FosterKnot, +) -> Result { + let key = foster_key(state, foster).await?; + let upstream = Upstream::Remote(foster.url(repo.as_str())); + let (object_format, advertised) = advertised_cobs(state, &upstream).await?; + if advertised.is_empty() { + return Err(XrpcError::bad_gateway(format!( + "the foster knot hosts {repo} with no cobs to its name, so an empty shadow \ + can't hand anything back" + ))); + } + + let layout = state.layout.clone(); + let target = repo.clone(); + let wanted = advertised.clone(); + let (local_format, local, missing) = run_blocking(move || { + let git = layout.open(&target)?; + let local: BTreeMap = git + .references()? + .into_iter() + .map(|record| (record.name.as_str().to_owned(), record.target)) + .collect(); + let missing: BTreeSet = wanted + .iter() + .filter(|(type_name, object, tip)| local.get(&cob_ref(type_name, *object)) != Some(tip)) + .map(|(_, _, tip)| *tip) + .collect(); + Ok::<_, XrpcError>((git.object_format(), local, missing)) + }) + .await?; + if local_format != object_format { + return Err(XrpcError::invalid_request( + "the foster knot and this knot disagree on the repo's object format", + )); + } + let wants: Vec = missing.into_iter().collect(); + let haves: Vec = local.values().copied().collect(); + if !wants.is_empty() { + forks::pull_objects(state, &upstream, repo, wants, haves, object_format).await?; + } + let settled = Arc::clone(state); + let target = repo.clone(); + let graphs = advertised.clone(); + let placed = local; + let seeded = run_blocking(move || { + let _guard = settled.cob_locks.repo(&target); + let git = settled.layout.open(&target)?; + let store = CobStore::new(&git); + let home = CobHome::from(&target); + for (type_name, object, tip) in &graphs { + let history = crate::foster::history_of(type_name).ok_or_else(|| { + XrpcError::bad_gateway(format!( + "the foster knot serves {} cobs whose history model this knot doesn't know", + type_name.as_str() + )) + })?; + store.verify_type_from_tip( + &home, + type_name, + *object, + ChangeId::new(*tip), + history, + &key, + )?; + } + apply_cob_refs(&git, &graphs, &placed)?; + let signer = settled.secrets.signer(&settled.knot_did)?; + crate::social::reconcile(&settled, &git, &target, &signer, settled.now())?; + let seeded = crate::foster::seeded_versions(&store)?; + confirm_local_leaves(&git, &seeded)?; + Ok(seeded) + }) + .await?; + confirm_foster_serving(state, foster, repo, &seeded).await?; + send_ack(state, foster, repo, seeded.len(), state.now()).await?; + Ok(seeded.len()) +} + +fn cob_ref(type_name: &TypeName, object: CobId) -> String { + format!("{COBS_PREFIX}{type_name}/{object}") +} + +async fn advertised_cobs( + state: &XrpcState, + upstream: &Upstream, +) -> Result<(ObjectFormat, Vec<(TypeName, CobId, Oid)>), XrpcError> { + let refs = forks::upstream_refs(state, upstream, vec![COBS_PREFIX.to_string()]).await?; + let parsed = refs + .refs + .iter() + .filter(|record| record.name.as_str().starts_with(COBS_PREFIX)) + .map(|record| match parse_cob_ref(record.name.as_str()) { + Some((type_name, object)) => Ok((type_name, object, record.target)), + None => Err(XrpcError::bad_gateway(format!( + "the foster knot serves a ref outside the cob ref rules: {}", + record.name.as_str() + ))), + }) + .collect::, XrpcError>>()?; + Ok((refs.object_format, parsed)) +} + +fn apply_cob_refs( + git: &Repo, + graphs: &[(TypeName, CobId, Oid)], + placed: &BTreeMap, +) -> Result<(), XrpcError> { + let updates = graphs + .iter() + .filter_map(|(type_name, object, tip)| { + let name = cob_ref(type_name, *object); + let parsed = RefName::new(&name).expect("a served cob ref name parses"); + match placed.get(&name) { + None => Some(Ok(RefUpdate::Create { + name: parsed, + new: *tip, + })), + Some(&old) if old == *tip => None, + Some(&old) => Some(git.merge_base(old, *tip).map_err(XrpcError::from).and_then( + |base| { + (base == Some(old)) + .then_some(RefUpdate::Update { + name: parsed, + old, + new: *tip, + }) + .ok_or_else(|| { + XrpcError::bad_gateway(format!( + "the foster knot serves {name} at {tip}, which isn't \ + ahead of the local {old}" + )) + }) + }, + )), + } + }) + .collect::, XrpcError>>()?; + if !updates.is_empty() { + git.update_refs(&updates)?; + } + Ok(()) +} + +fn confirm_local_leaves( + git: &Repo, + seeded: &[crate::foster::SeededVersion], +) -> Result<(), XrpcError> { + let leaves = futures::executor::block_on(chain::leaves(git))?; + seeded.iter().try_for_each(|version| { + (leaves.get(&version.address.mst_key()) == Some(&version.produces.cid())) + .then_some(()) + .ok_or_else(|| { + XrpcError::internal(format!( + "the rebuilt chain doesn't serve {} at {} the way the foster knot's \ + last change recorded it", + version.address.mst_key().as_str(), + version.produces.cid() + )) + }) + }) +} + +async fn foster_key( + state: &Arc>, + foster: &FosterKnot, +) -> Result { + let described = get_json( + state.git_http.as_ref(), + foster.url("xrpc/org.tangled.knot.describeKnot"), + ) + .await?; + let did = described["did"].as_str().ok_or_else(|| { + XrpcError::bad_gateway("describeKnot on the foster knot leaves out its own did") + })?; + let fostering = described["capabilities"] + .as_array() + .is_some_and(|capabilities| { + capabilities + .iter() + .any(|token| token.as_str() == Some(Capability::Foster.token())) + }); + if !fostering { + return Err(XrpcError::invalid_request(format!( + "{did} doesn't take fostered repos" + ))); + } + let document = get_json(state.git_http.as_ref(), foster.url(".well-known/did.json")).await?; + if document["id"].as_str() != Some(did) { + return Err(XrpcError::bad_gateway(format!( + "the did document served by {} belongs to a different knot than the one that \ + served describeKnot", + foster.0.as_str() + ))); + } + let wanted = format!("{did}#{}", knot_atproto::ATPROTO_METHOD); + let multibase = document["verificationMethod"] + .as_array() + .and_then(|methods| { + methods + .iter() + .find(|method| method["id"].as_str() == Some(wanted.as_str())) + }) + .and_then(|method| method["publicKeyMultibase"].as_str()) + .ok_or_else(|| { + XrpcError::bad_gateway( + "the foster knot's did document publishes no multikey under #atproto", + ) + })?; + ActorId::new(multibase) + .map_err(|_| XrpcError::bad_gateway("the foster knot's multikey doesn't parse as a key")) +} + +async fn confirm_foster_serving( + state: &Arc>, + foster: &FosterKnot, + repo: &RepoDid, + seeded: &[crate::foster::SeededVersion], +) -> Result<(), XrpcError> { + for version in seeded { + let mut url = foster.url("xrpc/com.atproto.repo.getRecord"); + url.query_pairs_mut() + .append_pair("repo", repo.as_str()) + .append_pair("collection", version.address.collection().as_str()) + .append_pair("rkey", version.address.rkey().as_str()); + let served = get_json(state.git_http.as_ref(), url).await?; + let cid = served["cid"].as_str().ok_or_else(|| { + XrpcError::bad_gateway(format!( + "the foster knot served getRecord for {} without a cid", + version.address.at_uri(repo) + )) + })?; + if cid != version.produces.cid().to_string() { + return Err(XrpcError::internal(format!( + "the foster knot serves {} at {cid} while its own graph recorded {}", + version.address.at_uri(repo), + version.produces.cid() + ))); + } + } + Ok(()) +} + +#[derive(Serialize)] +struct AckSigned<'a> { + context: &'static str, + #[serde(rename = "repoDid")] + repo_did: &'a RepoDid, + records: usize, + #[serde(rename = "ackedAt")] + acked_at: UnixSeconds, +} + +#[derive(Serialize)] +struct AckBody<'a> { + #[serde(flatten)] + signed: AckSigned<'a>, + signature: String, +} + +async fn send_ack( + state: &Arc>, + foster: &FosterKnot, + repo: &RepoDid, + records: usize, + now: UnixSeconds, +) -> Result<(), XrpcError> { + let signer = state.secrets.signer(&state.knot_did)?; + let signed = AckSigned { + context: ACK_CONTEXT, + repo_did: repo, + records, + acked_at: now, + }; + let bytes = serde_ipld_dagcbor::to_vec(&signed) + .map_err(|error| XrpcError::internal(error.to_string()))?; + let ack = AckBody { + signed, + signature: URL_SAFE_NO_PAD.encode(signer.sign(&bytes).as_bytes()), + }; + let body = serde_json::to_vec(&ack) + .map_err(|error| XrpcError::internal(error.to_string()))? + .into(); + let response = state + .git_http + .execute(HttpRequest::post( + foster.url(ACK_ROUTE.trim_start_matches('/')), + body, + )) + .await + .map_err(|error| { + XrpcError::bad_gateway(format!("the foster knot wouldn't take the ack: {error}")) + })?; + if response.status.is_success() { + return Ok(()); + } + Err(XrpcError::bad_gateway(format!( + "the foster knot refused the ack with {}: {}", + response.status.as_u16(), + String::from_utf8_lossy(&response.body) + ))) +} + +async fn get_json(http: &dyn HttpTransport, url: Url) -> Result { + let response = http.execute(HttpRequest::get(url)).await.map_err(|error| { + XrpcError::bad_gateway(format!("the foster knot is unreachable: {error}")) + })?; + if !response.status.is_success() { + return Err(XrpcError::bad_gateway(format!( + "the foster knot returned {} where a document belonged", + response.status.as_u16() + ))); + } + serde_json::from_slice(&response.body).map_err(|error| { + XrpcError::bad_gateway(format!( + "the foster knot returned bytes that aren't json: {error}" + )) + }) +} + +pub fn recover(state: &XrpcState) -> Vec { + crate::imports::child_dirs(state.layout.scratch_dir()) + .into_iter() + .flat_map(|method| crate::imports::child_dirs(&method)) + .flat_map(|shard| crate::imports::child_dirs(&shard)) + .filter_map(|repo_path| match HandoverJob::read(&repo_path) { + Some(Ok(job)) => matches!(state.index.owner_of(&job.repo), Resolved::Ready(Some(_))) + .then_some(job.repo), + Some(Err(error)) => { + tracing::error!(path = %repo_path.display(), %error, "couldn't decode the handover sidecar; leaving it alone"); + None + } + None => None, + }) + .collect() +} + +pub async fn run_boot( + state: Arc>, + shutdown: CancellationToken, +) { + let pending = recover(&state); + if pending.is_empty() { + return; + } + tracing::info!( + repos = pending.len(), + "handover jobs that a crash left half-finished will now replay" + ); + for repo in pending { + let Some(lease) = state.handover_sessions.acquire(&repo) else { + tracing::info!(repo = %repo, "skipping the handover replay because another session holds this repo"); + continue; + }; + tokio::select! { + () = shutdown.cancelled() => return, + () = drive(Arc::clone(&state), repo, lease) => {} + } + } +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn a_foster_url_takes_https_or_loopback_http_with_a_host() { + assert!(FosterKnot::new("https://foster.nel.pet").is_ok()); + assert!(FosterKnot::new("http://127.0.0.1:7433").is_ok()); + assert!(FosterKnot::new("http://localhost:9110").is_ok()); + assert!(FosterKnot::new("http://foster.nel.pet").is_err()); + assert!(FosterKnot::new("ftp://foster.nel.pet").is_err()); + assert!(FosterKnot::new("not a url").is_err()); + assert!(FosterKnot::new("https://").is_err()); + } + + #[test] + fn a_repo_url_hangs_off_the_foster_base() { + let foster = FosterKnot::new("https://foster.nel.pet/").unwrap(); + let repo = RepoDid::new("did:plc:feralpearl").unwrap(); + assert_eq!( + foster.url(repo.as_str()).as_str(), + "https://foster.nel.pet/did:plc:feralpearl" + ); + } +} diff --git a/knot2/crates/knot-xrpc/src/imports.rs b/knot2/crates/knot-xrpc/src/imports.rs index 68298fd41..fadd34713 100644 --- a/knot2/crates/knot-xrpc/src/imports.rs +++ b/knot2/crates/knot-xrpc/src/imports.rs @@ -353,7 +353,7 @@ pub(crate) async fn run_one( } } -fn child_dirs(path: &Path) -> Vec { +pub(crate) fn child_dirs(path: &Path) -> Vec { std::fs::read_dir(path) .into_iter() .flatten() diff --git a/knot2/crates/knot-xrpc/src/lib.rs b/knot2/crates/knot-xrpc/src/lib.rs index a3e5b4a2d..b634ee58d 100644 --- a/knot2/crates/knot-xrpc/src/lib.rs +++ b/knot2/crates/knot-xrpc/src/lib.rs @@ -11,6 +11,7 @@ mod forks; pub mod foster; mod git; pub mod grants; +pub mod handover; pub mod imports; pub mod invite_codes; pub mod legacy_admin; @@ -213,6 +214,7 @@ pub struct XrpcState { pub appview: AppviewEndpoint, pub slots: Slots, pub imports: Arc, + pub handover_sessions: Arc, pub lfs: Option, pub blobs: Arc, pub catalog: Arc, @@ -391,6 +393,8 @@ pub fn router(state: Arc>) -> Router ) .route(repos::CREATE_ROUTE, post(repos::create_repo::)) .route(repos::DELETE_ROUTE, post(repos::delete_repo::)) + .route(handover::TRIGGER_ROUTE, post(handover::trigger::)) + .route(handover::CANCEL_ROUTE, post(handover::cancel::)) .route(repos::RESERVE_ROUTE, post(repos::reserve_key::)) .route(mint::MINT_TOKEN_ROUTE, post(mint::mint_token::)) .route( diff --git a/knot2/crates/knot-xrpc/src/social.rs b/knot2/crates/knot-xrpc/src/social.rs index 55124a9e7..ebe94fcf1 100644 --- a/knot2/crates/knot-xrpc/src/social.rs +++ b/knot2/crates/knot-xrpc/src/social.rs @@ -955,7 +955,7 @@ pub(crate) fn reconcile( .collect(); let store = CobStore::new(git); let standing = futures::executor::block_on(chain::leaves(git))?; - let collections = knot_cobs::social_collections(); + let mut collections: Vec = knot_cobs::social_collections().to_vec(); let mut records = serving .iter() .try_fold(Kept::default(), |kept, (entry, version)| { @@ -974,7 +974,10 @@ pub(crate) fn reconcile( None => kept, }) })?; - for (key, due) in fostered_versions(&store, &standing, &collections)? { + for (key, collection, due) in fostered_versions(&store, &standing)? { + if !collections.contains(&collection) { + collections.push(collection); + } if !records.due.contains_key(&key) { records = records.taking(key, due); } @@ -1001,38 +1004,25 @@ pub(crate) fn reconcile( fn fostered_versions( store: &CobStore, leaves: &BTreeMap, - collections: &[RecordCollection], -) -> Result, XrpcError> { +) -> Result, XrpcError> { let seed_type = knot_types::TypeName::new(crate::foster::Seeded::TYPE) .expect("The seed type is a valid type name"); - let mut keeps = Vec::new(); - for (address, object) in crate::foster::fostered_objects(store)? { - if !collections.contains(address.collection()) { - continue; - } - let changes = store - .changes_of(&seed_type, object, knot_cob::HistoryModel::Convergent, None)? - .changes; - let Some(change) = changes.last() else { - continue; - }; - let seeded = crate::foster::Seeded::decode(change.payload()) - .map_err(|error| XrpcError::internal(error.to_string()))?; - let Some(produces) = seeded.emits().version() else { - continue; - }; - let key = address.mst_key(); - let due = resolve_due( - store, - &seed_type, - object, - produces, - leaves.get(&key).copied(), - None, - )?; - keeps.extend(due.map(|due| (key, due))); - } - Ok(keeps) + crate::foster::seeded_versions(store)? + .into_iter() + .map(|version| { + let key = version.address.mst_key(); + let due = resolve_due( + store, + &seed_type, + version.object, + version.produces, + leaves.get(&key).copied(), + None, + )?; + Ok::<_, XrpcError>(due.map(|due| (key, version.address.collection().clone(), due))) + }) + .collect::, XrpcError>>() + .map(|dues| dues.into_iter().flatten().collect()) } enum Doing<'a> { @@ -1078,6 +1068,18 @@ pub(crate) struct Contributor { pub(crate) session: Session, } +fn require_native( + state: &XrpcState, + repo: &RepoDid, +) -> Result<(), XrpcError> { + if ready(state.index.is_fostered(repo), "registry")? { + return Err(XrpcError::forbidden( + "this repo is only fostered here. its records change on its origin knot", + )); + } + Ok(()) +} + pub(crate) async fn session( state: &XrpcState, headers: &HeaderMap, @@ -1087,6 +1089,7 @@ pub(crate) async fn session( let actor = state .authenticate_write_for_repo(headers, method, &repo) .await?; + require_native(state, &repo)?; let permission = permission(state, &actor, &repo).await?; Ok(Session { actor, @@ -2043,9 +2046,11 @@ async fn grant_actor( collection: &RecordCollection, ) -> Result { if collection.as_str() == knot_record::grant::COLLABORATOR_COLLECTION { - state + let actor = state .authenticate_write_for_repo(headers, method, repo) - .await + .await?; + require_native(state, repo)?; + Ok(actor) } else { on_this_knot(repo, &state.knot_repo, "record")?; if let Some(actor) = state.legacy_admin_actor(headers)? { diff --git a/knot2/crates/knot-xrpc/src/tests.rs b/knot2/crates/knot-xrpc/src/tests.rs index 8ccb89e04..42cc7a7bc 100644 --- a/knot2/crates/knot-xrpc/src/tests.rs +++ b/knot2/crates/knot-xrpc/src/tests.rs @@ -348,6 +348,7 @@ fn state_from( let state = Arc::new(XrpcState { blobs: Default::default(), imports: crate::imports::ImportQueue::channel().0, + handover_sessions: Arc::new(crate::handover::HandoverSessions::default()), layout, index, atproto, diff --git a/knot2/crates/knot-xrpc/tests/atproto_interop.rs b/knot2/crates/knot-xrpc/tests/atproto_interop.rs index 92c7a918a..0a7607d54 100644 --- a/knot2/crates/knot-xrpc/tests/atproto_interop.rs +++ b/knot2/crates/knot-xrpc/tests/atproto_interop.rs @@ -280,16 +280,32 @@ async fn a_fostering_knot_says_so_in_its_capabilities() { ); } +fn foster_seed_body(repo_did: &str) -> serde_json::Value { + serde_json::json!({ + "name": "boltfulboy", + "ownerDid": FOSTER_OWNER, + "repoDid": repo_did, + "records": [{ + "collection": "org.tangled.track.ticket", + "rkey": FOSTER_RKEY, + "author": FOSTER_OWNER, + "createdAt": 1100, + "record": { + "$type": "org.tangled.track.ticket", + "title": {"$type": "org.tangled.markup.markdown#inline", "text": "a feral pearl"}, + "state": "feral", + "createdAt": FOSTER_CREATED_AT, + "x-tngl-editor": FOSTER_OWNER, + }, + }], + }) +} + #[tokio::test] async fn reseeding_under_a_different_repo_did_is_refused() { let world = World::fostering(FOSTER_SECRET); let credentials = common::basic_auth("admin", FOSTER_SECRET); - let seed = serde_json::json!({ - "name": "boltfulboy", - "ownerDid": FOSTER_OWNER, - "repoDid": FOSTER_REPO_DID, - "records": [], - }); + let seed = foster_seed_body(FOSTER_REPO_DID); let (status, _) = common::post_raw( &world, "/xrpc/org.tangled.temp.server.seedRepo", @@ -298,12 +314,7 @@ async fn reseeding_under_a_different_repo_did_is_refused() { ) .await; assert_eq!(status, StatusCode::OK); - let impostor = serde_json::json!({ - "name": "boltfulboy", - "ownerDid": FOSTER_OWNER, - "repoDid": "did:plc:somebodyelses", - "records": [], - }); + let impostor = foster_seed_body("did:plc:somebodyelses"); let (status, body) = common::post_raw( &world, "/xrpc/org.tangled.temp.server.seedRepo", diff --git a/knot2/crates/knot-xrpc/tests/common/mod.rs b/knot2/crates/knot-xrpc/tests/common/mod.rs index c9ebc5352..9300d374a 100644 --- a/knot2/crates/knot-xrpc/tests/common/mod.rs +++ b/knot2/crates/knot-xrpc/tests/common/mod.rs @@ -1,9 +1,12 @@ #![allow(dead_code)] pub async fn serve(world: &World) -> std::net::SocketAddr { + serve_router(world.router.clone()).await +} + +pub async fn serve_router(router: axum::Router) -> std::net::SocketAddr { let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap(); let addr = listener.local_addr().unwrap(); - let router = world.router.clone(); tokio::spawn(async move { axum::serve( listener, @@ -15,17 +18,6 @@ pub async fn serve(world: &World) -> std::net::SocketAddr { addr } -async fn serve_router(router: axum::Router) -> std::net::SocketAddr { - let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap(); - let addr = listener.local_addr().unwrap(); - tokio::spawn(async move { - axum::serve(listener, router.into_make_service()) - .await - .unwrap(); - }); - addr -} - fn knot_multikey(world: &World) -> ActorId { let signer = world.state.secrets.signer(&world.state.knot_did).unwrap(); let sec1 = knot_runtime::Signer::public_key(&signer) @@ -95,7 +87,7 @@ pub async fn serve_bob_pds( .await } -pub async fn serve_pushes(world: &World) -> std::net::SocketAddr { +pub fn pack_edge(world: &World) -> Router { let resolver: Arc = { let index = Arc::clone(&world.state.index); Arc::new(move |target: &knot_pack::RepoTarget| match target { @@ -116,21 +108,14 @@ pub async fn serve_pushes(world: &World) -> std::net::SocketAddr { Arc::new(knot_runtime::SystemClock), ) }); - let router = advertisement + advertisement .into_router() .merge(write) - .merge(world.router.clone()); - let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap(); - let addr = listener.local_addr().unwrap(); - tokio::spawn(async move { - axum::serve( - listener, - router.into_make_service_with_connect_info::(), - ) - .await - .unwrap(); - }); - addr + .merge(world.router.clone()) +} + +pub async fn serve_pushes(world: &World) -> std::net::SocketAddr { + serve_router(pack_edge(world)).await } pub async fn http_push( @@ -438,6 +423,7 @@ struct WorldSetup { lfs_enabled: bool, policy: knot_types::ContributionPolicy, legacy_admin_secret: Option, + git_http: Option>, } impl Default for WorldSetup { @@ -447,6 +433,7 @@ impl Default for WorldSetup { lfs_enabled: true, policy: knot_types::ContributionPolicy::Anyone, legacy_admin_secret: None, + git_http: None, } } } @@ -502,45 +489,43 @@ impl World { ) } - pub fn administered() -> Self { + fn tailored(setup: WorldSetup) -> Self { Self::assemble( true, ByteLimits::default(), ObjectFormat::SHA1, LimitConfig::default(), - WorldSetup { - admins: BTreeSet::from([AccountDid::new(OWNER).unwrap()]), - ..WorldSetup::default() - }, + setup, ) } + pub fn administered() -> Self { + Self::tailored(WorldSetup { + admins: BTreeSet::from([owner()]), + ..WorldSetup::default() + }) + } + pub fn fostering(secret: &str) -> Self { - Self::assemble( - true, - ByteLimits::default(), - ObjectFormat::SHA1, - LimitConfig::default(), - WorldSetup { - legacy_admin_secret: Some( - knot_xrpc::legacy_admin::LegacyAdminSecret::new(secret).unwrap(), - ), - ..WorldSetup::default() - }, - ) + Self::tailored(WorldSetup { + legacy_admin_secret: knot_xrpc::legacy_admin::LegacyAdminSecret::new(secret).ok(), + ..WorldSetup::default() + }) + } + + pub fn pulling(git_http: Arc) -> Self { + Self::tailored(WorldSetup { + admins: BTreeSet::from([owner()]), + git_http: Some(git_http), + ..WorldSetup::default() + }) } pub fn under(policy: knot_types::ContributionPolicy) -> Self { - Self::assemble( - true, - ByteLimits::default(), - ObjectFormat::SHA1, - LimitConfig::default(), - WorldSetup { - policy, - ..WorldSetup::default() - }, - ) + Self::tailored(WorldSetup { + policy, + ..WorldSetup::default() + }) } fn build(rebuilt: bool, byte_limits: ByteLimits, object_format: ObjectFormat) -> Self { @@ -548,16 +533,10 @@ impl World { } pub fn lfs_less() -> Self { - Self::assemble( - true, - ByteLimits::default(), - ObjectFormat::SHA1, - LimitConfig::default(), - WorldSetup { - lfs_enabled: false, - ..WorldSetup::default() - }, - ) + Self::tailored(WorldSetup { + lfs_enabled: false, + ..WorldSetup::default() + }) } fn build_with_limits( @@ -587,6 +566,7 @@ impl World { lfs_enabled, policy, legacy_admin_secret, + git_http, } = setup; let dir = tempfile::tempdir().unwrap(); let scan_path = dir.path().join("repos"); @@ -677,6 +657,7 @@ impl World { let outbox = knot_xrpc::Outbox::open(&meta_path, Default::default()); let state = Arc::new(XrpcState { imports, + handover_sessions: Arc::new(knot_xrpc::handover::HandoverSessions::default()), layout: layout.clone(), index, atproto, @@ -710,11 +691,13 @@ impl World { proxy_trust: knot_types::ProxyTrust::default(), byte_limits, budgets: Budgets::default(), - git_http: Arc::new(FakeHttp::new(|_request: &HttpRequest| { - Err(NetworkError::Connect( - "no git upstream is served in this test".to_string(), - )) - })), + git_http: git_http.unwrap_or_else(|| { + Arc::new(FakeHttp::new(|_request: &HttpRequest| { + Err(NetworkError::Connect( + "no git upstream is served in this test".to_string(), + )) + })) + }), pack_limits: knot_pack::PackLimits::default(), service_owner: AccountDid::new(OWNER).unwrap(), legacy_admin_secret, diff --git a/knot2/crates/knot-xrpc/tests/handover.rs b/knot2/crates/knot-xrpc/tests/handover.rs new file mode 100644 index 000000000..d436f72b3 --- /dev/null +++ b/knot2/crates/knot-xrpc/tests/handover.rs @@ -0,0 +1,758 @@ +mod common; + +use std::collections::BTreeMap; +use std::sync::Arc; + +use base64::Engine as _; +use base64::engine::general_purpose::URL_SAFE_NO_PAD; +use common::{World, nsid, owner, pack_edge, post_authed, serve_router}; +use http::StatusCode; +use knot_cob::{ChangePayload, CobHome, CobStore, KnotAuthored}; +use knot_git::RefUpdate; +use knot_runtime::{HttpLimits, ReqwestHttp, Signature, Signer as _}; +use knot_types::{AccountDid, Oid, RefName, RepoDid, RepoName, RepoRkey, ServiceDid, UnixSeconds}; +use parking_lot::Mutex; +use tokio_util::sync::CancellationToken; + +const FOSTER_SECRET: &str = "foster-pearl"; +const FOSTER_OWNER: &str = "did:plc:boltfulboy"; +const REPO: &str = "did:plc:feralpearl"; +const COLLECTION: &str = "org.tangled.track.ticket"; +const RKEY: &str = "the-feral-pearl-ledger"; +const CREATED_AT: &str = "1970-01-01T00:18:20Z"; +const ACK_PATH: &str = "/xrpc/org.tangled.temp.server.ackHandover"; +const HANDOVER: &str = "org.tangled.temp.server.handoverRepo"; +const CANCEL: &str = "org.tangled.temp.server.cancelHandover"; +const MARKDOWN: &str = "org.tangled.markup.markdown"; +const NOTES_RKEY: &str = "pearl-notes"; + +#[derive(Clone)] +struct AckSink { + bodies: Arc>>, + refuse_first: bool, +} + +impl AckSink { + fn len(&self) -> usize { + self.bodies.lock().len() + } +} + +fn ok_sink() -> AckSink { + AckSink { + bodies: Arc::new(Mutex::new(Vec::new())), + refuse_first: false, + } +} + +fn ack_edge(sink: AckSink) -> axum::Router { + axum::Router::new().route( + ACK_PATH, + axum::routing::post(move |body: axum::body::Bytes| async move { + let posted: serde_json::Value = serde_json::from_slice(&body).unwrap(); + let seen = sink.len(); + sink.bodies.lock().push(posted); + let status = match sink.refuse_first && seen == 0 { + true => StatusCode::INTERNAL_SERVER_ERROR, + false => StatusCode::OK, + }; + (status, "{}") + }), + ) +} + +fn did_edge(document: serde_json::Value) -> axum::Router { + axum::Router::new().route( + "/.well-known/did.json", + axum::routing::get(move || async move { axum::Json(document.clone()) }), + ) +} + +async fn serve_foster(world: &World, sink: AckSink) -> std::net::SocketAddr { + serve_router(ack_edge(sink).fallback_service(pack_edge(world))).await +} + +fn fostering() -> World { + World::fostering(FOSTER_SECRET) +} + +fn upgrading() -> World { + let outbound = ReqwestHttp::new(HttpLimits { + block_private_addresses: false, + ..HttpLimits::default() + }) + .unwrap(); + World::pulling(Arc::new(outbound)) +} + +fn repo_did() -> RepoDid { + RepoDid::new(REPO).unwrap() +} + +fn ticket_record(text: &str) -> serde_json::Value { + serde_json::json!({ + "collection": COLLECTION, + "rkey": RKEY, + "author": FOSTER_OWNER, + "createdAt": 1100, + "record": { + "$type": COLLECTION, + "title": {"$type": "org.tangled.markup.markdown#inline", "text": text}, + "state": "feral", + "createdAt": CREATED_AT, + "x-tngl-editor": FOSTER_OWNER, + }, + }) +} + +fn markdown_record() -> serde_json::Value { + serde_json::json!({ + "collection": MARKDOWN, + "rkey": NOTES_RKEY, + "author": FOSTER_OWNER, + "createdAt": 1200, + "record": { + "$type": MARKDOWN, + "text": "pearl notes from outside the social collections", + }, + }) +} + +async fn post_seed(world: &World, records: serde_json::Value) -> (StatusCode, serde_json::Value) { + let (status, body) = common::post_raw( + world, + "/xrpc/org.tangled.temp.server.seedRepo", + common::basic_auth("admin", FOSTER_SECRET), + serde_json::json!({ + "name": "feralpearl", + "ownerDid": FOSTER_OWNER, + "repoDid": REPO, + "defaultBranch": "main", + "records": records, + }), + ) + .await; + let parsed = serde_json::from_slice(&body).unwrap_or(serde_json::Value::Null); + (status, parsed) +} + +async fn seed_foster_record(world: &World) { + let (status, body) = + post_seed(world, serde_json::json!([ticket_record("a feral pearl")])).await; + assert_eq!(status, StatusCode::OK, "the seed must succeed: {body}"); +} + +fn host_locally(world: &World) { + let repo = repo_did(); + world.layout.create(&repo).unwrap(); + world.register( + &repo, + &RepoName::new("feralpearl").unwrap(), + &RepoRkey::new("feralpearl").unwrap(), + ); +} + +fn local_knot() -> World { + let upgrade = upgrading(); + host_locally(&upgrade); + upgrade +} + +async fn handover_pair( + records: serde_json::Value, + refuse_first: bool, +) -> (World, std::net::SocketAddr, AckSink, World) { + let foster = fostering(); + let (status, body) = post_seed(&foster, records).await; + assert_eq!(status, StatusCode::OK, "the seed must succeed: {body}"); + let sink = AckSink { + refuse_first, + ..ok_sink() + }; + let foster_addr = serve_foster(&foster, sink.clone()).await; + let upgrade = upgrading(); + host_locally(&upgrade); + (foster, foster_addr, sink, upgrade) +} + +async fn until(world: &World, ready: impl Fn(&World) -> bool) { + for _ in 0..400 { + if ready(world) { + return; + } + tokio::time::sleep(std::time::Duration::from_millis(25)).await; + } + panic!( + "the handover never reached the expected state: {}", + sidecar(world) + ); +} + +async fn settled(world: &World) -> serde_json::Value { + until(world, |world| { + let done = !sidecar_path(world).exists(); + let idle = world.state.handover_sessions.acquire(&repo_did()).is_some(); + done || (!sidecar(world)["error"].is_null() && idle) + }) + .await; + match sidecar_path(world).exists() { + false => serde_json::json!({"state": "done"}), + true => serde_json::json!({"state": "running", "error": sidecar(world)["error"]}), + } +} + +async fn post_trigger( + world: &World, + actor: &AccountDid, + foster_addr: std::net::SocketAddr, +) -> (StatusCode, serde_json::Value) { + post_authed( + world, + &nsid(HANDOVER), + actor, + serde_json::json!({ + "repoDid": REPO, + "fosterUrl": format!("http://{foster_addr}"), + }), + ) + .await +} + +async fn trigger(world: &World, foster_addr: std::net::SocketAddr) -> serde_json::Value { + let (status, _) = post_trigger(world, &owner(), foster_addr).await; + assert_eq!(status, StatusCode::OK, "the trigger route takes the job"); + settled(world).await +} + +async fn cancel(world: &World, actor: &AccountDid) -> (StatusCode, serde_json::Value) { + post_authed( + world, + &nsid(CANCEL), + actor, + serde_json::json!({ "repoDid": REPO }), + ) + .await +} + +fn sidecar_path(world: &World) -> std::path::PathBuf { + world + .layout + .repo_path(&repo_did()) + .unwrap() + .join("tangled.handover.json") +} + +fn sidecar(world: &World) -> serde_json::Value { + serde_json::from_slice(&std::fs::read(sidecar_path(world)).unwrap()).unwrap() +} + +fn assert_pending(world: &World) { + assert!(sidecar_path(world).exists(), "the failed job stays pending"); +} + +fn assert_retired(world: &World) { + assert!( + !sidecar_path(world).exists(), + "a finished job leaves no sidecar" + ); +} + +fn local_cob_refs(world: &World) -> BTreeMap { + let repo = repo_did(); + let git = world.layout.open(&repo).unwrap(); + git.references() + .unwrap() + .into_iter() + .filter(|record| record.name.as_str().starts_with("refs/cobs/")) + .map(|record| (record.name.as_str().to_owned(), record.target.to_string())) + .collect() +} + +async fn assert_cid_parity(foster: &World, upgrade: &World, collection: &str, rkey: &str) { + let path = + format!("/xrpc/com.atproto.repo.getRecord?repo={REPO}&collection={collection}&rkey={rkey}"); + let fetched = common::get_json(foster, &path).await; + let local = common::get_json(upgrade, &path).await; + assert_eq!( + local["cid"], fetched["cid"], + "the {rkey} record serves the foster knot's last cid" + ); + assert_eq!(local["value"], fetched["value"]); +} + +#[derive(serde::Serialize)] +struct AckSignedView<'a> { + context: &'a str, + #[serde(rename = "repoDid")] + repo_did: &'a str, + records: u64, + #[serde(rename = "ackedAt")] + acked_at: i64, +} + +fn ack_signature_checks_out(world: &World, ack: &serde_json::Value) -> bool { + let view = AckSignedView { + context: "org.tangled.temp.handover.ack.v1", + repo_did: ack["repoDid"].as_str().unwrap(), + records: ack["records"].as_u64().unwrap(), + acked_at: ack["ackedAt"].as_i64().unwrap(), + }; + let signed = serde_ipld_dagcbor::to_vec(&view).unwrap(); + let signature = Signature::from_bytes( + URL_SAFE_NO_PAD + .decode(ack["signature"].as_str().unwrap()) + .unwrap(), + ); + let public = world + .state + .secrets + .public_key(&world.state.knot_did) + .unwrap(); + knot_runtime::verify(&public, &signed, &signature) +} + +#[tokio::test] +async fn an_upgrading_knot_pulls_its_records_back_and_goes_green() { + let (foster, foster_addr, sink, upgrade) = + handover_pair(serde_json::json!([ticket_record("a feral pearl")]), false).await; + let outcome = trigger(&upgrade, foster_addr).await; + + assert_eq!(outcome["state"], "done", "handover outcome: {outcome}"); + assert_eq!( + sink.bodies.lock()[0]["records"], + 1, + "one seeded record comes home" + ); + assert_retired(&upgrade); + assert_cid_parity(&foster, &upgrade, COLLECTION, RKEY).await; + assert!( + !local_cob_refs(&upgrade).is_empty(), + "the cob refs came over with the objects" + ); + assert_eq!(sink.len(), 1, "exactly one ack leaves the session"); + let ack = &sink.bodies.lock()[0]; + assert_eq!(ack["repoDid"], REPO); + assert_eq!(ack["records"], 1); + assert!( + ack_signature_checks_out(&upgrade, ack), + "the ack is signed by the upgrading knot's own key" + ); +} + +#[tokio::test] +async fn a_foster_key_that_didnt_sign_the_graphs_stops_the_handover() { + let foster = fostering(); + seed_foster_record(&foster).await; + let sink = ok_sink(); + let knot = foster.state.knot_did.clone(); + let service = foster.state.knot_service_url.clone(); + let stranger = knot_runtime::K256Signer::generate(&knot_runtime::OsEntropy); + let lying = knot_atproto::knot_did_document(&knot, &stranger.public_key(), &service); + let addr = serve_router(did_edge(lying).fallback_service(pack_edge(&foster))).await; + + let upgrade = local_knot(); + let outcome = trigger(&upgrade, addr).await; + + assert_eq!( + outcome["state"], "running", + "a lying did document can't finish the handover: {outcome}" + ); + assert!( + outcome["error"] + .as_str() + .unwrap() + .contains("isn't validly signed"), + "the failure points at the unverified change: {outcome}" + ); + assert_pending(&upgrade); + assert!( + local_cob_refs(&upgrade).is_empty(), + "unverified graphs never touch the local refs" + ); + assert_eq!(sink.len(), 0, "no ack without green"); +} + +#[tokio::test] +async fn a_crashed_handover_replays_to_green() { + let (foster, foster_addr, sink, upgrade) = + handover_pair(serde_json::json!([ticket_record("a feral pearl")]), true).await; + let first = trigger(&upgrade, foster_addr).await; + assert_eq!( + first["state"], "running", + "the refused ack leaves it pending" + ); + assert_pending(&upgrade); + + let second = trigger(&upgrade, foster_addr).await; + assert_eq!(second["state"], "done", "the replay finishes: {second}"); + assert_eq!(sink.bodies.lock()[1]["records"], 1); + assert_retired(&upgrade); + assert_cid_parity(&foster, &upgrade, COLLECTION, RKEY).await; + assert_eq!(sink.len(), 2, "the ack is resent, not remembered"); + assert!( + sink.bodies + .lock() + .iter() + .all(|ack| ack_signature_checks_out(&upgrade, ack)), + "every ack is signed by the upgrading knot's key" + ); +} + +#[tokio::test] +async fn only_a_knot_admin_may_trigger_a_handover() { + let foster = fostering(); + let foster_addr = serve_foster(&foster, ok_sink()).await; + let upgrade = local_knot(); + let stranger = AccountDid::new("did:plc:visitor").unwrap(); + let (status, _) = post_trigger(&upgrade, &stranger, foster_addr).await; + assert_eq!(status, StatusCode::FORBIDDEN); + assert!( + !sidecar_path(&upgrade).exists(), + "a refused trigger stages no job" + ); +} + +#[tokio::test] +async fn a_repo_the_foster_knot_doesnt_host_is_rejected_early() { + let foster = fostering(); + let foster_addr = serve_foster(&foster, ok_sink()).await; + let upgrade = local_knot(); + let outcome = trigger(&upgrade, foster_addr).await; + assert_eq!(outcome["state"], "running"); + let error = outcome["error"].as_str().unwrap(); + assert!( + error.contains("unreachable") || error.contains("404"), + "the empty foster knot can't be pulled from: {error}" + ); + let git = upgrade.layout.open(&repo_did()).unwrap(); + assert_eq!( + CobStore::new(&git).cobs().unwrap().len(), + 0, + "no cobs were invented on the local side" + ); +} + +#[derive(serde::Serialize, serde::Deserialize)] +struct Stranger { + note: String, +} + +impl ChangePayload for Stranger { + const TYPE: &'static str = "org.tangled.test.stranger"; +} + +impl KnotAuthored for Stranger {} + +fn plant_stranger_cob(world: &World) { + let repo = repo_did(); + let git = world.layout.open(&repo).unwrap(); + let store = CobStore::new(&git); + let home = CobHome::from(&repo); + let signer = world.state.secrets.signer(&world.state.knot_did).unwrap(); + store + .create( + &home, + &Stranger { + note: "an odd barnacle".to_string(), + }, + &signer, + UnixSeconds::new(1100), + ) + .unwrap(); +} + +#[tokio::test] +async fn a_crashed_handover_replays_on_boot() { + let (_foster, foster_addr, sink, upgrade) = + handover_pair(serde_json::json!([ticket_record("a feral pearl")]), true).await; + let first = trigger(&upgrade, foster_addr).await; + assert_eq!( + first["state"], "running", + "the refused ack leaves it pending" + ); + + let stray = RepoDid::new("did:plc:straypearl").unwrap(); + upgrade.layout.create(&stray).unwrap(); + let stray_sidecar = upgrade + .layout + .repo_path(&stray) + .unwrap() + .join("tangled.handover.json"); + std::fs::write(&stray_sidecar, b"{").unwrap(); + assert_eq!( + knot_xrpc::handover::recover(&upgrade.state), + vec![repo_did()], + "recover finds the running job in the repo dir and skips the undecodable one" + ); + + knot_xrpc::handover::run_boot(Arc::clone(&upgrade.state), CancellationToken::new()).await; + + assert_retired(&upgrade); + assert_eq!(sink.bodies.lock()[1]["records"], 1); + assert!(knot_xrpc::handover::recover(&upgrade.state).is_empty()); + assert_eq!(sink.len(), 2, "the boot replay resends the ack"); + assert!( + stray_sidecar.exists(), + "an undecodable sidecar is left alone" + ); +} + +#[tokio::test] +async fn a_stuck_handover_job_can_be_canceled_by_an_admin() { + let (_foster, foster_addr, _, upgrade) = + handover_pair(serde_json::json!([ticket_record("a feral pearl")]), true).await; + let first = trigger(&upgrade, foster_addr).await; + assert_eq!( + first["state"], "running", + "the refused ack leaves the job stuck" + ); + + let (status, outcome) = cancel(&upgrade, &owner()).await; + assert_eq!(status, StatusCode::OK); + assert_eq!(outcome["repoDid"], REPO); + assert_retired(&upgrade); + assert!(knot_xrpc::handover::recover(&upgrade.state).is_empty()); + + let (status, _) = cancel(&upgrade, &owner()).await; + assert_eq!( + status, + StatusCode::NOT_FOUND, + "canceling twice finds no job" + ); +} + +#[tokio::test] +async fn a_reserved_handover_returns_409_to_a_second_trigger() { + let (_foster, foster_addr, _, upgrade) = + handover_pair(serde_json::json!([ticket_record("a feral pearl")]), false).await; + let reservation = upgrade + .state + .handover_sessions + .acquire(&repo_did()) + .unwrap(); + let (status, _) = post_trigger(&upgrade, &owner(), foster_addr).await; + assert_eq!(status, StatusCode::CONFLICT); + let (status, _) = cancel(&upgrade, &owner()).await; + assert_eq!(status, StatusCode::NOT_FOUND, "cancel finds no sidecar"); + drop(reservation); + + let outcome = trigger(&upgrade, foster_addr).await; + assert_eq!( + outcome["state"], "done", + "once the reservation drops the handover runs: {outcome}" + ); +} + +#[tokio::test] +async fn a_decoy_verification_method_cant_misdirect_the_foster_key() { + let foster = fostering(); + seed_foster_record(&foster).await; + let sink = ok_sink(); + let knot = foster.state.knot_did.clone(); + let service = foster.state.knot_service_url.clone(); + let real = foster.state.secrets.public_key(&knot).unwrap(); + let stranger = knot_runtime::K256Signer::generate(&knot_runtime::OsEntropy); + let mut document = + serde_json::to_value(knot_atproto::knot_did_document(&knot, &real, &service)).unwrap(); + let mut other = serde_json::to_value(knot_atproto::knot_did_document( + &knot, + &stranger.public_key(), + &service, + )) + .unwrap(); + other["verificationMethod"][0]["id"] = serde_json::json!(format!("{}#rotation", knot.as_str())); + let mut methods = other["verificationMethod"].as_array().unwrap().clone(); + methods.append(document["verificationMethod"].as_array_mut().unwrap()); + *document.get_mut("verificationMethod").unwrap() = serde_json::Value::Array(methods); + let addr = serve_router( + ack_edge(sink.clone()) + .fallback_service(did_edge(document).fallback_service(pack_edge(&foster))), + ) + .await; + + let upgrade = local_knot(); + let outcome = trigger(&upgrade, addr).await; + assert_eq!( + outcome["state"], "done", + "the key comes from the #atproto method, so the decoy is skipped: {outcome}" + ); + assert_eq!(sink.len(), 1); +} + +#[tokio::test] +async fn a_cob_type_with_an_unknown_history_model_blocks_the_handover() { + let (foster, foster_addr, _, upgrade) = + handover_pair(serde_json::json!([ticket_record("a feral pearl")]), false).await; + plant_stranger_cob(&foster); + let outcome = trigger(&upgrade, foster_addr).await; + assert_eq!( + outcome["state"], "running", + "an unverifiable history model can't go green: {outcome}" + ); + assert!( + outcome["error"].as_str().unwrap().contains("history model"), + "the error reports the type this knot can't vouch for: {outcome}" + ); + assert!( + local_cob_refs(&upgrade).is_empty(), + "no refs are applied while a served type can't be trusted" + ); +} + +#[tokio::test] +async fn an_empty_seed_is_refused() { + let foster = fostering(); + let (status, body) = post_seed(&foster, serde_json::json!([])).await; + assert_eq!(status, StatusCode::BAD_REQUEST, "{body}"); +} + +#[tokio::test] +async fn an_empty_shadow_cant_go_green() { + let foster = fostering(); + host_locally(&foster); + let sink = ok_sink(); + let foster_addr = serve_foster(&foster, sink.clone()).await; + + let upgrade = local_knot(); + let outcome = trigger(&upgrade, foster_addr).await; + assert_eq!(outcome["state"], "running", "{outcome}"); + assert!( + outcome["error"].as_str().unwrap().contains("empty shadow"), + "the failure points at the empty shell: {outcome}" + ); + assert!(local_cob_refs(&upgrade).is_empty()); + assert_eq!(sink.len(), 0, "no ack without records"); +} + +#[tokio::test] +async fn a_social_write_onto_a_shadow_repo_is_refused() { + let foster = fostering(); + seed_foster_record(&foster).await; + let repo = repo_did(); + let collection = common::ticket_collection(); + let (status, body) = common::post_as( + &foster, + &nsid("com.atproto.repo.createRecord"), + &owner(), + &ServiceDid::new(repo.as_str()).unwrap(), + serde_json::json!({ + "repo": repo.as_str(), + "collection": collection.as_str(), + "record": { + "$type": collection.as_str(), + "repo": repo.as_str(), + "title": "stray", + "body": "a social write onto a shadow repo", + "state": "open", + "createdAt": "2026-09-11T00:00:00Z" + } + }), + ) + .await; + assert_eq!(status, StatusCode::FORBIDDEN, "{body}"); + assert!( + body.to_string().contains("only fostered here"), + "the 403 comes from the fostered-repo check itself: {body}" + ); +} + +#[tokio::test] +async fn a_record_outside_the_social_collections_comes_home_too() { + let (foster, foster_addr, sink, upgrade) = handover_pair( + serde_json::json!([ticket_record("a feral pearl"), markdown_record()]), + false, + ) + .await; + let outcome = trigger(&upgrade, foster_addr).await; + assert_eq!(outcome["state"], "done", "{outcome}"); + assert_eq!( + sink.bodies.lock()[0]["records"], + 2, + "both records come home" + ); + for (collection, rkey) in [(COLLECTION, RKEY), (MARKDOWN, NOTES_RKEY)] { + assert_cid_parity(&foster, &upgrade, collection, rkey).await; + } +} + +fn regress_foster_refs(world: &World, prior: &BTreeMap) { + let git = world.layout.open(&repo_did()).unwrap(); + let back = git + .references() + .unwrap() + .into_iter() + .filter_map(|record| { + let name = record.name.as_str().to_owned(); + let old = prior.get(&name)?; + (*old != record.target.to_string()).then(|| RefUpdate::Update { + name: RefName::new(&name).unwrap(), + old: record.target, + new: Oid::from_hex(old).unwrap(), + }) + }) + .collect::>(); + git.update_refs(&back).unwrap(); +} + +#[tokio::test] +async fn a_regressed_foster_ref_cant_pull_the_local_ref_backwards() { + let (foster, foster_addr, _, upgrade) = + handover_pair(serde_json::json!([ticket_record("a feral pearl")]), false).await; + assert_eq!(trigger(&upgrade, foster_addr).await["state"], "done"); + let after_first = local_cob_refs(&foster); + + let (status, body) = post_seed( + &foster, + serde_json::json!([ticket_record("a feraler pearl")]), + ) + .await; + assert_eq!(status, StatusCode::OK, "{body}"); + let outcome = trigger(&upgrade, foster_addr).await; + assert_eq!(outcome["state"], "done", "{outcome}"); + let after_second = local_cob_refs(&foster); + assert!( + after_second != after_first, + "the reseed must move the foster knot's cob refs" + ); + + regress_foster_refs(&foster, &after_first); + let outcome = trigger(&upgrade, foster_addr).await; + assert_eq!(outcome["state"], "running", "{outcome}"); + assert!( + outcome["error"].as_str().unwrap().contains("isn't ahead"), + "the error reports the ref wasn't moved backwards: {outcome}" + ); + assert_eq!( + local_cob_refs(&upgrade), + after_second, + "the local refs keep the newer tips" + ); +} + +fn hanging_ack_edge() -> axum::Router { + axum::Router::new().route( + ACK_PATH, + axum::routing::post(|| async { std::future::pending::<()>().await }), + ) +} + +#[tokio::test] +async fn a_canceled_live_pull_stops_and_releases_the_repo() { + let foster = fostering(); + seed_foster_record(&foster).await; + let addr = serve_router(hanging_ack_edge().fallback_service(pack_edge(&foster))).await; + let upgrade = local_knot(); + let (status, _) = post_trigger(&upgrade, &owner(), addr).await; + assert_eq!(status, StatusCode::OK); + until(&upgrade, |world| { + world.state.handover_sessions.acquire(&repo_did()).is_none() + }) + .await; + let (status, outcome) = cancel(&upgrade, &owner()).await; + assert_eq!(status, StatusCode::OK, "{outcome}"); + until(&upgrade, |world| { + world.state.handover_sessions.acquire(&repo_did()).is_some() + }) + .await; + assert_retired(&upgrade); +} diff --git a/lexicons/org/tangled/temp/server/ackHandover.json b/lexicons/org/tangled/temp/server/ackHandover.json new file mode 100644 index 000000000..7e700bba9 --- /dev/null +++ b/lexicons/org/tangled/temp/server/ackHandover.json @@ -0,0 +1,33 @@ +{ + "lexicon": 1, + "id": "org.tangled.temp.server.ackHandover", + "defs": { + "main": { + "type": "procedure", + "description": "Upgrading knot sends this once it is ready to serve *every* fostered record of a repo - at exact CIDs the foster knot was giving out at ingestion time. The signature thingy is this 3rd party requesting-knot's key over dag-cbor of the same fields using the context org.tangled.temp.handover.ack.v1, such that the foster knot can verify it against the requester's did:web doc.", + "input": { + "encoding": "application/json", + "schema": { + "type": "object", + "required": ["repoDid", "records", "ackedAt", "signature"], + "properties": { + "repoDid": { + "type": "string", + "format": "did" + }, + "records": { + "type": "integer" + }, + "ackedAt": { + "type": "integer", + "description": "Ack time in Unix seconds." + }, + "signature": { + "type": "string" + } + } + } + } + } + } +} diff --git a/lexicons/org/tangled/temp/server/cancelHandover.json b/lexicons/org/tangled/temp/server/cancelHandover.json new file mode 100644 index 000000000..777d00bcd --- /dev/null +++ b/lexicons/org/tangled/temp/server/cancelHandover.json @@ -0,0 +1,36 @@ +{ + "lexicon": 1, + "id": "org.tangled.temp.server.cancelHandover", + "defs": { + "main": { + "type": "procedure", + "description": "Abandoning a repo's handover job mid-way; sidecar is removed, boot replay will stop retrying sidecarring. Only a knot admin should/could ever call this.", + "input": { + "encoding": "application/json", + "schema": { + "type": "object", + "required": ["repoDid"], + "properties": { + "repoDid": { + "type": "string", + "format": "did" + } + } + } + }, + "output": { + "encoding": "application/json", + "schema": { + "type": "object", + "required": ["repoDid"], + "properties": { + "repoDid": { + "type": "string", + "format": "did" + } + } + } + } + } + } +} diff --git a/lexicons/org/tangled/temp/server/handoverRepo.json b/lexicons/org/tangled/temp/server/handoverRepo.json new file mode 100644 index 000000000..343fdc1b4 --- /dev/null +++ b/lexicons/org/tangled/temp/server/handoverRepo.json @@ -0,0 +1,44 @@ +{ + "lexicon": 1, + "id": "org.tangled.temp.server.handoverRepo", + "defs": { + "main": { + "type": "procedure", + "description": "Pull a hosted repo's shadowed COB records back from a given foster knot, rebuild the record chain under this knot's own authority, confirm record CIDs against foster knot's state, finally ack the handover. Since it may take a bit, the endpoint will already give a response as soon as the job is staged up with state \"running\". Then the next steps run in background finish into the sidecar + logs. Only a knot admin can call this!", + "input": { + "encoding": "application/json", + "schema": { + "type": "object", + "required": ["repoDid", "fosterUrl"], + "properties": { + "repoDid": { + "type": "string", + "format": "did" + }, + "fosterUrl": { + "type": "string", + "format": "uri" + } + } + } + }, + "output": { + "encoding": "application/json", + "schema": { + "type": "object", + "required": ["repoDid", "state"], + "properties": { + "repoDid": { + "type": "string", + "format": "did" + }, + "state": { + "type": "string", + "knownValues": ["running"] + } + } + } + } + } + } +} diff --git a/lexicons/org/tangled/temp/server/seedRepo.json b/lexicons/org/tangled/temp/server/seedRepo.json index 17e39f812..f81450653 100644 --- a/lexicons/org/tangled/temp/server/seedRepo.json +++ b/lexicons/org/tangled/temp/server/seedRepo.json @@ -36,6 +36,7 @@ }, "records": { "type": "array", + "minLength": 1, "maxLength": 200, "description": "Records to write. Max 200 per req.", "items": {