use crate::{
credentials::Credentials,
download_metadata::{DownloadMetadata, FileChecksum, PartDetails, ResponseHeader},
error::MetadataError,
fs_utils,
hash::HashDigest,
response_info::ResponseInfo,
};
use chrono::{DateTime, Utc};
use derive_builder::{Builder, UninitializedFieldError};
use reqwest::{
Proxy, Url,
header::{HeaderMap, HeaderName, HeaderValue},
};
use std::{
collections::HashMap,
path::{self, PathBuf},
str::FromStr,
};
use thiserror::Error;
use tokio::sync::Semaphore;
use ulid::Ulid;
#[derive(Builder, Debug, Clone)]
#[builder(build_fn(validate = "Self::validate", error = "DownloadBuilderError"))]
pub struct Download {
download_dir: path::PathBuf,
url: Url,
#[builder(default = false)]
is_resumable: bool,
#[builder(default = false)]
use_server_time: bool,
filename: String,
save_dir: path::PathBuf,
#[builder(default = None)]
size: Option<u64>,
#[builder(default = Vec::new())]
checksums: Vec<HashDigest>,
#[builder(default = None)]
etag: Option<String>,
#[builder(default = None)]
last_modified: Option<i64>,
#[builder(default = false)]
requires_auth: bool,
#[builder(default = false)]
requires_basic_auth: bool,
#[builder(default = None)]
credentials: Option<Credentials>,
#[builder(default = None)]
proxy: Option<Proxy>,
#[builder(default = None)]
headers: Option<HeaderMap>,
#[builder(default = None)]
response_headers: Option<HeaderMap>,
#[builder(default = None)]
response_headers_probed_at: Option<i64>,
#[builder(default = 6)]
max_connections: u64,
parts: HashMap<String, PartDetails>,
#[builder(default = false)]
finished: bool,
}
const RESPONSE_HEADER_DENYLIST: &[&str] = &[
"set-cookie",
"set-cookie2",
"www-authenticate",
"proxy-authenticate",
"authentication-info",
"proxy-authentication-info",
"authorization",
"proxy-authorization",
];
const RESPONSE_HEADER_SECRET_MARKERS: &[&str] = &[
"auth",
"cookie",
"credential",
"key",
"password",
"secret",
"session",
"signature",
"token",
];
const MAX_STORED_RESPONSE_HEADERS_BYTES: usize = 8 * 1024;
const MAX_STORED_RESPONSE_HEADER_VALUE_BYTES: usize = 1024;
fn is_secret_response_header(name: &str) -> bool {
RESPONSE_HEADER_DENYLIST.contains(&name)
|| RESPONSE_HEADER_SECRET_MARKERS
.iter()
.any(|marker| name.contains(marker))
}
#[derive(Debug, Clone, Copy)]
pub struct PartSplit {
pub new_left_size: u64,
pub new_right_offset: u64,
pub new_right_size: u64,
}
impl Download {
const METADATA_FILENAME: &'static str = "metadata.pb";
const METADATA_TEMP_FILENAME: &'static str = "metadata.pb.temp";
const LOCK_FILENAME: &'static str = "odl.lock";
pub const PART_EXTENSION: &'static str = "part";
pub const MIN_PART_SIZE: u64 = 300 * 1024; pub const UNKNOWN_PART_SIZE: u64 = u64::MAX;
pub const ASSEMBLY_CLUSTER_SIZE: u64 = 4096;
const _ASSERT_MIN_PART_GE_CLUSTER: () =
assert!(Self::MIN_PART_SIZE >= Self::ASSEMBLY_CLUSTER_SIZE);
pub fn download_dir(&self) -> &path::PathBuf {
&self.download_dir
}
pub fn part_path(&self, ulid: &str) -> path::PathBuf {
self.download_dir
.join(format!("{}.{}", ulid, Self::PART_EXTENSION))
}
pub fn set_download_dir(&mut self, path: PathBuf) {
self.download_dir = path
}
pub fn lockfile_path(&self) -> path::PathBuf {
self.download_dir.join(Self::LOCK_FILENAME)
}
pub fn metadata_path(&self) -> path::PathBuf {
self.download_dir.join(Self::METADATA_FILENAME)
}
pub fn metadata_temp_path(&self) -> path::PathBuf {
self.download_dir.join(Self::METADATA_TEMP_FILENAME)
}
pub fn final_file_path(&self) -> path::PathBuf {
self.save_dir.join(&self.filename)
}
pub fn url(&self) -> &Url {
&self.url
}
pub fn is_resumable(&self) -> bool {
self.is_resumable
}
pub fn use_server_time(&self) -> bool {
self.use_server_time
}
pub fn filename(&self) -> &str {
&self.filename
}
pub fn set_filename(&mut self, filename: String) {
self.filename = filename;
}
pub fn add_checksums(&mut self, extra: impl IntoIterator<Item = HashDigest>) {
for c in extra {
if !self.checksums.contains(&c) {
self.checksums.push(c);
}
}
}
pub fn save_dir(&self) -> &path::PathBuf {
&self.save_dir
}
pub fn set_save_dir(&mut self, path: PathBuf) {
self.save_dir = path
}
pub fn size(&self) -> Option<u64> {
self.size
}
pub fn etag(&self) -> &Option<String> {
&self.etag
}
pub fn last_modified(&self) -> Option<i64> {
self.last_modified
}
pub fn last_modified_as_date(&self) -> Option<DateTime<Utc>> {
self.last_modified
.and_then(|x| chrono::DateTime::from_timestamp(x, 0))
}
pub fn requires_auth(&self) -> bool {
self.requires_auth
}
pub fn requires_basic_auth(&self) -> bool {
self.requires_basic_auth
}
pub fn credentials(&self) -> &Option<Credentials> {
&self.credentials
}
pub fn proxy(&self) -> &Option<Proxy> {
&self.proxy
}
pub fn headers(&self) -> &Option<HeaderMap> {
&self.headers
}
pub fn response_headers(&self) -> Option<&HeaderMap> {
self.response_headers.as_ref()
}
pub fn response_headers_probed_at(&self) -> Option<i64> {
self.response_headers_probed_at
}
pub fn response_headers_probed_at_as_date(&self) -> Option<DateTime<Utc>> {
self.response_headers_probed_at
.and_then(|x| DateTime::from_timestamp(x, 0))
}
pub fn max_connections(&self) -> u64 {
self.max_connections
}
pub fn parts(&self) -> &HashMap<String, PartDetails> {
&self.parts
}
pub fn finished(&self) -> bool {
self.finished
}
pub fn stored_response_headers(&self) -> Vec<ResponseHeader> {
let Some(headers) = &self.response_headers else {
return Vec::new();
};
let mut out = Vec::new();
let mut budget = MAX_STORED_RESPONSE_HEADERS_BYTES;
for (name, value) in headers.iter() {
let name = name.as_str();
if is_secret_response_header(name) {
continue;
}
let Ok(value) = value.to_str() else {
continue;
};
if value.len() > MAX_STORED_RESPONSE_HEADER_VALUE_BYTES {
continue;
}
let cost = name.len() + value.len();
if cost > budget {
break;
}
budget -= cost;
out.push(ResponseHeader {
name: name.to_string(),
value: value.to_string(),
});
}
out
}
pub fn from_metadata(
download_dir: path::PathBuf,
metadata: DownloadMetadata,
) -> Result<Download, MetadataError> {
let url = Url::parse(&metadata.url).map_err(|e| MetadataError::Other {
message: e.to_string(),
})?;
Ok(Self {
download_dir,
url,
is_resumable: metadata.is_resumable,
use_server_time: metadata.use_server_time,
filename: metadata.filename, save_dir: PathBuf::from(metadata.save_dir),
etag: metadata.last_etag,
last_modified: metadata.last_modified,
size: metadata.size,
checksums: metadata
.checksums
.into_iter()
.map(|c| c.try_into())
.collect::<Result<Vec<HashDigest>, _>>()
.unwrap_or_default(),
credentials: None,
requires_auth: metadata.requires_auth,
requires_basic_auth: metadata.requires_basic_auth,
proxy: None,
headers: if metadata.headers.is_empty() {
None
} else {
let mut map = HeaderMap::new();
for (k, v) in metadata.headers {
if let (Ok(header_name), Ok(header_value)) =
(HeaderName::from_str(&k), HeaderValue::from_str(&v))
{
map.insert(header_name, header_value);
}
}
Some(map)
},
response_headers: if metadata.response_headers.is_empty() {
None
} else {
let mut map = HeaderMap::new();
for h in metadata.response_headers {
if let (Ok(name), Ok(value)) = (
HeaderName::from_str(&h.name),
HeaderValue::from_str(&h.value),
) {
map.append(name, value);
}
}
Some(map)
},
response_headers_probed_at: metadata.response_headers_probed_at,
max_connections: metadata.max_connections,
parts: metadata.parts,
finished: metadata.finished,
})
}
pub fn as_metadata(&self) -> DownloadMetadata {
DownloadMetadata {
url: self.url.to_string(),
filename: self.filename.clone(),
save_dir: self.save_dir.to_string_lossy().into_owned(),
is_resumable: self.is_resumable,
use_server_time: self.use_server_time,
last_modified: self.last_modified,
last_etag: self.etag.clone(),
size: self.size,
checksums: self
.checksums
.iter()
.map(|h| h.clone().into())
.collect::<Vec<FileChecksum>>(),
requires_auth: self.requires_auth,
requires_basic_auth: self.requires_basic_auth,
headers: self
.headers
.as_ref()
.map(|h| {
h.iter()
.map(|(k, v)| (k.to_string(), v.to_str().unwrap_or("").to_string()))
.collect()
})
.unwrap_or_default(),
response_headers: self.stored_response_headers(),
response_headers_probed_at: self.response_headers_probed_at,
max_connections: self.max_connections,
parts: self.parts.clone(),
finished: self.finished,
}
}
#[allow(clippy::too_many_arguments)]
pub fn from_response_info(
download_dir: &std::path::Path,
save_dir: path::PathBuf,
response_info: ResponseInfo,
max_connections: u64,
use_server_time: bool,
credentials: Option<Credentials>,
proxy: Option<Proxy>,
headers: Option<HeaderMap>,
) -> Download {
let filename = fs_utils::cleanup_filename(response_info.extract_filename().as_str());
let response_headers = {
let h = response_info.response_headers();
(!h.is_empty()).then(|| h.clone())
};
Self {
download_dir: download_dir.join(&filename),
url: response_info.url().clone(),
is_resumable: response_info.is_resumable(),
use_server_time,
filename,
save_dir,
etag: response_info.etag(),
last_modified: response_info.parse_last_modified(),
size: response_info.total_length(),
checksums: response_info.extract_hashes(),
credentials,
requires_auth: response_info.requires_auth(),
requires_basic_auth: response_info.requires_basic_auth(),
proxy,
headers,
response_headers_probed_at: response_headers.is_some().then(|| Utc::now().timestamp()),
response_headers,
max_connections,
parts: Download::determine_parts(
response_info.total_length(),
if response_info.is_resumable() {
max_connections
} else {
1
},
),
finished: false,
}
}
pub fn compute_split(
offset: u64,
size: u64,
already_consumed: u64,
min_part_size: u64,
) -> Option<PartSplit> {
if !offset.is_multiple_of(Self::ASSEMBLY_CLUSTER_SIZE) {
debug_assert!(
false,
"compute_split: offset {offset:#x} not cluster-aligned",
);
return None;
}
if already_consumed >= size {
return None;
}
let remaining = size - already_consumed;
if remaining < min_part_size * 2 {
return None;
}
let candidate = already_consumed + remaining / 2;
let new_left_size = candidate - candidate % Self::ASSEMBLY_CLUSTER_SIZE;
if new_left_size <= already_consumed {
return None;
}
let new_right_size = size - new_left_size;
if new_right_size < min_part_size || new_left_size - already_consumed < min_part_size {
return None;
}
Some(PartSplit {
new_left_size,
new_right_offset: offset + new_left_size,
new_right_size,
})
}
pub fn determine_parts(
size: Option<u64>,
max_connections: u64,
) -> HashMap<String, PartDetails> {
let mut parts = HashMap::new();
let max_connections = if max_connections > 0 {
max_connections
} else {
1
};
if size.is_none() {
let ulid = Ulid::new().to_string();
parts.insert(
ulid.clone(),
PartDetails {
offset: 0,
size: Self::UNKNOWN_PART_SIZE,
ulid,
finished: false,
},
);
return parts;
}
let size = size.unwrap_or(0);
if size <= Self::MIN_PART_SIZE {
let ulid = Ulid::new().to_string();
parts.insert(
ulid.clone(),
PartDetails {
offset: 0,
size,
ulid,
finished: size == 0,
},
);
return parts;
}
let mut actual_connections = max_connections;
let min_connections = size.div_ceil(Self::MIN_PART_SIZE);
if actual_connections > min_connections {
actual_connections = min_connections;
}
let raw_base = size / actual_connections;
let base_size = raw_base - raw_base % Self::ASSEMBLY_CLUSTER_SIZE;
let mut offset = 0;
for i in 0..actual_connections {
let part_size = if i == actual_connections - 1 {
size - offset
} else {
base_size
};
let ulid = Ulid::new().to_string();
parts.insert(
ulid.clone(),
PartDetails {
offset,
size: part_size,
ulid,
finished: false,
},
);
offset += part_size;
}
parts
}
}
impl PartialEq for Download {
fn eq(&self, other: &Self) -> bool {
self.url == other.url
&& self.download_dir == other.download_dir
&& self.filename == other.filename
}
}
impl DownloadBuilder {
fn validate(&self) -> Result<(), DownloadBuilderError> {
if self.download_dir.is_none() {
return Err(DownloadBuilderError::MissingDownloadDir);
}
if self.save_dir.is_none() {
return Err(DownloadBuilderError::MissingSaveDir);
}
if self.url.is_none() {
return Err(DownloadBuilderError::MissingUrl);
}
if self.filename.is_none() {
return Err(DownloadBuilderError::MissingFilename);
}
if self
.max_connections
.is_none_or(|x| x == 0 || x >= Semaphore::MAX_PERMITS.try_into().unwrap_or(1_000_000))
{
return Err(DownloadBuilderError::InvalidNumConnections);
}
Ok(())
}
}
#[derive(Error, Debug)]
pub enum DownloadBuilderError {
#[error("download_dir is required")]
MissingDownloadDir,
#[error("save_dir is required")]
MissingSaveDir,
#[error("url is required")]
MissingUrl,
#[error("filename is required")]
MissingFilename,
#[error("max_connections must be at least 1")]
InvalidNumConnections,
#[error("uninitialized field: {0}")]
UninitializedField(String),
#[error("validation error: {0}")]
ValidationError(String),
}
impl From<String> for DownloadBuilderError {
fn from(s: String) -> Self {
Self::ValidationError(s)
}
}
impl From<UninitializedFieldError> for DownloadBuilderError {
fn from(ufe: UninitializedFieldError) -> Self {
Self::UninitializedField(ufe.to_string())
}
}
#[cfg(test)]
mod tests {
use super::*;
fn test_download(checksums: Vec<HashDigest>) -> Download {
DownloadBuilder::default()
.download_dir(PathBuf::from("/tmp/dl"))
.save_dir(PathBuf::from("/tmp/save"))
.url(Url::parse("https://example.com/file").unwrap())
.filename("file".to_string())
.max_connections(1)
.checksums(checksums)
.parts(Download::determine_parts(Some(0), 1))
.build()
.unwrap()
}
fn download_with_response_headers(headers: Vec<(&str, &str)>) -> Download {
let mut map = HeaderMap::new();
for (name, value) in headers {
map.append(
HeaderName::from_str(name).unwrap(),
HeaderValue::from_str(value).unwrap(),
);
}
let mut dl = test_download(vec![]);
dl.response_headers = Some(map);
dl.response_headers_probed_at = Some(1_700_000_000);
dl
}
#[test]
fn stored_response_headers_drops_credential_bearing_ones() {
let dl = download_with_response_headers(vec![
("content-type", "application/zip"),
("set-cookie", "session=secret"),
("www-authenticate", "Basic realm=\"x\""),
("x-amz-security-token", "AQoDYXdz"),
("x-api-key", "k-123"),
("x-goog-signature", "deadbeef"),
("x-cache", "HIT"),
]);
let stored: Vec<String> = dl
.stored_response_headers()
.into_iter()
.map(|h| h.name)
.collect();
assert_eq!(stored, ["content-type", "x-cache"]);
}
#[test]
fn stored_response_headers_respects_caps() {
let huge = "v".repeat(MAX_STORED_RESPONSE_HEADER_VALUE_BYTES + 1);
let dl = download_with_response_headers(vec![
("x-huge", huge.as_str()),
("content-type", "application/zip"),
]);
let stored = dl.stored_response_headers();
assert_eq!(stored.len(), 1, "oversized value must be skipped");
assert_eq!(stored[0].name, "content-type");
let value = "v".repeat(512);
let names: Vec<String> = (0..40).map(|i| format!("x-pad-{i}")).collect();
let dl = download_with_response_headers(
names.iter().map(|n| (n.as_str(), value.as_str())).collect(),
);
let stored = dl.stored_response_headers();
let total: usize = stored.iter().map(|h| h.name.len() + h.value.len()).sum();
assert!(total <= MAX_STORED_RESPONSE_HEADERS_BYTES, "total {total}");
assert!(!stored.is_empty(), "budget must fit at least some headers");
assert!(stored.len() < names.len(), "budget must actually bind");
}
#[test]
fn response_headers_round_trip_through_metadata() {
let dl = download_with_response_headers(vec![
("content-type", "application/zip"),
("x-trace", "first"),
("x-trace", "second"),
("set-cookie", "session=secret"),
]);
let metadata = dl.as_metadata();
assert!(
!metadata
.response_headers
.iter()
.any(|h| h.name == "set-cookie"),
"filtering must apply on the way to disk"
);
assert_eq!(metadata.response_headers_probed_at, Some(1_700_000_000));
let restored = Download::from_metadata(PathBuf::from("/tmp/dl"), metadata).unwrap();
let headers = restored.response_headers().expect("headers survive resume");
assert_eq!(headers.get("content-type").unwrap(), "application/zip");
let traces: Vec<&str> = headers
.get_all("x-trace")
.iter()
.map(|v| v.to_str().unwrap())
.collect();
assert_eq!(traces, vec!["first", "second"], "duplicates preserved");
assert!(headers.get("set-cookie").is_none());
assert_eq!(restored.response_headers_probed_at(), Some(1_700_000_000));
}
#[test]
fn add_checksums_merges_and_dedups() {
use crate::hash::HashEncoding;
let server = HashDigest::SHA256("aa".repeat(32), HashEncoding::Hex);
let mut dl = test_download(vec![server.clone()]);
let user_new = HashDigest::MD5("bb".repeat(16), HashEncoding::Hex);
dl.add_checksums(vec![server.clone(), user_new.clone()]);
assert_eq!(dl.checksums, vec![server, user_new]);
}
#[test]
fn add_checksums_into_empty() {
use crate::hash::HashEncoding;
let mut dl = test_download(vec![]);
let c = HashDigest::SHA512("cc".repeat(64), HashEncoding::Hex);
dl.add_checksums(vec![c.clone()]);
assert_eq!(dl.checksums, vec![c]);
}
#[test]
fn test_determine_parts_unknown_size_streams_until_eof() {
let parts = Download::determine_parts(None, 4);
assert_eq!(parts.len(), 1);
let part = parts.values().next().unwrap();
assert_eq!(part.offset, 0);
assert_eq!(part.size, Download::UNKNOWN_PART_SIZE);
assert!(!part.finished);
}
#[test]
fn test_determine_parts_zero_size() {
let parts = Download::determine_parts(Some(0), 4);
assert_eq!(parts.len(), 1);
let part_vec: Vec<_> = parts.values().collect();
let part = part_vec[0];
assert_eq!(part.offset, 0);
assert_eq!(part.size, 0);
assert!(part.finished);
}
#[test]
fn test_determine_parts_zero_connections() {
let parts = Download::determine_parts(Some(1024 * 1024), 0);
assert_eq!(parts.len(), 1);
let part_vec: Vec<_> = parts.values().collect();
let part = part_vec[0];
assert_eq!(part.offset, 0);
assert_eq!(part.size, 1024 * 1024);
assert!(!part.finished);
}
#[test]
fn test_determine_parts_small_file() {
let size = 200 * 1024;
let parts = Download::determine_parts(Some(size), 4);
assert_eq!(parts.len(), 1);
let part_vec: Vec<_> = parts.values().collect();
let part = part_vec[0];
assert_eq!(part.offset, 0);
assert_eq!(part.size, size);
assert!(!part.finished);
}
#[test]
fn test_determine_parts_exact_min_part_size() {
let size = 300 * 1024;
let parts = Download::determine_parts(Some(size), 4);
assert_eq!(parts.len(), 1);
let part_vec: Vec<_> = parts.values().collect();
let part = part_vec[0];
assert_eq!(part.offset, 0);
assert_eq!(part.size, size);
assert!(!part.finished);
}
#[test]
fn test_determine_parts_even_split() {
let size = 1024 * 1024;
let max_connections = 4;
let parts = Download::determine_parts(Some(size), max_connections);
assert_eq!(parts.len(), max_connections as usize);
let mut part_vec: Vec<_> = parts.values().collect();
part_vec.sort_by_key(|p| p.offset);
let total: u64 = part_vec.iter().map(|p| p.size).sum();
assert_eq!(total, size);
assert_eq!(part_vec[0].offset, 0);
assert_eq!(part_vec[1].offset, part_vec[0].size);
assert_eq!(part_vec[2].offset, part_vec[0].size + part_vec[1].size);
assert_eq!(
part_vec[3].offset,
part_vec[0].size + part_vec[1].size + part_vec[2].size
);
}
#[test]
fn test_determine_parts_uneven_split() {
let size = 1024 * 1024 + 123;
let max_connections = 3;
let parts = Download::determine_parts(Some(size), max_connections);
assert_eq!(parts.len(), max_connections as usize);
let mut part_vec: Vec<_> = parts.values().collect();
part_vec.sort_by_key(|p| p.offset);
let total: u64 = part_vec.iter().map(|p| p.size).sum();
assert_eq!(total, size);
assert!(part_vec[2].size >= part_vec[1].size);
assert_eq!(part_vec[0].size, part_vec[1].size);
}
#[test]
fn test_determine_parts_too_many_connections() {
let size = 900 * 1024; let max_connections = 10;
let parts = Download::determine_parts(Some(size), max_connections);
assert_eq!(parts.len(), 3);
let mut part_vec: Vec<_> = parts.values().collect();
part_vec.sort_by_key(|p| p.offset);
let total: u64 = part_vec.iter().map(|p| p.size).sum();
assert_eq!(total, size);
}
#[test]
fn test_determine_parts_800kb_file() {
let size = 800 * 1024;
let max_connections = 10; let parts = Download::determine_parts(Some(size), max_connections);
assert_eq!(parts.len(), 3);
let mut part_vec: Vec<_> = parts.values().collect();
part_vec.sort_by_key(|p| p.offset);
let total: u64 = part_vec.iter().map(|p| p.size).sum();
assert_eq!(total, size);
assert_eq!(part_vec[0].offset, 0);
assert_eq!(part_vec[1].offset, part_vec[0].offset + part_vec[0].size);
assert_eq!(part_vec[2].offset, part_vec[1].offset + part_vec[1].size);
assert_eq!(part_vec[0].size, part_vec[1].size);
assert!(part_vec[2].size >= part_vec[1].size);
assert_eq!(part_vec[0].offset % Download::ASSEMBLY_CLUSTER_SIZE, 0);
assert_eq!(part_vec[1].offset % Download::ASSEMBLY_CLUSTER_SIZE, 0);
assert_eq!(part_vec[2].offset % Download::ASSEMBLY_CLUSTER_SIZE, 0);
}
#[test]
fn compute_split_returns_none_when_remaining_below_double_min() {
let size = Download::MIN_PART_SIZE * 2 - 1;
assert!(Download::compute_split(0, size, 0, Download::MIN_PART_SIZE).is_none());
}
#[test]
fn compute_split_aligns_boundary_and_preserves_total() {
let size = Download::MIN_PART_SIZE * 8;
let split = Download::compute_split(1024 * 1024, size, 0, Download::MIN_PART_SIZE)
.expect("split expected");
assert_eq!(split.new_left_size % Download::ASSEMBLY_CLUSTER_SIZE, 0);
assert_eq!(split.new_left_size + split.new_right_size, size);
assert_eq!(split.new_right_offset, 1024 * 1024 + split.new_left_size);
assert!(split.new_left_size >= Download::MIN_PART_SIZE);
assert!(split.new_right_size >= Download::MIN_PART_SIZE);
}
#[test]
fn compute_split_keeps_offsets_cluster_aligned_for_reflink() {
let size = 50 * 1024 * 1024 + 1234; let mut parts = Download::determine_parts(Some(size), 4);
for _ in 0..8 {
let candidate = parts
.values()
.filter_map(|p| {
Download::compute_split(p.offset, p.size, 0, Download::MIN_PART_SIZE)
.map(|s| (p.ulid.clone(), p.offset, p.size, s))
})
.max_by_key(|(_, _, _, s)| s.new_right_size);
let Some((ulid, _, _, split)) = candidate else {
break;
};
if let Some(p) = parts.get_mut(&ulid) {
p.size = split.new_left_size;
}
let new_ulid = ulid::Ulid::new().to_string();
parts.insert(
new_ulid.clone(),
crate::download_metadata::PartDetails {
offset: split.new_right_offset,
size: split.new_right_size,
ulid: new_ulid,
finished: false,
},
);
}
for p in parts.values() {
assert_eq!(
p.offset % Download::ASSEMBLY_CLUSTER_SIZE,
0,
"offset {} broke cluster alignment after split",
p.offset
);
}
let total: u64 = parts.values().map(|p| p.size).sum();
assert_eq!(total, size);
let mut sorted: Vec<_> = parts.values().collect();
sorted.sort_by_key(|p| p.offset);
for p in &sorted[..sorted.len() - 1] {
assert_eq!(
p.size % Download::ASSEMBLY_CLUSTER_SIZE,
0,
"non-last part size {} broke cluster alignment",
p.size
);
}
}
#[test]
fn compute_split_respects_already_consumed_floor() {
let size = Download::MIN_PART_SIZE * 8;
let consumed = Download::MIN_PART_SIZE * 4;
let split = Download::compute_split(0, size, consumed, Download::MIN_PART_SIZE)
.expect("split expected");
assert!(
split.new_left_size > consumed,
"boundary must move past already-consumed prefix"
);
assert_eq!(split.new_left_size + split.new_right_size, size);
}
}