1use std::fs::{self, File, OpenOptions, TryLockError};
2use std::io::{self, Read, Write};
3use std::path::{Component, Path, PathBuf};
4use std::thread;
5use std::time::{Duration, Instant};
6
7use sha2::{Digest, Sha256};
8use ureq::unversioned::resolver::DefaultResolver;
9use ureq::unversioned::transport::{Connector, DefaultConnector};
10
11use crate::cancel::CancelToken;
12use crate::digest_hex;
13use crate::error::FetchError;
14use crate::poll::BodyPoll;
15use crate::spec::{FetchProgress, FetchSpec, Request};
16
17const CONNECT_TIMEOUT: Duration = Duration::from_secs(10);
18const STALL_TIMEOUT: Duration = Duration::from_secs(30);
19const POLL_INTERVAL: Duration = Duration::from_millis(200);
20const BODY_BUDGET: Duration = Duration::from_secs(24 * 60 * 60);
21const BACKOFF: [Duration; 3] = [
22 Duration::from_secs(1),
23 Duration::from_secs(4),
24 Duration::from_secs(9),
25];
26const CHUNK_SIZE: usize = 64 * 1024;
27const PROGRESS_INTERVAL: Duration = Duration::from_millis(100);
28const SLEEP_STEP: Duration = Duration::from_millis(50);
29const ABSOLUTE_BYTE_CAP: u64 = 8 * 1024 * 1024 * 1024;
30const MIN_HINT_CAP: u64 = 1024 * 1024;
31
32#[derive(Debug, Clone, Copy)]
33pub struct Timings {
34 pub connect_timeout: Duration,
35 pub stall_timeout: Duration,
36 pub poll_interval: Duration,
37 pub backoff: [Duration; 3],
38}
39
40impl Default for Timings {
41 fn default() -> Self {
42 Timings {
43 connect_timeout: CONNECT_TIMEOUT,
44 stall_timeout: STALL_TIMEOUT,
45 poll_interval: POLL_INTERVAL,
46 backoff: BACKOFF,
47 }
48 }
49}
50
51pub struct Fetcher {
52 dest_root: PathBuf,
53 agent: ureq::Agent,
54 timings: Timings,
55}
56
57enum AttemptError {
58 Cancelled,
59 Offline,
60 Stalled,
61 Http(u16),
62 Io(io::Error),
63 Superseded,
64}
65
66enum Download {
67 Done,
68 Superseded,
69}
70
71enum LockError {
72 Cancelled,
73 Io(io::Error),
74}
75
76impl From<LockError> for AttemptError {
77 fn from(err: LockError) -> Self {
78 match err {
79 LockError::Cancelled => AttemptError::Cancelled,
80 LockError::Io(source) => AttemptError::Io(source),
81 }
82 }
83}
84
85struct PartWriter<'a> {
86 file: File,
87 path: &'a Path,
88 len: u64,
89}
90
91impl<'a> PartWriter<'a> {
92 fn open(
93 path: &'a Path,
94 resume_offset: u64,
95 cancel: &CancelToken,
96 ) -> Result<Self, AttemptError> {
97 let file = if resume_offset > 0 {
98 OpenOptions::new().append(true).open(path)
99 } else {
100 OpenOptions::new().create(true).append(true).open(path)
101 }
102 .map_err(AttemptError::Io)?;
103 let writer = PartWriter {
104 file,
105 path,
106 len: resume_offset,
107 };
108 if resume_offset == 0 {
109 with_part_lock(&writer.file, writer.path, None, cancel, |file| {
110 file.set_len(0)
111 })?;
112 }
113 Ok(writer)
114 }
115
116 fn write(&mut self, buf: &[u8], cancel: &CancelToken) -> Result<(), AttemptError> {
117 with_part_lock(&self.file, self.path, Some(self.len), cancel, |mut file| {
118 file.write_all(buf)
119 })?;
120 self.len += buf.len() as u64;
121 Ok(())
122 }
123
124 fn finish(self) -> Result<(), AttemptError> {
125 self.file.sync_all().map_err(AttemptError::Io)
126 }
127}
128
129struct StreamLimits {
130 stall_timeout: Duration,
131 byte_cap: u64,
132}
133
134#[derive(Clone, Copy)]
135struct RequestRef<'a> {
136 url: &'a str,
137 dest_rel: &'a str,
138 sha256: &'a str,
139 size_hint: Option<u64>,
140}
141
142impl Fetcher {
143 pub fn new(dest_root: impl Into<PathBuf>) -> Self {
144 Self::with_timings(dest_root, Timings::default())
145 }
146
147 pub fn with_timings(dest_root: impl Into<PathBuf>, timings: Timings) -> Self {
148 let config = ureq::Agent::config_builder()
149 .timeout_connect(Some(timings.connect_timeout))
150 .timeout_send_request(Some(timings.stall_timeout))
151 .timeout_recv_response(Some(timings.stall_timeout))
152 .timeout_recv_body(Some(BODY_BUDGET))
153 .http_status_as_error(false)
154 .max_redirects(10)
155 .build();
156 Fetcher {
157 dest_root: dest_root.into(),
158 agent: ureq::Agent::with_parts(
159 config,
160 DefaultConnector::new().chain(BodyPoll(timings.poll_interval)),
161 DefaultResolver::default(),
162 ),
163 timings,
164 }
165 }
166
167 pub fn fetch(
168 &self,
169 spec: &FetchSpec,
170 cancel: &CancelToken,
171 mut progress: impl FnMut(FetchProgress),
172 ) -> Result<PathBuf, FetchError> {
173 self.fetch_inner(
174 RequestRef {
175 url: spec.url,
176 dest_rel: spec.dest_rel,
177 sha256: spec.sha256,
178 size_hint: spec.size_hint,
179 },
180 cancel,
181 &mut progress,
182 )
183 }
184
185 pub fn fetch_request(
186 &self,
187 request: &Request,
188 cancel: &CancelToken,
189 mut progress: impl FnMut(FetchProgress),
190 ) -> Result<PathBuf, FetchError> {
191 self.fetch_inner(
192 RequestRef {
193 url: &request.url,
194 dest_rel: &request.dest_rel,
195 sha256: &request.sha256,
196 size_hint: request.size_hint,
197 },
198 cancel,
199 &mut progress,
200 )
201 }
202
203 fn fetch_inner(
204 &self,
205 request: RequestRef<'_>,
206 cancel: &CancelToken,
207 progress: &mut dyn FnMut(FetchProgress),
208 ) -> Result<PathBuf, FetchError> {
209 if cancel.is_cancelled() {
210 return Err(FetchError::Cancelled);
211 }
212 let final_path = self.contained_dest(request.dest_rel)?;
213 let part_path = part_path_of(&final_path);
214 if let Some(parent) = final_path.parent() {
215 fs::create_dir_all(parent).map_err(|source| FetchError::Io {
216 path: parent.to_path_buf(),
217 source,
218 })?;
219 }
220
221 loop {
222 match self.download_with_retries(request, &part_path, cancel, progress)? {
223 Download::Done => {
224 return self.verify_and_rename(request, &part_path, &final_path, cancel);
225 }
226 Download::Superseded => {
227 if let Some(path) = self.follow(request, &part_path, &final_path, cancel)? {
228 return Ok(path);
229 }
230 }
231 }
232 }
233 }
234
235 fn follow(
236 &self,
237 request: RequestRef<'_>,
238 part_path: &Path,
239 final_path: &Path,
240 cancel: &CancelToken,
241 ) -> Result<Option<PathBuf>, FetchError> {
242 let mut last_len = None;
243 let mut quiet_since = Instant::now();
244 loop {
245 sleep_interruptible(self.timings.poll_interval, cancel)?;
246 match fs::metadata(part_path) {
247 Ok(meta) if last_len == Some(meta.len()) => {
248 if quiet_since.elapsed() >= self.timings.stall_timeout {
249 return Ok(None);
250 }
251 }
252 Ok(meta) => {
253 last_len = Some(meta.len());
254 quiet_since = Instant::now();
255 }
256 Err(source) if source.kind() == io::ErrorKind::NotFound => {
257 return Ok(verify_completed(request, final_path).ok());
258 }
259 Err(source) => {
260 return Err(FetchError::Io {
261 path: part_path.to_path_buf(),
262 source,
263 });
264 }
265 }
266 }
267 }
268
269 fn contained_dest(&self, dest_rel: &str) -> Result<PathBuf, FetchError> {
270 let reject = || FetchError::Io {
271 path: PathBuf::from(dest_rel),
272 source: io::Error::new(
273 io::ErrorKind::InvalidInput,
274 "dest_rel must be a plain relative path inside dest_root",
275 ),
276 };
277 let mut contained = PathBuf::new();
278 for component in Path::new(dest_rel).components() {
279 match component {
280 Component::Normal(part) => contained.push(part),
281 Component::CurDir => {}
282 Component::ParentDir | Component::RootDir | Component::Prefix(_) => {
283 return Err(reject());
284 }
285 }
286 }
287 if contained.as_os_str().is_empty() {
288 return Err(reject());
289 }
290 Ok(self.dest_root.join(contained))
291 }
292
293 fn download_with_retries(
294 &self,
295 request: RequestRef<'_>,
296 part_path: &Path,
297 cancel: &CancelToken,
298 progress: &mut dyn FnMut(FetchProgress),
299 ) -> Result<Download, FetchError> {
300 for attempt in 0..=self.timings.backoff.len() {
301 if attempt > 0 {
302 sleep_interruptible(self.timings.backoff[attempt - 1], cancel)?;
303 }
304 match self.attempt(request, part_path, cancel, progress) {
305 Ok(()) => return Ok(Download::Done),
306 Err(AttemptError::Superseded) => return Ok(Download::Superseded),
307 Err(AttemptError::Cancelled) => return Err(FetchError::Cancelled),
308 Err(AttemptError::Offline) => {
309 return Err(FetchError::Offline {
310 url: request.url.to_string(),
311 });
312 }
313 Err(AttemptError::Stalled) => {
314 return Err(FetchError::Stalled {
315 url: request.url.to_string(),
316 });
317 }
318 Err(AttemptError::Http(status)) => {
319 return Err(FetchError::Http {
320 status,
321 url: request.url.to_string(),
322 });
323 }
324 Err(AttemptError::Io(_)) if attempt < self.timings.backoff.len() => {}
325 Err(AttemptError::Io(source)) => {
326 return Err(FetchError::Io {
327 path: part_path.to_path_buf(),
328 source,
329 });
330 }
331 }
332 }
333 unreachable!("loop always returns before exhausting its range")
334 }
335
336 fn attempt(
337 &self,
338 request: RequestRef<'_>,
339 part_path: &Path,
340 cancel: &CancelToken,
341 progress: &mut dyn FnMut(FetchProgress),
342 ) -> Result<(), AttemptError> {
343 if cancel.is_cancelled() {
344 return Err(AttemptError::Cancelled);
345 }
346 let existing_len = fs::metadata(part_path).map(|m| m.len()).unwrap_or(0);
347 let (resume_offset, status, headers, body) =
348 self.open_response(request, part_path, existing_len)?;
349 let total = total_size(&headers, status, resume_offset, request.size_hint);
350 let mut part = PartWriter::open(part_path, resume_offset, cancel)?;
351 let reader = body.into_reader();
352 let limits = StreamLimits {
353 stall_timeout: self.timings.stall_timeout,
354 byte_cap: byte_cap(request.size_hint),
355 };
356 stream_body(
357 reader,
358 &mut part,
359 resume_offset,
360 total,
361 limits,
362 cancel,
363 progress,
364 )?;
365 part.finish()
366 }
367
368 fn open_response(
369 &self,
370 request: RequestRef<'_>,
371 part_path: &Path,
372 existing_len: u64,
373 ) -> Result<(u64, u16, ureq::http::HeaderMap, ureq::Body), AttemptError> {
374 let mut want_range = existing_len > 0;
375 let mut offset = existing_len;
376 loop {
377 let mut request_builder = self.agent.get(request.url);
378 if want_range {
379 request_builder = request_builder.header("Range", format!("bytes={offset}-"));
380 }
381 let response = request_builder.call().map_err(classify_call_error)?;
382 let status = response.status().as_u16();
383 let (parts, body) = response.into_parts();
384 match status {
385 200 if want_range => {
386 let _ = fs::remove_file(part_path);
387 want_range = false;
388 offset = 0;
389 continue;
390 }
391 200 | 206 => return Ok((offset, status, parts.headers, body)),
392 416 if want_range => {
393 let _ = fs::remove_file(part_path);
394 want_range = false;
395 offset = 0;
396 continue;
397 }
398 other => return Err(AttemptError::Http(other)),
399 }
400 }
401 }
402
403 fn verify_and_rename(
404 &self,
405 request: RequestRef<'_>,
406 part_path: &Path,
407 final_path: &Path,
408 cancel: &CancelToken,
409 ) -> Result<PathBuf, FetchError> {
410 let file = match File::open(part_path) {
411 Ok(file) => file,
412 Err(source) if source.kind() == io::ErrorKind::NotFound => {
413 return verify_completed(request, final_path);
414 }
415 Err(source) => {
416 return Err(FetchError::Io {
417 path: part_path.to_path_buf(),
418 source,
419 });
420 }
421 };
422 match acquire_part_lock(&file, cancel) {
423 Ok(()) => {}
424 Err(LockError::Cancelled) => return Err(FetchError::Cancelled),
425 Err(LockError::Io(source)) => {
426 return Err(FetchError::Io {
427 path: part_path.to_path_buf(),
428 source,
429 });
430 }
431 }
432 if !part_still_ours(part_path, &file) {
433 return verify_completed(request, final_path);
434 }
435 let got = hash_reader(&file).map_err(|source| FetchError::Io {
436 path: part_path.to_path_buf(),
437 source,
438 })?;
439 let got_hex = digest_hex(&got);
440 if got_hex != request.sha256 {
441 let _ = fs::remove_file(part_path);
442 return Err(FetchError::ChecksumMismatch {
443 expected: request.sha256.to_string(),
444 got: got_hex,
445 });
446 }
447 fs::rename(part_path, final_path).map_err(|source| FetchError::Io {
448 path: final_path.to_path_buf(),
449 source,
450 })?;
451 Ok(final_path.to_path_buf())
452 }
453}
454
455fn verify_completed(request: RequestRef<'_>, final_path: &Path) -> Result<PathBuf, FetchError> {
456 let got = hash_file(final_path).map_err(|source| FetchError::Io {
457 path: final_path.to_path_buf(),
458 source,
459 })?;
460 let got_hex = digest_hex(&got);
461 if got_hex == request.sha256 {
462 Ok(final_path.to_path_buf())
463 } else {
464 Err(FetchError::ChecksumMismatch {
465 expected: request.sha256.to_string(),
466 got: got_hex,
467 })
468 }
469}
470
471fn acquire_part_lock(file: &File, cancel: &CancelToken) -> Result<(), LockError> {
472 loop {
473 match file.try_lock() {
474 Ok(()) => return Ok(()),
475 Err(TryLockError::WouldBlock) => {
476 if cancel.is_cancelled() {
477 return Err(LockError::Cancelled);
478 }
479 thread::sleep(SLEEP_STEP);
480 }
481 Err(TryLockError::Error(source)) => return Err(LockError::Io(source)),
482 }
483 }
484}
485
486fn with_part_lock<T>(
487 file: &File,
488 part_path: &Path,
489 expected_len: Option<u64>,
490 cancel: &CancelToken,
491 op: impl FnOnce(&File) -> io::Result<T>,
492) -> Result<T, AttemptError> {
493 acquire_part_lock(file, cancel).map_err(AttemptError::from)?;
494 let ours = part_len_if_ours(part_path, file)
495 .is_some_and(|len| expected_len.is_none_or(|expected| expected == len));
496 let outcome = if ours {
497 op(file).map_err(AttemptError::Io)
498 } else {
499 Err(AttemptError::Superseded)
500 };
501 let _ = file.unlock();
502 outcome
503}
504
505fn part_path_of(final_path: &Path) -> PathBuf {
506 let mut os = final_path.as_os_str().to_os_string();
507 os.push(".part");
508 PathBuf::from(os)
509}
510
511fn total_size(
512 headers: &ureq::http::HeaderMap,
513 status: u16,
514 resume_offset: u64,
515 size_hint: Option<u64>,
516) -> Option<u64> {
517 if status == 206
518 && let Some(total) = headers
519 .get("content-range")
520 .and_then(|v| v.to_str().ok())
521 .and_then(parse_content_range_total)
522 {
523 return Some(total);
524 }
525 if let Some(len) = headers
526 .get("content-length")
527 .and_then(|v| v.to_str().ok())
528 .and_then(|s| s.parse::<u64>().ok())
529 {
530 return Some(resume_offset + len);
531 }
532 size_hint
533}
534
535fn parse_content_range_total(value: &str) -> Option<u64> {
536 value.rsplit('/').next()?.trim().parse().ok()
537}
538
539fn sleep_interruptible(total: Duration, cancel: &CancelToken) -> Result<(), FetchError> {
540 let mut waited = Duration::ZERO;
541 while waited < total {
542 if cancel.is_cancelled() {
543 return Err(FetchError::Cancelled);
544 }
545 let step = SLEEP_STEP.min(total - waited);
546 thread::sleep(step);
547 waited += step;
548 }
549 if cancel.is_cancelled() {
550 return Err(FetchError::Cancelled);
551 }
552 Ok(())
553}
554
555fn classify_call_error(err: ureq::Error) -> AttemptError {
556 match err {
557 ureq::Error::Timeout(ureq::Timeout::Connect) => AttemptError::Offline,
558 ureq::Error::Timeout(ureq::Timeout::Resolve) => AttemptError::Offline,
559 ureq::Error::Timeout(_) => AttemptError::Stalled,
560 _ => AttemptError::Offline,
561 }
562}
563
564fn byte_cap(size_hint: Option<u64>) -> u64 {
565 size_hint
566 .map(|hint| hint.saturating_mul(2).max(MIN_HINT_CAP))
567 .unwrap_or(ABSOLUTE_BYTE_CAP)
568}
569
570#[cfg(unix)]
571fn identity_of(meta: &fs::Metadata) -> (u64, u64) {
572 use std::os::unix::fs::MetadataExt;
573 (meta.dev(), meta.ino())
574}
575
576#[cfg(not(unix))]
577fn identity_of(_meta: &fs::Metadata) -> (u64, u64) {
578 (0, 0)
579}
580
581fn part_still_ours(part_path: &Path, file: &File) -> bool {
582 part_len_if_ours(part_path, file).is_some()
583}
584
585fn part_len_if_ours(part_path: &Path, file: &File) -> Option<u64> {
586 match (fs::metadata(part_path), file.metadata()) {
587 (Ok(on_disk), Ok(ours)) if identity_of(&on_disk) == identity_of(&ours) => Some(ours.len()),
588 _ => None,
589 }
590}
591
592fn stream_body(
593 mut reader: impl Read,
594 part: &mut PartWriter<'_>,
595 resume_offset: u64,
596 total: Option<u64>,
597 limits: StreamLimits,
598 cancel: &CancelToken,
599 progress: &mut dyn FnMut(FetchProgress),
600) -> Result<(), AttemptError> {
601 let mut buf = [0u8; CHUNK_SIZE];
602 let start = Instant::now();
603 let mut last_data = Instant::now();
604 let mut last_emit = start.checked_sub(PROGRESS_INTERVAL).unwrap_or(start);
605 let mut received = resume_offset;
606 loop {
607 if cancel.is_cancelled() {
608 return Err(AttemptError::Cancelled);
609 }
610 match reader.read(&mut buf) {
611 Ok(0) => return Ok(()),
612 Ok(n) => {
613 part.write(&buf[..n], cancel)?;
614 received += n as u64;
615 if received >= limits.byte_cap {
616 return Ok(());
617 }
618 last_data = Instant::now();
619 if last_emit.elapsed() >= PROGRESS_INTERVAL {
620 let elapsed = start.elapsed().as_secs_f64().max(0.001);
621 let sent = received.saturating_sub(resume_offset);
622 let speed_bps = (sent as f64 / elapsed) as u64;
623 progress(FetchProgress {
624 received,
625 total,
626 speed_bps,
627 });
628 last_emit = Instant::now();
629 }
630 }
631 Err(err) => {
632 if cancel.is_cancelled() {
633 return Err(AttemptError::Cancelled);
634 }
635 if is_recv_body_poll_timeout(&err) {
636 if last_data.elapsed() >= limits.stall_timeout {
637 return Err(AttemptError::Stalled);
638 }
639 continue;
640 }
641 return Err(AttemptError::Io(err));
642 }
643 }
644 }
645}
646
647fn is_recv_body_poll_timeout(err: &io::Error) -> bool {
648 err.get_ref()
649 .and_then(|inner| inner.downcast_ref::<ureq::Error>())
650 .map(|ureq_err| matches!(ureq_err, ureq::Error::Timeout(ureq::Timeout::RecvBody)))
651 .unwrap_or(false)
652}
653
654fn hash_file(path: &Path) -> io::Result<[u8; 32]> {
655 let file = File::open(path)?;
656 hash_reader(&file)
657}
658
659fn hash_reader(mut file: &File) -> io::Result<[u8; 32]> {
660 let mut hasher = Sha256::new();
661 let mut buf = [0u8; CHUNK_SIZE];
662 loop {
663 let n = file.read(&mut buf)?;
664 if n == 0 {
665 break;
666 }
667 hasher.update(&buf[..n]);
668 }
669 Ok(hasher.finalize().into())
670}
671
672#[cfg(test)]
673mod owned_request_tests {
674 use super::*;
675 use crate::{Request, sha256_hex};
676 use std::net::TcpListener;
677
678 fn serve(body: Vec<u8>, requests: usize) -> String {
679 let listener = TcpListener::bind("127.0.0.1:0").unwrap();
680 let address = listener.local_addr().unwrap();
681 std::thread::spawn(move || {
682 for stream in listener.incoming().take(requests) {
683 let mut stream = stream.unwrap();
684 let mut request = [0; 1024];
685 let _ = stream.read(&mut request);
686 write!(
687 stream,
688 "HTTP/1.1 200 OK\r\nContent-Length: {}\r\nConnection: close\r\n\r\n",
689 body.len()
690 )
691 .unwrap();
692 stream.write_all(&body).unwrap();
693 }
694 });
695 format!("http://{address}/asset")
696 }
697
698 #[test]
699 fn owned_request_matches_static_spec_and_checksum_refusal() {
700 let scratch = Path::new(env!("CARGO_MANIFEST_DIR")).join("tmp/testscratch");
701 fs::create_dir_all(&scratch).unwrap();
702 let root = tempfile::Builder::new()
703 .prefix("fetch-owned-")
704 .tempdir_in(scratch)
705 .unwrap();
706 let body = b"runtime catalog asset".to_vec();
707 let good_url = Box::leak(serve(body.clone(), 2).into_boxed_str());
708 let digest = Box::leak(sha256_hex(&body).into_boxed_str());
709 let static_spec = FetchSpec {
710 url: good_url,
711 dest_rel: "static.bin",
712 sha256: digest,
713 size_hint: Some(body.len() as u64),
714 };
715 let owned = Request {
716 url: good_url.to_string(),
717 dest_rel: "owned.bin".into(),
718 sha256: digest.to_string(),
719 size_hint: Some(body.len() as u64),
720 };
721 let fetcher = Fetcher::new(root.path());
722 let static_path = fetcher
723 .fetch(&static_spec, &CancelToken::new(), |_| {})
724 .unwrap();
725 let owned_path = fetcher
726 .fetch_request(&owned, &CancelToken::new(), |_| {})
727 .unwrap();
728 assert_eq!(
729 fs::read(static_path).unwrap(),
730 fs::read(owned_path).unwrap()
731 );
732
733 let bad_url = Box::leak(serve(body, 2).into_boxed_str());
734 let bad_spec = FetchSpec {
735 url: bad_url,
736 dest_rel: "bad-static.bin",
737 sha256: "00",
738 size_hint: None,
739 };
740 let bad_owned = Request {
741 url: bad_url.to_string(),
742 dest_rel: "bad-owned.bin".into(),
743 sha256: "00".into(),
744 size_hint: None,
745 };
746 assert!(matches!(
747 fetcher.fetch(&bad_spec, &CancelToken::new(), |_| {}),
748 Err(FetchError::ChecksumMismatch { .. })
749 ));
750 assert!(matches!(
751 fetcher.fetch_request(&bad_owned, &CancelToken::new(), |_| {}),
752 Err(FetchError::ChecksumMismatch { .. })
753 ));
754 }
755}