alice.pds.demo.boltless.dev
This repository has no description
did:plc:3fwecdnvtcscjnrx2p4n7a…
1mod common;
2
3use axum::body::Body;
4use futures::StreamExt;
5use http::{Request, StatusCode, header};
6use sha2::{Digest, Sha256};
7use tower::ServiceExt;
8
9use knot_lfs::{LfsOid, LfsStore};
10use knot_types::RepoDid;
11
12use common::{World, empty_repo};
13
14const OBJECT_BYTES: usize = 8 * 1024 * 1024;
15const OBJECTS: usize = 4;
16const DOWNLOADERS: usize = 12;
17const ROUNDS: usize = 4;
18const PAGE_BYTES: u64 = 4096;
19const PEAK_CEILING: u64 = 512 * 1024 * 1024;
20const GROWTH_SLACK: u64 = 64 * 1024 * 1024;
21
22fn rss_bytes() -> u64 {
23 let statm = std::fs::read_to_string("/proc/self/statm").expect("/proc/self/statm is readable");
24 statm
25 .split_whitespace()
26 .nth(1)
27 .and_then(|pages| pages.parse::<u64>().ok())
28 .map(|pages| pages * PAGE_BYTES)
29 .expect("statm lists the resident page count")
30}
31
32fn incompressible(len: usize, seed: u64) -> Vec<u8> {
33 let mut state = seed | 1;
34 (0..len)
35 .map(|_| {
36 state ^= state << 13;
37 state ^= state >> 7;
38 state ^= state << 17;
39 (state & 0xff) as u8
40 })
41 .collect()
42}
43
44fn seed_objects(world: &World, repo: &RepoDid) -> Vec<LfsOid> {
45 let store = &world.state.lfs.as_ref().unwrap().handle.store;
46 (0..OBJECTS)
47 .map(|index| {
48 let body = incompressible(OBJECT_BYTES, 0x5eed_0000 + index as u64);
49 let oid = LfsOid::from_digest(Sha256::digest(&body).into());
50 store
51 .put(
52 repo,
53 &oid,
54 knot_lfs::ClaimedSize::new(body.len() as u64),
55 &mut &body[..],
56 )
57 .unwrap();
58 oid
59 })
60 .collect()
61}
62
63async fn download(world: &World, did: &RepoDid, oid: &LfsOid, tag: &str) {
64 let request = Request::get(format!("/{did}/info/lfs/objects/{oid}"))
65 .body(Body::empty())
66 .unwrap();
67 let response = world.router.clone().oneshot(request).await.unwrap();
68 assert_eq!(response.status(), StatusCode::OK, "{tag}");
69 let (streamed, hasher) = response
70 .into_body()
71 .into_data_stream()
72 .fold(
73 (0u64, Sha256::new()),
74 |(streamed, mut hasher), chunk| async move {
75 let chunk = chunk.unwrap();
76 hasher.update(&chunk);
77 (streamed + chunk.len() as u64, hasher)
78 },
79 )
80 .await;
81 assert_eq!(streamed, OBJECT_BYTES as u64, "{tag}");
82 assert_eq!(
83 LfsOid::from_digest(hasher.finalize().into()),
84 oid.clone(),
85 "{tag}: downloaded bytes must hash to the requested oid"
86 );
87}
88
89async fn batch(world: &World, did: &RepoDid, oids: &[LfsOid], tag: &str) {
90 let objects: Vec<serde_json::Value> = oids
91 .iter()
92 .map(|oid| serde_json::json!({"oid": oid.as_str(), "size": OBJECT_BYTES}))
93 .collect();
94 let body = serde_json::json!({"operation": "download", "objects": objects}).to_string();
95 let request = Request::post(format!("/{did}/info/lfs/objects/batch"))
96 .header(header::CONTENT_TYPE, "application/vnd.git-lfs+json")
97 .body(Body::from(body))
98 .unwrap();
99 let response = world.router.clone().oneshot(request).await.unwrap();
100 let status = response.status();
101 let body = axum::body::to_bytes(response.into_body(), usize::MAX)
102 .await
103 .unwrap();
104 assert_eq!(
105 status,
106 StatusCode::OK,
107 "{tag}: {}",
108 String::from_utf8_lossy(&body)
109 );
110}
111
112#[tokio::test(flavor = "multi_thread")]
113async fn sustained_anonymous_downloads_stay_bounded_and_leak_nothing() {
114 let world = World::unshed();
115 let (did, _bare, _work) = empty_repo(&world, "nautilus");
116 let oids = seed_objects(&world, &did);
117
118 let storm = |round: usize| {
119 let world = &world;
120 let did = &did;
121 let oids = &oids;
122 async move {
123 futures::future::join_all((0..DOWNLOADERS).map(|task| {
124 let oid = oids[task % OBJECTS].clone();
125 async move {
126 let tag = format!("round {round} task {task}");
127 batch(world, did, oids, &tag).await;
128 download(world, did, &oid, &tag).await;
129 }
130 }))
131 .await;
132 }
133 };
134
135 storm(0).await;
136 let settled = rss_bytes();
137
138 let peaks: Vec<u64> = futures::stream::iter(1..ROUNDS)
139 .then(|round| {
140 let storm = &storm;
141 async move {
142 storm(round).await;
143 rss_bytes()
144 }
145 })
146 .collect()
147 .await;
148
149 let peak = peaks.iter().copied().max().unwrap_or(settled);
150 assert!(
151 peak < PEAK_CEILING,
152 "concurrent downloads peaked at {peak} bytes, ceiling {PEAK_CEILING}"
153 );
154 let last = *peaks.last().unwrap_or(&settled);
155 assert!(
156 last <= settled + GROWTH_SLACK,
157 "rss grew from {settled} to {last} across rounds, downloads are leaking"
158 );
159}