1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
use crate::{cargo::Source, util, Krate};
use anyhow::Context as _;
use bytes::Bytes;
use tracing::warn;
pub(crate) enum KratePackage {
Registry(Bytes),
Git(crate::git::GitPackage),
}
impl KratePackage {
pub(crate) fn len(&self) -> usize {
match self {
Self::Registry(bytes) => bytes.len(),
Self::Git(gs) => gs.db.len() + gs.checkout.as_ref().map_or(0, |s| s.len()),
}
}
}
#[tracing::instrument(level = "debug")]
pub(crate) async fn from_registry(
client: &crate::HttpClient,
krate: &Krate,
) -> anyhow::Result<KratePackage> {
match &krate.source {
Source::Git(gs) => {
let gs = gs.clone();
tokio::task::spawn_blocking(move || crate::git::clone(&gs).map(KratePackage::Git))
.await
.unwrap()
}
Source::Registry(rs) => {
let url = rs.registry.download_url(krate);
// Depending on how many crates we are mirroring, we can be sending
// hundreds of concurrent requests to crates.io...and hit
// https://github.com/seanmonstar/reqwest/issues/1748
let res = loop {
let res = client.get(&url).send().await;
match res {
Err(err) if err.is_connect() || err.is_timeout() || err.is_request() => {
continue
}
Err(err) => return Err(err.into()),
Ok(res) => break res,
}
};
let response = res.error_for_status()?;
let res = util::convert_response(response).await?;
let content = res.into_body();
util::validate_checksum(&content, &rs.chksum)?;
Ok(KratePackage::Registry(content))
}
}
}
#[tracing::instrument(level = "debug", skip(krates))]
pub async fn registry(
client: &crate::HttpClient,
registry: &crate::cargo::Registry,
krates: Vec<String>,
) -> anyhow::Result<Bytes> {
use tame_index::index;
// We don't bother to support older versions of cargo that don't support
// bare checkouts of registry indexes, as that has been in since early 2017
// See https://github.com/rust-lang/cargo/blob/0e38712d4d7b346747bf91fb26cce8df6934e178/src/cargo/sources/registry/remote.rs#L61
// for details on why cargo still does what it does
let temp_dir = tempfile::tempdir()?;
let temp_dir_path = util::path(temp_dir.path())?;
let index_url = registry.index.as_str().to_owned();
let write_cache = tracing::span!(tracing::Level::DEBUG, "write-cache-entries");
let location = index::IndexLocation {
// note this is a bit of a misnomer, it could be the crates.io registry
url: index::IndexUrl::NonCratesIo(index_url.clone().into()),
root: index::IndexPath::Exact(temp_dir_path.to_owned()),
};
// Writes .cache entries in the registry's directory for all of the specified
// crates.
//
// Cargo will write these entries itself if they don't exist the first time it
// tries to access the crate's metadata in the case of git, but this noticeably
// increases initial fetch times. (see src/cargo/sources/registry/index.rs)
//
// For sparse indices, the cache entries are the _only_ local state, and if
// not present means every missing crate needs to be fetched, without the
// possibility of the local cache entry being up to date according to the
// etag/modified time of the remote
match registry.protocol {
crate::cargo::RegistryProtocol::Git => {
tokio::task::spawn_blocking(move || -> anyhow::Result<()> {
let rgi = {
let span = tracing::debug_span!("fetch");
let _fs = span.enter();
tame_index::index::RemoteGitIndex::new(
tame_index::index::GitIndex::new(location)
.context("unable to open git index")?,
)
.context("failed to fetch")?
};
write_cache.in_scope(|| {
// As with git2, gix::Repository is not thread safe, we _could_
// read blobs in serial then write in parallel, but that's not really
// worth it for a few hundred crates (probably), but see
// https://github.com/frewsxcv/rust-crates-index/blob/a9b60653efb72d9e6be98c4f8fe56194475cbd3f/src/git/mod.rs#L316-L360
// for a way this could be done in the future
for name in krates {
let Ok(name) = name.as_str().try_into() else {
warn!("crate name '{name}' is invalid");
continue;
};
if let Err(err) = rgi.krate(name, true /* write the cache entry */) {
warn!("unable to write .cache entry: {err:#}");
}
}
});
Ok(())
})
.await
.unwrap()?;
}
crate::cargo::RegistryProtocol::Sparse => {
let index = index::AsyncRemoteSparseIndex::new(
index::SparseIndex::new(location)?,
client.clone(),
);
#[allow(unsafe_code)]
// SAFETY: we don't forget the future :p
unsafe {
async_scoped::TokioScope::scope_and_collect(|s| {
s.spawn(async {
// We don't particularly care if an individual crate fails here
// since the index will be healed by cargo, but still good to
// know if something was amiss
for (name, res) in
index.krates(krates.into_iter().collect(), true, None).await
{
match res {
Ok(Some(_)) => {}
Ok(None) => {
warn!("index entry for '{name}' was not found");
}
Err(err) => {
warn!("unable to write .cache entry for '{name}': {err:#}");
}
}
}
});
s.spawn(async {
let write_config = async {
let url =
format!("{}config.json", index_url.split_once('+').unwrap().1);
let res = loop {
let res = client.get(&url).send().await;
match res {
Err(err)
if err.is_connect()
|| err.is_timeout()
|| err.is_request() =>
{
continue
}
Err(err) => Err(err)
.context("failed to send request for config.json")?,
Ok(res) => break res,
}
};
let config_body = res
.bytes()
.await
.context("failed to read config.json response body")?;
std::fs::write(temp_dir.path().join("config.json"), &config_body)
.context("failed to write config.json")
};
if let Err(err) = write_config.await {
warn!("unable to write config.json: {err:#}");
}
});
})
.await;
}
}
};
util::pack_tar(temp_dir_path)
}