alice.pds.demo.boltless.dev
This repository has no description
did:plc:3fwecdnvtcscjnrx2p4n7a…
1use axum::body::Bytes;
2use base64::Engine;
3use base64::engine::general_purpose::URL_SAFE_NO_PAD;
4use futures::future::join_all;
5use futures::stream::StreamExt;
6use http::Method;
7use http::header::AUTHORIZATION;
8use serde_json::{Value, json};
9use std::collections::BTreeSet;
10use std::net::SocketAddr;
11use std::sync::Arc;
12use tower::ServiceExt;
13
14use knot_runtime::{Entropy, K256Signer, SeededEntropy, Signer};
15use knot_types::{AccountDid, HttpStatus, KnotHostname, KnotId, RepoDid, UnixSeconds};
16
17use crate::harness::{Harness, SUBJECT_DIDS};
18use crate::trace::{OperationIndex, Outcome, Projection, RoundNumber, Step, Trace, fnv1a};
19
20const SKEW_BACKDATE_SECS: i64 = 600;
21const SKEW_LIFETIME_SECS: i64 = 60;
22
23#[derive(Clone, Copy)]
24struct RepoIndex(usize);
25
26#[derive(Clone, Copy)]
27struct SubjectIndex(usize);
28
29#[derive(Clone, Copy)]
30struct StrangerIndex(usize);
31
32#[derive(Clone, Copy)]
33enum ReadOp {
34 Version,
35 Owner,
36 ListMembers,
37 DidJson,
38 InfoRefs(RepoIndex),
39 Branches(RepoIndex),
40 Log(RepoIndex),
41 DescribeRepo(RepoIndex),
42 Tree(RepoIndex),
43 Blob(RepoIndex),
44 Languages(RepoIndex),
45}
46
47impl ReadOp {
48 fn name(self) -> &'static str {
49 match self {
50 ReadOp::Version => "version",
51 ReadOp::Owner => "owner",
52 ReadOp::ListMembers => "listMembers",
53 ReadOp::DidJson => "didJson",
54 ReadOp::InfoRefs(_) => "infoRefs",
55 ReadOp::Branches(_) => "branches",
56 ReadOp::Log(_) => "log",
57 ReadOp::DescribeRepo(_) => "describeRepo",
58 ReadOp::Tree(_) => "tree",
59 ReadOp::Blob(_) => "blob",
60 ReadOp::Languages(_) => "languages",
61 }
62 }
63}
64
65#[derive(Clone, Copy)]
66enum AdminOp {
67 AddMember(SubjectIndex),
68 RemoveMember(SubjectIndex),
69 Ban(SubjectIndex),
70 Unban(SubjectIndex),
71 CreateRepo(u32),
72 AddCollaborator(RepoIndex, SubjectIndex),
73}
74
75impl AdminOp {
76 fn name(self) -> &'static str {
77 match self {
78 AdminOp::AddMember(_) => "addMember",
79 AdminOp::RemoveMember(_) => "removeMember",
80 AdminOp::Ban(_) => "ban",
81 AdminOp::Unban(_) => "unban",
82 AdminOp::CreateRepo(_) => "createRepo",
83 AdminOp::AddCollaborator(_, _) => "addCollaborator",
84 }
85 }
86}
87
88#[derive(Clone, Copy)]
89enum Planned {
90 Read { op: ReadOp, killed: bool },
91 Admin { op: AdminOp, skew: bool },
92 Probe { stranger: StrangerIndex, drop: bool },
93 Maintain { repo: RepoIndex },
94}
95
96pub(crate) struct Round {
97 ops: Vec<Planned>,
98 advance: std::time::Duration,
99}
100
101pub(crate) struct Rng(SeededEntropy);
102
103impl Rng {
104 pub(crate) fn new(seed: u64) -> Self {
105 Self(SeededEntropy::new(seed ^ 0x57ee_d000))
106 }
107
108 pub(crate) fn below(&self, n: u64) -> u64 {
109 self.0.next_u64() % n.max(1)
110 }
111
112 pub(crate) fn chance(&self, num: u64, den: u64) -> bool {
113 assert!(num <= den, "chance numerator exceeds denominator");
114 self.below(den) < num
115 }
116}
117
118pub(crate) fn plan(seed: u64, rounds: u32, subjects: usize) -> Vec<Round> {
119 let rng = Rng::new(seed);
120 let mut available: u32 = 1;
121 let mut rkey: u32 = 0;
122 let mut stranger: usize = 0;
123 (0..rounds)
124 .map(|round| {
125 let (ops, fresh_repos) = if round % 2 == 0 {
126 mutate_round(&rng, subjects, available, &mut rkey, &mut stranger)
127 } else {
128 (read_round(&rng, available), 0)
129 };
130 let advance = std::time::Duration::from_micros(rng.below(3_000_000));
131 available += fresh_repos;
132 Round { ops, advance }
133 })
134 .collect()
135}
136
137pub(crate) fn predict(seed: u64, rounds: u32) -> Projection {
138 let (members, blocked) = plan(seed, rounds, SUBJECT_DIDS.len())
139 .iter()
140 .flat_map(|round| round.ops.iter())
141 .fold(
142 (BTreeSet::<AccountDid>::new(), BTreeSet::<AccountDid>::new()),
143 |(mut members, mut blocked), planned| {
144 let subject_did = |subject: &SubjectIndex| {
145 AccountDid::new(SUBJECT_DIDS[subject.0]).expect("subject did")
146 };
147 if let Planned::Admin { op, skew: false } = planned {
148 match op {
149 AdminOp::AddMember(subject) => {
150 members.insert(subject_did(subject));
151 }
152 AdminOp::RemoveMember(subject) => {
153 members.remove(&subject_did(subject));
154 }
155 AdminOp::Ban(subject) => {
156 blocked.insert(subject_did(subject));
157 }
158 AdminOp::Unban(subject) => {
159 blocked.remove(&subject_did(subject));
160 }
161 AdminOp::CreateRepo(_) | AdminOp::AddCollaborator(_, _) => {}
162 }
163 }
164 (members, blocked)
165 },
166 );
167 Projection {
168 members: members.into_iter().collect(),
169 blocked: blocked.into_iter().collect(),
170 }
171}
172
173fn mutate_round(
174 rng: &Rng,
175 subjects: usize,
176 available: u32,
177 rkey: &mut u32,
178 stranger: &mut usize,
179) -> (Vec<Planned>, u32) {
180 let mut ops: Vec<Planned> = (0..subjects)
181 .filter(|_| rng.chance(2, 3))
182 .map(|subject| {
183 let subject = SubjectIndex(subject);
184 let op = match rng.below(4) {
185 0 => AdminOp::AddMember(subject),
186 1 => AdminOp::RemoveMember(subject),
187 2 => AdminOp::Ban(subject),
188 _ => AdminOp::Unban(subject),
189 };
190 Planned::Admin {
191 op,
192 skew: rng.chance(1, 5),
193 }
194 })
195 .collect();
196
197 (0..available)
198 .filter(|_| rng.chance(1, 2))
199 .for_each(|repo| {
200 let planned = if rng.chance(1, 2) {
201 let subject = SubjectIndex(rng.below(subjects as u64) as usize);
202 Planned::Admin {
203 op: AdminOp::AddCollaborator(RepoIndex(repo as usize), subject),
204 skew: rng.chance(1, 6),
205 }
206 } else {
207 Planned::Maintain {
208 repo: RepoIndex(repo as usize),
209 }
210 };
211 ops.push(planned);
212 });
213
214 let mut fresh_repos = 0;
215 (0..rng.below(3)).for_each(|_| {
216 let key = *rkey;
217 *rkey += 1;
218 let skew = rng.chance(1, 8);
219 if !skew {
220 fresh_repos += 1;
221 }
222 ops.push(Planned::Admin {
223 op: AdminOp::CreateRepo(key),
224 skew,
225 });
226 });
227
228 (0..rng.below(3)).for_each(|_| {
229 let stranger_index = StrangerIndex(*stranger);
230 *stranger += 1;
231 ops.push(Planned::Probe {
232 stranger: stranger_index,
233 drop: rng.chance(1, 2),
234 });
235 });
236
237 (ops, fresh_repos)
238}
239
240fn read_round(rng: &Rng, available: u32) -> Vec<Planned> {
241 let mut ops: Vec<Planned> = [
242 ReadOp::Version,
243 ReadOp::Owner,
244 ReadOp::ListMembers,
245 ReadOp::DidJson,
246 ]
247 .into_iter()
248 .map(|op| Planned::Read {
249 op,
250 killed: rng.chance(1, 5),
251 })
252 .collect();
253 (0..available)
254 .filter(|_| rng.chance(2, 3))
255 .for_each(|repo| {
256 let repo = RepoIndex(repo as usize);
257 let op = match rng.below(7) {
258 0 => ReadOp::Branches(repo),
259 1 => ReadOp::Log(repo),
260 2 => ReadOp::DescribeRepo(repo),
261 3 => ReadOp::InfoRefs(repo),
262 4 => ReadOp::Tree(repo),
263 5 => ReadOp::Blob(repo),
264 _ => ReadOp::Languages(repo),
265 };
266 ops.push(Planned::Read {
267 op,
268 killed: rng.chance(1, 5),
269 });
270 });
271 ops
272}
273
274struct OpResult {
275 step: Step,
276 created: Option<RepoDid>,
277}
278
279pub(crate) async fn execute(harness: Arc<Harness>, seed: u64, plan: Vec<Round>) -> Trace {
280 let initial = (
281 vec![harness.seed_repo.clone()],
282 Vec::<Step>::new(),
283 Vec::new(),
284 );
285 let harness = &harness;
286 let (repos, steps, snapshots) = futures::stream::iter(plan.into_iter().enumerate())
287 .fold(
288 initial,
289 |(repos, mut steps, mut snapshots), (round_index, round)| {
290 let harness = Arc::clone(harness);
291 async move {
292 let round_no = RoundNumber::new(round_index as u32);
293 let drops = arm_drops(&harness, &round.ops);
294 let repos = Arc::new(repos);
295 let tasks = round
296 .ops
297 .iter()
298 .enumerate()
299 .map(|(index, planned)| {
300 let harness = Arc::clone(&harness);
301 let repos = Arc::clone(&repos);
302 let planned = *planned;
303 tokio::spawn(async move {
304 run_op(
305 &harness,
306 &repos,
307 round_no,
308 OperationIndex::new(index as u32),
309 planned,
310 )
311 .await
312 })
313 })
314 .collect::<Vec<_>>();
315 let results: Vec<OpResult> = join_all(tasks)
316 .await
317 .into_iter()
318 .map(|joined| joined.expect("sim op task mustn't panic"))
319 .collect();
320 drops
321 .iter()
322 .for_each(|host| harness.faults.clear_host(host));
323
324 let created: Vec<RepoDid> = results
325 .iter()
326 .filter_map(|result| result.created.clone())
327 .collect();
328 let planned_creates = round
329 .ops
330 .iter()
331 .filter(|planned| {
332 matches!(
333 planned,
334 Planned::Admin {
335 op: AdminOp::CreateRepo(_),
336 skew: false,
337 }
338 )
339 })
340 .count();
341 assert_eq!(
342 planned_creates,
343 created.len(),
344 "round {}: {planned_creates} non-skew creates planned but \
345 {} materialized, so plan/execute repo indices have drifted apart",
346 round_no.get(),
347 created.len()
348 );
349 created.iter().for_each(|did| harness.populate(did));
350 steps.extend(results.into_iter().map(|result| result.step));
351 let mut repos = Arc::into_inner(repos)
352 .expect("all op tasks released the round repo snapshot");
353 repos.extend(created);
354
355 harness.advance(round.advance);
356 snapshots.push(harness.snapshot(round_no, &repos));
357 (repos, steps, snapshots)
358 }
359 },
360 )
361 .await;
362 let no_fault_creates = steps
363 .iter()
364 .filter(|step| step.op == "createRepo" && step.fault == "none")
365 .count();
366 let materialized = repos.len() - 1;
367 assert_eq!(
368 no_fault_creates, materialized,
369 "no-fault createRepo count {no_fault_creates} doesn't match the {materialized} repos \
370 materialized: a planned create silently failed and repo_at would have masked the drift"
371 );
372 Trace {
373 seed,
374 steps,
375 snapshots,
376 }
377}
378
379fn arm_drops(harness: &Harness, ops: &[Planned]) -> Vec<KnotHostname> {
380 let hosts: Vec<KnotHostname> = ops
381 .iter()
382 .filter_map(|planned| match planned {
383 Planned::Probe {
384 stranger,
385 drop: true,
386 } => Some(harness.strangers[stranger.0].host.clone()),
387 _ => None,
388 })
389 .collect();
390 hosts.iter().for_each(|host| harness.faults.drop_host(host));
391 hosts
392}
393
394async fn run_op(
395 harness: &Harness,
396 repos: &[RepoDid],
397 round: RoundNumber,
398 index: OperationIndex,
399 planned: Planned,
400) -> OpResult {
401 let make = |op: &'static str, actor: String, fault: &'static str, outcome: Outcome| Step {
402 round,
403 index,
404 op,
405 actor,
406 fault,
407 outcome,
408 };
409
410 match planned {
411 Planned::Maintain { repo } => {
412 let repo = repo_at(repos, repo);
413 let outcome = match harness.maintain(repo) {
414 Ok(()) => Outcome::Answered {
415 status: HttpStatus::new(200),
416 body: 0,
417 },
418 Err(message) => Outcome::Answered {
419 status: HttpStatus::new(500),
420 body: fnv1a(message.as_bytes()),
421 },
422 };
423 OpResult {
424 step: make("maintain", "knot".to_string(), "none", outcome),
425 created: None,
426 }
427 }
428 Planned::Read { op, killed } => {
429 let request = read_request(repos, op);
430 if killed {
431 drive_kill(harness.router(), request.method, &request.uri, request.body).await;
432 return OpResult {
433 step: make(op.name(), request.actor, "killed", Outcome::Killed),
434 created: None,
435 };
436 }
437 let (status, body) = http_call(
438 harness.router(),
439 request.method,
440 &request.uri,
441 None,
442 request.body,
443 )
444 .await;
445 OpResult {
446 step: make(
447 op.name(),
448 request.actor,
449 "none",
450 Outcome::Answered {
451 status,
452 body: body_digest(&body),
453 },
454 ),
455 created: None,
456 }
457 }
458 Planned::Admin { op, skew } => {
459 let request = admin_request(harness, repos, op, skew, round, index);
460 let (status, body) = http_call(
461 harness.router(),
462 request.method,
463 &request.uri,
464 request.token.as_deref(),
465 request.body,
466 )
467 .await;
468 let created = match op {
469 AdminOp::CreateRepo(_) if status == HttpStatus::new(200) => repo_did_of(&body),
470 _ => None,
471 };
472 OpResult {
473 step: make(
474 op.name(),
475 request.actor,
476 if skew { "clock_skew" } else { "none" },
477 Outcome::Answered {
478 status,
479 body: body_digest(&body),
480 },
481 ),
482 created,
483 }
484 }
485 Planned::Probe { stranger, drop } => {
486 let request = probe_request(harness, stranger, round, index);
487 let (status, body) = http_call(
488 harness.router(),
489 request.method,
490 &request.uri,
491 request.token.as_deref(),
492 request.body,
493 )
494 .await;
495 OpResult {
496 step: make(
497 "probe",
498 request.actor,
499 if drop { "drop_identity" } else { "none" },
500 Outcome::Answered {
501 status,
502 body: body_digest(&body),
503 },
504 ),
505 created: None,
506 }
507 }
508 }
509}
510
511pub(crate) struct Request {
512 pub(crate) method: Method,
513 pub(crate) uri: String,
514 pub(crate) token: Option<String>,
515 pub(crate) body: Bytes,
516 pub(crate) actor: String,
517}
518
519fn read_request(repos: &[RepoDid], op: ReadOp) -> Request {
520 match op {
521 ReadOp::Version => get("/xrpc/sh.tangled.knot.version"),
522 ReadOp::Owner => get("/xrpc/sh.tangled.owner"),
523 ReadOp::ListMembers => {
524 get("/xrpc/sh.tangled.knot.listMembers?subject=did:web:knot.nel.pet")
525 }
526 ReadOp::DidJson => get("/.well-known/did.json"),
527 ReadOp::InfoRefs(repo) => Request {
528 method: Method::GET,
529 uri: format!(
530 "/{}/info/refs?service=git-upload-pack",
531 repo_at(repos, repo).as_str()
532 ),
533 token: None,
534 body: Bytes::new(),
535 actor: "anon".to_string(),
536 },
537 ReadOp::Branches(repo) => repo_get("branches", "repo", repo_at(repos, repo)),
538 ReadOp::Log(repo) => repo_get("log", "repo", repo_at(repos, repo)),
539 ReadOp::DescribeRepo(repo) => repo_get("describeRepo", "repoDid", repo_at(repos, repo)),
540 ReadOp::Tree(repo) => repo_get("tree", "repo", repo_at(repos, repo)),
541 ReadOp::Languages(repo) => repo_get("languages", "repo", repo_at(repos, repo)),
542 ReadOp::Blob(repo) => Request {
543 method: Method::GET,
544 uri: format!(
545 "/xrpc/sh.tangled.repo.blob?repo={}&path=README.md",
546 enc(repo_at(repos, repo).as_str())
547 ),
548 token: None,
549 body: Bytes::new(),
550 actor: "anon".to_string(),
551 },
552 }
553}
554
555fn admin_request(
556 harness: &Harness,
557 repos: &[RepoDid],
558 op: AdminOp,
559 skew: bool,
560 round: RoundNumber,
561 index: OperationIndex,
562) -> Request {
563 let subjects = &harness.subjects;
564 match op {
565 AdminOp::AddMember(subject) => admin_post(
566 harness,
567 "addMember",
568 "sh.tangled.knot.addMember",
569 json!({ "subject": subjects[subject.0].as_str() }),
570 skew,
571 round,
572 index,
573 ),
574 AdminOp::RemoveMember(subject) => admin_post(
575 harness,
576 "removeMember",
577 "sh.tangled.knot.removeMember",
578 json!({ "subject": subjects[subject.0].as_str() }),
579 skew,
580 round,
581 index,
582 ),
583 AdminOp::Ban(subject) => admin_post(
584 harness,
585 "ban",
586 "sh.tangled.knot.ban",
587 json!({ "subject": subjects[subject.0].as_str() }),
588 skew,
589 round,
590 index,
591 ),
592 AdminOp::Unban(subject) => admin_post(
593 harness,
594 "unban",
595 "sh.tangled.knot.unban",
596 json!({ "subject": subjects[subject.0].as_str() }),
597 skew,
598 round,
599 index,
600 ),
601 AdminOp::CreateRepo(key) => {
602 let name = format!("repo{key}");
603 admin_post(
604 harness,
605 "create",
606 "sh.tangled.repo.create",
607 json!({ "rkey": name, "name": name }),
608 skew,
609 round,
610 index,
611 )
612 }
613 AdminOp::AddCollaborator(repo, subject) => admin_post(
614 harness,
615 "addCollaborator",
616 "sh.tangled.repo.addCollaborator",
617 json!({
618 "repo": repo_at(repos, repo).as_str(),
619 "subject": subjects[subject.0].as_str(),
620 }),
621 skew,
622 round,
623 index,
624 ),
625 }
626}
627
628fn probe_request(
629 harness: &Harness,
630 stranger: StrangerIndex,
631 round: RoundNumber,
632 index: OperationIndex,
633) -> Request {
634 let actor = &harness.strangers[stranger.0];
635 let token = mint(
636 &actor.signer,
637 &actor.did,
638 &harness.knot_aud,
639 "sh.tangled.knot.addMember",
640 jwt_window(harness, false),
641 round,
642 index,
643 );
644 Request {
645 method: Method::POST,
646 uri: "/xrpc/sh.tangled.knot.addMember".to_string(),
647 token: Some(token),
648 body: encode_body(json!({ "subject": harness.subjects[0].as_str() })),
649 actor: actor.host.to_string(),
650 }
651}
652
653fn get(path: &str) -> Request {
654 Request {
655 method: Method::GET,
656 uri: path.to_string(),
657 token: None,
658 body: Bytes::new(),
659 actor: "anon".to_string(),
660 }
661}
662
663fn repo_get(method: &str, param: &str, repo: &RepoDid) -> Request {
664 Request {
665 method: Method::GET,
666 uri: format!(
667 "/xrpc/sh.tangled.repo.{method}?{param}={}",
668 enc(repo.as_str())
669 ),
670 token: None,
671 body: Bytes::new(),
672 actor: "anon".to_string(),
673 }
674}
675
676fn admin_post(
677 harness: &Harness,
678 method_short: &str,
679 nsid: &'static str,
680 body: Value,
681 skew: bool,
682 round: RoundNumber,
683 index: OperationIndex,
684) -> Request {
685 let admin = &harness.admin;
686 let token = mint(
687 &admin.signer,
688 &admin.did,
689 &harness.knot_aud,
690 nsid,
691 jwt_window(harness, skew),
692 round,
693 index,
694 );
695 Request {
696 method: Method::POST,
697 uri: format!("/xrpc/{nsid}"),
698 token: Some(token),
699 body: encode_body(body),
700 actor: format!("admin:{method_short}"),
701 }
702}
703
704pub(crate) fn jwt_window(harness: &Harness, skew: bool) -> (UnixSeconds, UnixSeconds) {
705 let now = harness.now_seconds();
706 if skew {
707 (
708 now.saturating_sub_secs(SKEW_BACKDATE_SECS + SKEW_LIFETIME_SECS),
709 now.saturating_sub_secs(SKEW_BACKDATE_SECS),
710 )
711 } else {
712 (now, now.saturating_add_secs(60))
713 }
714}
715
716pub(crate) fn mint(
717 signer: &K256Signer,
718 issuer: &AccountDid,
719 aud: &KnotId,
720 nsid: &str,
721 window: (UnixSeconds, UnixSeconds),
722 round: RoundNumber,
723 index: OperationIndex,
724) -> String {
725 let header = URL_SAFE_NO_PAD.encode(br#"{"alg":"ES256K","typ":"JWT"}"#);
726 let payload = URL_SAFE_NO_PAD.encode(
727 serde_json::to_vec(&json!({
728 "iss": issuer.as_str(),
729 "aud": aud.as_str(),
730 "iat": window.0.get(),
731 "exp": window.1.get(),
732 "jti": format!("sim-{}-{}", round.get(), index.get()),
733 "lxm": nsid,
734 }))
735 .expect("claims serialize"),
736 );
737 let signing_input = format!("{header}.{payload}");
738 let signature = signer.sign(signing_input.as_bytes());
739 format!(
740 "{signing_input}.{}",
741 URL_SAFE_NO_PAD.encode(signature.as_bytes())
742 )
743}
744
745pub(crate) async fn http_call(
746 router: axum::Router,
747 method: Method,
748 uri: &str,
749 token: Option<&str>,
750 body: Bytes,
751) -> (HttpStatus, Bytes) {
752 let mut request = http::Request::builder()
753 .method(method)
754 .uri(uri)
755 .body(axum::body::Body::from(body))
756 .expect("request builds");
757 if let Some(token) = token {
758 request.headers_mut().insert(
759 AUTHORIZATION,
760 http::HeaderValue::from_str(&format!("Bearer {token}")).expect("bearer header"),
761 );
762 }
763 request
764 .extensions_mut()
765 .insert(axum::extract::ConnectInfo(SocketAddr::from((
766 [127, 0, 0, 1],
767 4242,
768 ))));
769 let response = router.oneshot(request).await.expect("router answers");
770 let status = HttpStatus::from(response.status());
771 let bytes = axum::body::to_bytes(response.into_body(), usize::MAX)
772 .await
773 .expect("response body");
774 (status, bytes)
775}
776
777pub(crate) async fn drive_kill(router: axum::Router, method: Method, uri: &str, body: Bytes) {
778 let call = http_call(router, method, uri, None, body);
779 futures::pin_mut!(call);
780 tokio::select! {
781 biased;
782 _ = &mut call => {}
783 _ = tokio::task::yield_now() => {}
784 }
785}
786
787pub(crate) fn repo_did_of(body: &Bytes) -> Option<RepoDid> {
788 serde_json::from_slice::<Value>(body)
789 .ok()
790 .and_then(|value| {
791 value
792 .get("repoDid")
793 .and_then(Value::as_str)
794 .map(str::to_string)
795 })
796 .and_then(|did| RepoDid::new(did).ok())
797}
798
799pub(crate) fn body_digest(body: &Bytes) -> u64 {
800 match serde_json::from_slice::<Value>(body) {
801 Ok(mut value) => {
802 canonicalize(&mut value);
803 fnv1a(&serde_json::to_vec(&value).expect("canonical body serializes"))
804 }
805 Err(_) => fnv1a(body),
806 }
807}
808
809fn canonicalize(value: &mut Value) {
810 match value {
811 Value::Array(items) => {
812 items.iter_mut().for_each(canonicalize);
813 items.sort_by_cached_key(|item| serde_json::to_string(item).expect("array item"));
814 }
815 Value::Object(map) => map.values_mut().for_each(canonicalize),
816 _ => {}
817 }
818}
819
820pub(crate) fn encode_body(value: Value) -> Bytes {
821 Bytes::from(serde_json::to_vec(&value).expect("request body serializes"))
822}
823
824pub(crate) fn enc(did: &str) -> String {
825 did.replace(':', "%3A")
826}
827
828fn repo_at(repos: &[RepoDid], index: RepoIndex) -> &RepoDid {
829 repos.get(index.0).unwrap_or_else(|| {
830 panic!(
831 "plan/execute repo drift: index {} exceeds {} repos created so far",
832 index.0,
833 repos.len()
834 )
835 })
836}