use axum::body::Bytes; use base64::Engine; use base64::engine::general_purpose::URL_SAFE_NO_PAD; use futures::future::join_all; use futures::stream::StreamExt; use http::Method; use http::header::AUTHORIZATION; use serde_json::{Value, json}; use std::collections::{BTreeMap, BTreeSet}; use std::net::SocketAddr; use std::sync::Arc; use tower::ServiceExt; use knot_runtime::{Entropy, K256Signer, SeededEntropy, Signer}; use knot_types::{AccountDid, DidRkey, HttpStatus, KnotHostname, RepoDid, UnixSeconds}; use crate::harness::{AcceptanceKind, AcceptanceRecord, Harness, SUBJECT_DIDS}; use crate::trace::{OperationIndex, Outcome, Projection, RoundNumber, Step, Trace, fnv1a}; const SKEW_BACKDATE_SECS: i64 = 600; const SKEW_LIFETIME_SECS: i64 = 60; #[derive(Clone, Copy, PartialEq, Eq, PartialOrd, Ord)] struct RepoIndex(usize); #[derive(Clone, Copy, PartialEq, Eq, PartialOrd, Ord)] struct SubjectIndex(usize); #[derive(Clone, Copy)] struct StrangerIndex(usize); #[derive(Clone, Copy)] enum ReadOp { Version, Owner, ListMembers, DidJson, InfoRefs(RepoIndex), Branches(RepoIndex), Log(RepoIndex), DescribeRepo(RepoIndex), Tree(RepoIndex), Blob(RepoIndex), Languages(RepoIndex), ListCollaborators(RepoIndex), } impl ReadOp { fn name(self) -> &'static str { match self { ReadOp::Version => "version", ReadOp::Owner => "owner", ReadOp::ListMembers => "listMembers", ReadOp::DidJson => "didJson", ReadOp::InfoRefs(_) => "infoRefs", ReadOp::Branches(_) => "branches", ReadOp::Log(_) => "log", ReadOp::DescribeRepo(_) => "describeRepo", ReadOp::Tree(_) => "tree", ReadOp::Blob(_) => "blob", ReadOp::Languages(_) => "languages", ReadOp::ListCollaborators(_) => "listCollaborators", } } } #[derive(Clone, Copy)] enum AdminOp { AddMember(SubjectIndex), RemoveMember(SubjectIndex), Ban(SubjectIndex), Unban(SubjectIndex), CreateRepo(u32), AddCollaborator(RepoIndex, SubjectIndex), } impl AdminOp { fn name(self) -> &'static str { match self { AdminOp::AddMember(_) => "addMember", AdminOp::RemoveMember(_) => "removeMember", AdminOp::Ban(_) => "ban", AdminOp::Unban(_) => "unban", AdminOp::CreateRepo(_) => "createRepo", AdminOp::AddCollaborator(_, _) => "addCollaborator", } } } #[derive(Clone, Copy)] enum Mutation { Admin { op: AdminOp, skew: bool }, Probe { stranger: StrangerIndex, drop: bool }, Maintain { repo: RepoIndex }, } #[derive(Clone, Copy, PartialEq, Eq, PartialOrd, Ord)] enum Offered { Membership(SubjectIndex), Collaboration(RepoIndex, SubjectIndex), } #[derive(Clone, Copy)] struct Answer { offer: Offered, published: bool, skew: bool, } impl Answer { fn name(self) -> &'static str { match self.offer { Offered::Membership(_) => "acceptMembership", Offered::Collaboration(_, _) => "acceptCollaboration", } } fn fault(self) -> &'static str { match (self.skew, self.published) { (true, _) => "clock_skew", (false, false) => "unpublished", (false, true) => "none", } } } #[derive(Clone, Copy)] struct Observation { op: ReadOp, killed: bool, } #[derive(Clone, Copy)] enum Planned { Mutate(Mutation), Answer(Answer), Observe(Observation), } impl Planned { fn creates_a_repo(self) -> bool { matches!( self, Planned::Mutate(Mutation::Admin { op: AdminOp::CreateRepo(_), skew: false, }) ) } } enum Phase { Mutate(Vec), Answer(Vec), Observe(Vec), } impl Phase { fn ops(&self) -> Vec { match self { Phase::Mutate(ops) => ops.iter().copied().map(Planned::Mutate).collect(), Phase::Answer(ops) => ops.iter().copied().map(Planned::Answer).collect(), Phase::Observe(ops) => ops.iter().copied().map(Planned::Observe).collect(), } } } pub(crate) struct Round { phase: Phase, advance: std::time::Duration, } pub(crate) struct Rng(SeededEntropy); impl Rng { pub(crate) fn new(seed: u64) -> Self { Self(SeededEntropy::new(seed ^ 0x57ee_d000)) } pub(crate) fn below(&self, n: u64) -> u64 { self.0.next_u64() % n.max(1) } pub(crate) fn chance(&self, num: u64, den: u64) -> bool { assert!(num <= den, "chance numerator exceeds denominator"); self.below(den) < num } } pub(crate) fn plan(seed: u64, rounds: u32, subjects: usize) -> Vec { let rng = Rng::new(seed); let mut available: u32 = 1; let mut rkey: u32 = 0; let mut stranger: usize = 0; let mut unanswered: BTreeSet = BTreeSet::new(); (0..rounds) .map(|round| { let (phase, fresh_repos) = match round % 3 { 0 => { let (ops, fresh_repos) = mutate_round( &rng, subjects, available, &mut rkey, &mut stranger, &mut unanswered, ); (Phase::Mutate(ops), fresh_repos) } 1 => ( Phase::Answer(answer_round(&rng, subjects, &mut unanswered)), 0, ), _ => (Phase::Observe(observe_round(&rng, available)), 0), }; let advance = std::time::Duration::from_micros(rng.below(3_000_000)); available += fresh_repos; Round { phase, advance } }) .collect() } #[derive(Default)] struct Model { members: BTreeSet, blocked: BTreeSet, collaborators: BTreeMap>, offered: BTreeSet, written: BTreeSet, } pub(crate) fn predict(seed: u64, rounds: u32) -> Projection { let subject_did = |subject: SubjectIndex| AccountDid::new(SUBJECT_DIDS[subject.0]).expect("subject did"); let plan = plan(seed, rounds, SUBJECT_DIDS.len()); let ops = || plan.iter().flat_map(|round| round.phase.ops()); let repos = 1 + ops().filter(|planned| planned.creates_a_repo()).count(); let model = ops().fold(Model::default(), |mut model, planned| { match planned { Planned::Mutate(Mutation::Admin { op, skew: false }) => match op { AdminOp::AddMember(subject) => { model.offered.insert(Offered::Membership(subject)); } AdminOp::RemoveMember(subject) => { model.offered.remove(&Offered::Membership(subject)); model.members.remove(&subject_did(subject)); } AdminOp::Ban(subject) => { model.blocked.insert(subject_did(subject)); } AdminOp::Unban(subject) => { model.blocked.remove(&subject_did(subject)); } AdminOp::AddCollaborator(repo, subject) => { model.offered.insert(Offered::Collaboration(repo, subject)); } AdminOp::CreateRepo(_) => {} }, Planned::Answer(Answer { offer, published, skew, }) => { if published { model.written.insert(offer); } let admitted = !skew && model.offered.contains(&offer) && model.written.contains(&offer); match (admitted, offer) { (false, _) => {} (true, Offered::Membership(subject)) => { model.members.insert(subject_did(subject)); } (true, Offered::Collaboration(repo, subject)) => { model .collaborators .entry(repo.0) .or_default() .insert(subject_did(subject)); } } } Planned::Mutate(Mutation::Admin { skew: true, .. }) | Planned::Mutate(Mutation::Probe { .. }) | Planned::Mutate(Mutation::Maintain { .. }) | Planned::Observe(_) => {} } model }); let mut rosters = model.collaborators; Projection { members: model.members.into_iter().collect(), blocked: model.blocked.into_iter().collect(), collaborators: (0..repos) .map(|repo| { rosters .remove(&repo) .unwrap_or_default() .into_iter() .collect() }) .collect(), } } fn mutate_round( rng: &Rng, subjects: usize, available: u32, rkey: &mut u32, stranger: &mut usize, unanswered: &mut BTreeSet, ) -> (Vec, u32) { let mut ops: Vec = (0..subjects) .filter(|_| rng.chance(2, 3)) .map(|subject| { let subject = SubjectIndex(subject); let op = match rng.below(4) { 0 => AdminOp::AddMember(subject), 1 => AdminOp::RemoveMember(subject), 2 => AdminOp::Ban(subject), _ => AdminOp::Unban(subject), }; Mutation::Admin { op, skew: rng.chance(1, 5), } }) .collect(); (0..available) .filter(|_| rng.chance(1, 2)) .for_each(|repo| { let repo = RepoIndex(repo as usize); let planned = if rng.chance(1, 2) { let subject = SubjectIndex(rng.below(subjects as u64) as usize); let skew = rng.chance(1, 6); if !skew { unanswered.insert(Offered::Collaboration(repo, subject)); } Mutation::Admin { op: AdminOp::AddCollaborator(repo, subject), skew, } } else { Mutation::Maintain { repo } }; ops.push(planned); }); let mut fresh_repos = 0; (0..rng.below(3)).for_each(|_| { let key = *rkey; *rkey += 1; let skew = rng.chance(1, 8); if !skew { fresh_repos += 1; } ops.push(Mutation::Admin { op: AdminOp::CreateRepo(key), skew, }); }); (0..rng.below(3)).for_each(|_| { let stranger_index = StrangerIndex(*stranger); *stranger += 1; ops.push(Mutation::Probe { stranger: stranger_index, drop: rng.chance(1, 2), }); }); (ops, fresh_repos) } fn answer_round(rng: &Rng, subjects: usize, unanswered: &mut BTreeSet) -> Vec { let answer = |offer| Answer { offer, published: rng.chance(3, 4), skew: rng.chance(1, 6), }; let mut ops: Vec = (0..subjects) .filter(|_| rng.chance(1, 2)) .map(|subject| answer(Offered::Membership(SubjectIndex(subject)))) .collect(); let answering: Vec = unanswered .iter() .copied() .filter(|_| rng.chance(2, 3)) .collect(); answering.iter().for_each(|offer| { unanswered.remove(offer); }); ops.extend(answering.into_iter().map(answer)); ops } fn observe_round(rng: &Rng, available: u32) -> Vec { let mut ops: Vec = [ ReadOp::Version, ReadOp::Owner, ReadOp::ListMembers, ReadOp::DidJson, ] .into_iter() .map(|op| Observation { op, killed: rng.chance(1, 5), }) .collect(); (0..available) .filter(|_| rng.chance(2, 3)) .for_each(|repo| { let repo = RepoIndex(repo as usize); let op = match rng.below(8) { 0 => ReadOp::Branches(repo), 1 => ReadOp::Log(repo), 2 => ReadOp::DescribeRepo(repo), 3 => ReadOp::InfoRefs(repo), 4 => ReadOp::Tree(repo), 5 => ReadOp::Blob(repo), 6 => ReadOp::ListCollaborators(repo), _ => ReadOp::Languages(repo), }; ops.push(Observation { op, killed: rng.chance(1, 5), }); }); ops } struct OpResult { step: Step, created: Option, } pub(crate) async fn execute(harness: Arc, seed: u64, plan: Vec) -> Trace { let initial = ( vec![harness.seed_repo.clone()], Vec::::new(), Vec::new(), ); let harness = &harness; let (repos, steps, snapshots) = futures::stream::iter(plan.into_iter().enumerate()) .fold( initial, |(repos, mut steps, mut snapshots), (round_index, round)| { let harness = Arc::clone(harness); async move { let round_no = RoundNumber::new(round_index as u32); let ops = round.phase.ops(); let drops = arm_drops(&harness, &ops); let repos = Arc::new(repos); let tasks = ops .iter() .enumerate() .map(|(index, planned)| { let harness = Arc::clone(&harness); let repos = Arc::clone(&repos); let planned = *planned; tokio::spawn(async move { run_op( &harness, &repos, round_no, OperationIndex::new(index as u32), planned, ) .await }) }) .collect::>(); let results: Vec = join_all(tasks) .await .into_iter() .map(|joined| joined.expect("sim op task mustn't panic")) .collect(); drops .iter() .for_each(|host| harness.faults.clear_host(host)); let created: Vec = results .iter() .filter_map(|result| result.created.clone()) .collect(); let planned_creates = ops .iter() .filter(|planned| planned.creates_a_repo()) .count(); assert_eq!( planned_creates, created.len(), "round {}: {planned_creates} non-skew creates planned but \ {} materialized, so plan/execute repo indices have drifted apart", round_no.get(), created.len() ); created.iter().for_each(|did| harness.populate(did)); steps.extend(results.into_iter().map(|result| result.step)); let mut repos = Arc::into_inner(repos) .expect("all op tasks released the round repo snapshot"); repos.extend(created); harness.advance(round.advance); snapshots.push(harness.snapshot(round_no, &repos)); (repos, steps, snapshots) } }, ) .await; let no_fault_creates = steps .iter() .filter(|step| step.op == "createRepo" && step.fault == "none") .count(); let materialized = repos.len() - 1; assert_eq!( no_fault_creates, materialized, "no-fault createRepo count {no_fault_creates} doesn't match the {materialized} repos \ materialized: a planned create silently failed and repo_at would have masked the drift" ); Trace { seed, steps, snapshots, } } fn arm_drops(harness: &Harness, ops: &[Planned]) -> Vec { let hosts: Vec = ops .iter() .filter_map(|planned| match planned { Planned::Mutate(Mutation::Probe { stranger, drop: true, }) => Some(harness.strangers[stranger.0].host.clone()), _ => None, }) .collect(); hosts.iter().for_each(|host| harness.faults.drop_host(host)); hosts } async fn run_op( harness: &Harness, repos: &[RepoDid], round: RoundNumber, index: OperationIndex, planned: Planned, ) -> OpResult { let make = |op: &'static str, actor: String, fault: &'static str, outcome: Outcome| Step { round, index, op, actor, fault, outcome, }; match planned { Planned::Mutate(Mutation::Maintain { repo }) => { let repo = repo_at(repos, repo); let outcome = match harness.maintain(repo) { Ok(()) => Outcome::Answered { status: HttpStatus::new(200), body: 0, }, Err(message) => Outcome::Answered { status: HttpStatus::new(500), body: fnv1a(message.as_bytes()), }, }; OpResult { step: make("maintain", "knot".to_string(), "none", outcome), created: None, } } Planned::Observe(Observation { op, killed }) => { let request = read_request(repos, op); if killed { drive_kill(harness.router(), request.method, &request.uri, request.body).await; return OpResult { step: make(op.name(), request.actor, "killed", Outcome::Killed), created: None, }; } let (status, body) = http_call( harness.router(), request.method, &request.uri, None, request.body, ) .await; OpResult { step: make( op.name(), request.actor, "none", Outcome::Answered { status, body: body_digest(&body), }, ), created: None, } } Planned::Mutate(Mutation::Admin { op, skew }) => { let request = admin_request(harness, repos, op, skew, round, index); let (status, body) = http_call( harness.router(), request.method, &request.uri, request.token.as_deref(), request.body, ) .await; let created = match op { AdminOp::CreateRepo(_) if status == HttpStatus::new(200) => repo_did_of(&body), _ => None, }; OpResult { step: make( op.name(), request.actor, if skew { "clock_skew" } else { "none" }, Outcome::Answered { status, body: body_digest(&body), }, ), created, } } Planned::Mutate(Mutation::Probe { stranger, drop }) => { let request = probe_request(harness, stranger, round, index); let (status, body) = http_call( harness.router(), request.method, &request.uri, request.token.as_deref(), request.body, ) .await; OpResult { step: make( "probe", request.actor, if drop { "drop_identity" } else { "none" }, Outcome::Answered { status, body: body_digest(&body), }, ), created: None, } } Planned::Answer(answer) => { let record = acceptance_of(harness, repos, answer); if answer.published { harness.published.write(record.clone()); } let request = answer_request(harness, &record, answer.skew, round, index); let (status, body) = http_call( harness.router(), request.method, &request.uri, request.token.as_deref(), request.body, ) .await; OpResult { step: make( answer.name(), request.actor, answer.fault(), Outcome::Answered { status, body: body_digest(&body), }, ), created: None, } } } } pub(crate) struct Request { pub(crate) method: Method, pub(crate) uri: String, pub(crate) token: Option, pub(crate) body: Bytes, pub(crate) actor: String, } fn read_request(repos: &[RepoDid], op: ReadOp) -> Request { match op { ReadOp::Version => get("/xrpc/sh.tangled.knot.version"), ReadOp::Owner => get("/xrpc/sh.tangled.owner"), ReadOp::ListMembers => { get("/xrpc/sh.tangled.knot.listMembers?subject=did:web:knot.nel.pet") } ReadOp::DidJson => get("/.well-known/did.json"), ReadOp::InfoRefs(repo) => Request { method: Method::GET, uri: format!( "/{}/info/refs?service=git-upload-pack", repo_at(repos, repo).as_str() ), token: None, body: Bytes::new(), actor: "anon".to_string(), }, ReadOp::Branches(repo) => repo_get("branches", "repo", repo_at(repos, repo)), ReadOp::Log(repo) => repo_get("log", "repo", repo_at(repos, repo)), ReadOp::DescribeRepo(repo) => repo_get("describeRepo", "repoDid", repo_at(repos, repo)), ReadOp::Tree(repo) => repo_get("tree", "repo", repo_at(repos, repo)), ReadOp::Languages(repo) => repo_get("languages", "repo", repo_at(repos, repo)), ReadOp::ListCollaborators(repo) => { repo_get("listCollaborators", "subject", repo_at(repos, repo)) } ReadOp::Blob(repo) => Request { method: Method::GET, uri: format!( "/xrpc/sh.tangled.repo.blob?repo={}&path=README.md", enc(repo_at(repos, repo).as_str()) ), token: None, body: Bytes::new(), actor: "anon".to_string(), }, } } fn admin_request( harness: &Harness, repos: &[RepoDid], op: AdminOp, skew: bool, round: RoundNumber, index: OperationIndex, ) -> Request { let subjects = &harness.subjects; match op { AdminOp::AddMember(subject) => admin_post( harness, "addMember", "sh.tangled.knot.addMember", json!({ "subject": subjects[subject.0].as_str() }), skew, round, index, ), AdminOp::RemoveMember(subject) => admin_post( harness, "removeMember", "sh.tangled.knot.removeMember", json!({ "subject": subjects[subject.0].as_str() }), skew, round, index, ), AdminOp::Ban(subject) => admin_post( harness, "ban", "sh.tangled.knot.ban", json!({ "subject": subjects[subject.0].as_str() }), skew, round, index, ), AdminOp::Unban(subject) => admin_post( harness, "unban", "sh.tangled.knot.unban", json!({ "subject": subjects[subject.0].as_str() }), skew, round, index, ), AdminOp::CreateRepo(key) => { let name = format!("repo{key}"); admin_post( harness, "create", "sh.tangled.repo.create", json!({ "rkey": name, "name": name }), skew, round, index, ) } AdminOp::AddCollaborator(repo, subject) => admin_post( harness, "addCollaborator", "sh.tangled.repo.addCollaborator", json!({ "repo": repo_at(repos, repo).as_str(), "subject": subjects[subject.0].as_str(), }), skew, round, index, ), } } fn probe_request( harness: &Harness, stranger: StrangerIndex, round: RoundNumber, index: OperationIndex, ) -> Request { let actor = &harness.strangers[stranger.0]; let token = mint( &actor.signer, &actor.did, &harness.knot_aud, "sh.tangled.knot.addMember", jwt_window(harness, false), round, index, ); Request { method: Method::POST, uri: "/xrpc/sh.tangled.knot.addMember".to_string(), token: Some(token), body: encode_body(json!({ "subject": harness.subjects[0].as_str() })), actor: actor.host.to_string(), } } fn acceptance_of(harness: &Harness, repos: &[RepoDid], answer: Answer) -> AcceptanceRecord { let (subject, kind, rkey) = match answer.offer { Offered::Membership(subject) => ( subject, AcceptanceKind::Membership, harness.knot_rkey.clone(), ), Offered::Collaboration(repo, subject) => ( subject, AcceptanceKind::Collaboration, DidRkey::new(repo_at(repos, repo).as_str()).expect("repo did is a record key"), ), }; AcceptanceRecord { author: harness.subjects[subject.0].clone(), kind, rkey, } } fn answer_request( harness: &Harness, record: &AcceptanceRecord, skew: bool, round: RoundNumber, index: OperationIndex, ) -> Request { let (author, nsid) = (record.author.as_str(), record.kind.accept_method()); let token = mint( &harness.plc_signer, &record.author, &record.rkey, nsid, jwt_window(harness, skew), round, index, ); Request { method: Method::POST, uri: format!("/xrpc/{nsid}"), token: Some(token), body: encode_body(json!({ "acceptance": format!( "at://{author}/{}/{}", record.kind.collection(), record.rkey.as_str() ) })), actor: author.to_string(), } } fn get(path: &str) -> Request { Request { method: Method::GET, uri: path.to_string(), token: None, body: Bytes::new(), actor: "anon".to_string(), } } fn repo_get(method: &str, param: &str, repo: &RepoDid) -> Request { Request { method: Method::GET, uri: format!( "/xrpc/sh.tangled.repo.{method}?{param}={}", enc(repo.as_str()) ), token: None, body: Bytes::new(), actor: "anon".to_string(), } } fn admin_post( harness: &Harness, method_short: &str, nsid: &'static str, body: Value, skew: bool, round: RoundNumber, index: OperationIndex, ) -> Request { let admin = &harness.admin; let token = mint( &admin.signer, &admin.did, &harness.knot_aud, nsid, jwt_window(harness, skew), round, index, ); Request { method: Method::POST, uri: format!("/xrpc/{nsid}"), token: Some(token), body: encode_body(body), actor: format!("admin:{method_short}"), } } pub(crate) fn jwt_window(harness: &Harness, skew: bool) -> (UnixSeconds, UnixSeconds) { let now = harness.now_seconds(); if skew { ( now.saturating_sub_secs(SKEW_BACKDATE_SECS + SKEW_LIFETIME_SECS), now.saturating_sub_secs(SKEW_BACKDATE_SECS), ) } else { (now, now.saturating_add_secs(60)) } } pub(crate) fn mint( signer: &K256Signer, issuer: &AccountDid, aud: &impl AsRef, nsid: &str, window: (UnixSeconds, UnixSeconds), round: RoundNumber, index: OperationIndex, ) -> String { let header = URL_SAFE_NO_PAD.encode(br#"{"alg":"ES256K","typ":"JWT"}"#); let payload = URL_SAFE_NO_PAD.encode( serde_json::to_vec(&json!({ "iss": issuer.as_str(), "aud": aud.as_ref(), "iat": window.0.get(), "exp": window.1.get(), "jti": format!("sim-{}-{}", round.get(), index.get()), "lxm": nsid, })) .expect("claims serialize"), ); let signing_input = format!("{header}.{payload}"); let signature = signer.sign(signing_input.as_bytes()); format!( "{signing_input}.{}", URL_SAFE_NO_PAD.encode(signature.as_bytes()) ) } pub(crate) async fn http_call( router: axum::Router, method: Method, uri: &str, token: Option<&str>, body: Bytes, ) -> (HttpStatus, Bytes) { let mut request = http::Request::builder() .method(method) .uri(uri) .body(axum::body::Body::from(body)) .expect("request builds"); if let Some(token) = token { request.headers_mut().insert( AUTHORIZATION, http::HeaderValue::from_str(&format!("Bearer {token}")).expect("bearer header"), ); } request .extensions_mut() .insert(axum::extract::ConnectInfo(SocketAddr::from(( [127, 0, 0, 1], 4242, )))); let response = router.oneshot(request).await.expect("router answers"); let status = HttpStatus::from(response.status()); let bytes = axum::body::to_bytes(response.into_body(), usize::MAX) .await .expect("response body"); (status, bytes) } pub(crate) async fn drive_kill(router: axum::Router, method: Method, uri: &str, body: Bytes) { let call = http_call(router, method, uri, None, body); futures::pin_mut!(call); tokio::select! { biased; _ = &mut call => {} _ = tokio::task::yield_now() => {} } } pub(crate) fn repo_did_of(body: &Bytes) -> Option { serde_json::from_slice::(body) .ok() .and_then(|value| { value .get("repoDid") .and_then(Value::as_str) .map(str::to_string) }) .and_then(|did| RepoDid::new(did).ok()) } pub(crate) fn body_digest(body: &Bytes) -> u64 { match serde_json::from_slice::(body) { Ok(mut value) => { canonicalize(&mut value); fnv1a(&serde_json::to_vec(&value).expect("canonical body serializes")) } Err(_) => fnv1a(body), } } fn canonicalize(value: &mut Value) { match value { Value::Array(items) => { items.iter_mut().for_each(canonicalize); items.sort_by_cached_key(|item| serde_json::to_string(item).expect("array item")); } Value::Object(map) => map.values_mut().for_each(canonicalize), _ => {} } } pub(crate) fn encode_body(value: Value) -> Bytes { Bytes::from(serde_json::to_vec(&value).expect("request body serializes")) } pub(crate) fn enc(did: &str) -> String { did.replace(':', "%3A") } fn repo_at(repos: &[RepoDid], index: RepoIndex) -> &RepoDid { repos.get(index.0).unwrap_or_else(|| { panic!( "plan/execute repo drift: index {} exceeds {} repos created so far", index.0, repos.len() ) }) }