Something went wrong. Try again.
Monorepo for Tangled
Something went wrong. Try again.
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603604605606607608609610611612613614615616617618619620621622623624625626627628629630631632633634635636637638639640641642643644645646647648649650651652653654655656657658659660661662663664665666667668669670671672673674675676677678679680681682683684685686687688689690691692693694695696697698699700701702703704705706707708709710711712713714715716717718719720721722723724725726727728729730731732733734735736737738739740741742743744745746747748749750751752753754755756757758759760761762763764765766767768769770771772773774775776777778779780781782783784785786787788789790791792793794795796797798799800801802803804805806807808809810811812813814815816817818819820821822823824825826827828829830831832833834835836837838839840841842843844845846847848849850851852853854855856857858859860861862863864865866867868869870871872873874875876877878879880881882883884885886887888889890891892893894895896897use 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, HttpStatus, KnotHostname, RepoDid, UnixSeconds};
use crate::harness::{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)]struct Observation { op: ReadOp, killed: bool,}
#[derive(Clone, Copy)]enum Planned { Mutate(Mutation), 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<Mutation>), Observe(Vec<Observation>),}
impl Phase { fn ops(&self) -> Vec<Planned> { match self { Phase::Mutate(ops) => ops.iter().copied().map(Planned::Mutate).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<Round> { let rng = Rng::new(seed); let mut available: u32 = 1; let mut rkey: u32 = 0; let mut stranger: usize = 0; (0..rounds) .map(|round| { let (phase, fresh_repos) = match round % 2 { 0 => { let (ops, fresh_repos) = mutate_round(&rng, subjects, available, &mut rkey, &mut stranger); (Phase::Mutate(ops), fresh_repos) } _ => (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<AccountDid>, blocked: BTreeSet<AccountDid>, collaborators: BTreeMap<usize, BTreeSet<AccountDid>>,}
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.members.insert(subject_did(subject)); } AdminOp::RemoveMember(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 .collaborators .entry(repo.0) .or_default() .insert(subject_did(subject)); } AdminOp::CreateRepo(_) => {} }, 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,) -> (Vec<Mutation>, u32) { let mut ops: Vec<Mutation> = (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); Mutation::Admin { op: AdminOp::AddCollaborator(repo, subject), skew: rng.chance(1, 6), } } 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 observe_round(rng: &Rng, available: u32) -> Vec<Observation> { let mut ops: Vec<Observation> = [ 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<RepoDid>,}
pub(crate) async fn execute(harness: Arc<Harness>, seed: u64, plan: Vec<Round>) -> Trace { let initial = ( vec![harness.seed_repo.clone()], Vec::<Step>::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::<Vec<_>>(); let results: Vec<OpResult> = 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<RepoDid> = 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<KnotHostname> { let hosts: Vec<KnotHostname> = 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, } } }}
pub(crate) struct Request { pub(crate) method: Method, pub(crate) uri: String, pub(crate) token: Option<String>, 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 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<str>, 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<RepoDid> { serde_json::from_slice::<Value>(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::<Value>(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() ) })}