| 1 | //! The registry side of a push: the token, blobs in chunks, the manifest. |
| 2 | //! |
| 3 | //! Every blob is checked with `HEAD` first and skipped when the registry |
| 4 | //! has it. Otherwise it is uploaded in chunks, each its own `PATCH` under |
| 5 | //! the 100 MB a request may carry, then closed with `PUT ?digest=`. A |
| 6 | //! `429` or `5xx` is retried after a wait (the `Retry-After` when there is |
| 7 | //! one), and an interrupted upload continues from the offset the registry |
| 8 | //! says it holds; one the registry let go is started again. |
| 9 | |
| 10 | use crate::chunk; |
| 11 | use crate::credentials::Credentials; |
| 12 | use crate::image::{Blob, Source}; |
| 13 | use base64::Engine as _; |
| 14 | use serde_json::Value; |
| 15 | use std::fs::File; |
| 16 | use std::io::{Read, Seek, SeekFrom}; |
| 17 | use std::path::Path; |
| 18 | use std::time::Duration; |
| 19 | |
| 20 | /// How many times a request is tried before the push gives up. |
| 21 | const ATTEMPTS: u32 = 6; |
| 22 | /// How many times one blob's upload may start over. |
| 23 | const RESTARTS: u32 = 3; |
| 24 | |
| 25 | pub struct Registry { |
| 26 | agent: ureq::Agent, |
| 27 | base: String, |
| 28 | host: String, |
| 29 | name: String, |
| 30 | credentials: Credentials, |
| 31 | token: Option<String>, |
| 32 | } |
| 33 | |
| 34 | /// A registry's answer, whatever its status. |
| 35 | pub struct Reply { |
| 36 | pub status: u16, |
| 37 | location: Option<String>, |
| 38 | range: Option<String>, |
| 39 | digest: Option<String>, |
| 40 | retry_after: Option<u64>, |
| 41 | challenge: Option<String>, |
| 42 | body: String, |
| 43 | } |
| 44 | |
| 45 | pub enum Outcome { |
| 46 | Exists, |
| 47 | Uploaded { chunks: u64 }, |
| 48 | } |
| 49 | |
| 50 | enum Failed { |
| 51 | /// The registry no longer has the upload: start the blob again. |
| 52 | Restart(String), |
| 53 | Fatal(String), |
| 54 | } |
| 55 | |
| 56 | type Result<T> = std::result::Result<T, String>; |
| 57 | |
| 58 | impl Registry { |
| 59 | pub fn new(base: String, host: String, name: String, credentials: Credentials) -> Registry { |
| 60 | let agent = ureq::AgentBuilder::new() |
| 61 | .timeout_connect(Duration::from_secs(30)) |
| 62 | .timeout_read(Duration::from_secs(300)) |
| 63 | .user_agent(concat!("g1t/", env!("CARGO_PKG_VERSION"))) |
| 64 | .try_proxy_from_env(true) |
| 65 | .build(); |
| 66 | Registry { |
| 67 | agent, |
| 68 | base, |
| 69 | host, |
| 70 | name, |
| 71 | credentials, |
| 72 | token: None, |
| 73 | } |
| 74 | } |
| 75 | |
| 76 | /// Trades the g1t token for a registry token that may pull and push |
| 77 | /// this image, at the token endpoint the registry names. |
| 78 | pub fn sign_in(&mut self) -> Result<()> { |
| 79 | let ping = self.raw("GET", &format!("{}/v2/", self.base), &[], None, false)?; |
| 80 | let (realm, service) = ping |
| 81 | .challenge |
| 82 | .as_deref() |
| 83 | .and_then(parse_challenge) |
| 84 | .unwrap_or_else(|| (format!("{}/v2/token", self.base), self.host.clone())); |
| 85 | let url = format!( |
| 86 | "{realm}?service={service}&scope=repository:{}:pull,push", |
| 87 | self.name |
| 88 | ); |
| 89 | let basic = base64::engine::general_purpose::STANDARD.encode(format!( |
| 90 | "{}:{}", |
| 91 | self.credentials.username, self.credentials.secret |
| 92 | )); |
| 93 | let authorization = format!("Basic {basic}"); |
| 94 | let mut reply = None; |
| 95 | for attempt in 0..ATTEMPTS { |
| 96 | match self.raw( |
| 97 | "GET", |
| 98 | &url, |
| 99 | &[("authorization", &authorization)], |
| 100 | None, |
| 101 | false, |
| 102 | ) { |
| 103 | Ok(r) if retry_wait(&r, attempt).is_none() => { |
| 104 | reply = Some(r); |
| 105 | break; |
| 106 | } |
| 107 | Ok(r) => wait(retry_wait(&r, attempt)), |
| 108 | Err(error) if attempt + 1 == ATTEMPTS => return Err(error), |
| 109 | Err(_) => wait(Some(backoff(attempt))), |
| 110 | } |
| 111 | } |
| 112 | let reply = reply.ok_or_else(|| format!("{} did not answer the sign-in.", self.host))?; |
| 113 | if reply.status != 200 { |
| 114 | return Err(format!( |
| 115 | "{} refused the token: {}", |
| 116 | self.host, |
| 117 | explain(&reply) |
| 118 | )); |
| 119 | } |
| 120 | let body: Value = serde_json::from_str(&reply.body).map_err(|_| { |
| 121 | format!( |
| 122 | "{} answered the sign-in with something that is not JSON.", |
| 123 | self.host |
| 124 | ) |
| 125 | })?; |
| 126 | let token = body |
| 127 | .get("token") |
| 128 | .or_else(|| body.get("access_token")) |
| 129 | .and_then(Value::as_str); |
| 130 | self.token = Some( |
| 131 | token |
| 132 | .ok_or_else(|| format!("{} answered the sign-in without a token.", self.host))? |
| 133 | .to_owned(), |
| 134 | ); |
| 135 | Ok(()) |
| 136 | } |
| 137 | |
| 138 | /// One request, as it went. Transport failures are errors. |
| 139 | fn raw( |
| 140 | &self, |
| 141 | method: &str, |
| 142 | url: &str, |
| 143 | headers: &[(&str, &str)], |
| 144 | body: Option<&[u8]>, |
| 145 | bearer: bool, |
| 146 | ) -> Result<Reply> { |
| 147 | let mut request = self.agent.request(method, url); |
| 148 | for (name, value) in headers { |
| 149 | request = request.set(name, value); |
| 150 | } |
| 151 | if bearer && let Some(token) = &self.token { |
| 152 | request = request.set("authorization", &format!("Bearer {token}")); |
| 153 | } |
| 154 | let result = match body { |
| 155 | Some(bytes) => request.send_bytes(bytes), |
| 156 | None => request.call(), |
| 157 | }; |
| 158 | let response = match result { |
| 159 | Ok(response) | Err(ureq::Error::Status(_, response)) => response, |
| 160 | Err(ureq::Error::Transport(error)) => return Err(format!("{method} {url}: {error}")), |
| 161 | }; |
| 162 | let header = |name: &str| response.header(name).map(str::to_owned); |
| 163 | let mut reply = Reply { |
| 164 | status: response.status(), |
| 165 | location: header("location"), |
| 166 | range: header("range"), |
| 167 | digest: header("docker-content-digest"), |
| 168 | retry_after: header("retry-after").and_then(|s| s.trim().parse().ok()), |
| 169 | challenge: header("www-authenticate"), |
| 170 | body: String::new(), |
| 171 | }; |
| 172 | if method != "HEAD" { |
| 173 | let mut body = String::new(); |
| 174 | let _ = response |
| 175 | .into_reader() |
| 176 | .take(1 << 20) |
| 177 | .read_to_string(&mut body); |
| 178 | reply.body = body; |
| 179 | } |
| 180 | Ok(reply) |
| 181 | } |
| 182 | |
| 183 | /// One request with the registry token, signing in again once if the |
| 184 | /// token has run out. |
| 185 | fn send( |
| 186 | &mut self, |
| 187 | method: &str, |
| 188 | url: &str, |
| 189 | headers: &[(&str, &str)], |
| 190 | body: Option<&[u8]>, |
| 191 | ) -> Result<Reply> { |
| 192 | let reply = self.raw(method, url, headers, body, true)?; |
| 193 | if reply.status == 401 { |
| 194 | self.sign_in()?; |
| 195 | return self.raw(method, url, headers, body, true); |
| 196 | } |
| 197 | Ok(reply) |
| 198 | } |
| 199 | |
| 200 | /// A request that may be sent again as it is, tried until it gets an |
| 201 | /// answer that is not a `429` or `5xx`. |
| 202 | fn call( |
| 203 | &mut self, |
| 204 | method: &str, |
| 205 | url: &str, |
| 206 | headers: &[(&str, &str)], |
| 207 | body: Option<&[u8]>, |
| 208 | ) -> Result<Reply> { |
| 209 | let mut last = String::new(); |
| 210 | for attempt in 0..ATTEMPTS { |
| 211 | match self.send(method, url, headers, body) { |
| 212 | Ok(reply) => match retry_wait(&reply, attempt) { |
| 213 | None => return Ok(reply), |
| 214 | Some(delay) => { |
| 215 | last = explain(&reply); |
| 216 | wait(Some(delay)); |
| 217 | } |
| 218 | }, |
| 219 | Err(error) => { |
| 220 | last = error; |
| 221 | wait(Some(backoff(attempt))); |
| 222 | } |
| 223 | } |
| 224 | } |
| 225 | Err(format!( |
| 226 | "{method} {url} failed {ATTEMPTS} times; the last: {last}" |
| 227 | )) |
| 228 | } |
| 229 | |
| 230 | pub fn blob_exists(&mut self, digest: &str) -> Result<bool> { |
| 231 | let url = format!("{}/v2/{}/blobs/{digest}", self.base, self.name); |
| 232 | let reply = self.call("HEAD", &url, &[], None)?; |
| 233 | match reply.status { |
| 234 | 200 => Ok(true), |
| 235 | 404 => Ok(false), |
| 236 | _ => Err(format!("Could not check for {digest}: {}", explain(&reply))), |
| 237 | } |
| 238 | } |
| 239 | |
| 240 | /// Puts a blob in the registry unless it is there already. |
| 241 | pub fn push_blob(&mut self, blob: &Blob, tar: &Path, chunk_size: u64) -> Result<Outcome> { |
| 242 | if self.blob_exists(&blob.digest)? { |
| 243 | return Ok(Outcome::Exists); |
| 244 | } |
| 245 | let mut restarts = 0; |
| 246 | loop { |
| 247 | match self.upload(blob, tar, chunk_size) { |
| 248 | Ok(chunks) => return Ok(Outcome::Uploaded { chunks }), |
| 249 | Err(Failed::Restart(reason)) if restarts < RESTARTS => { |
| 250 | restarts += 1; |
| 251 | eprintln!(" {reason}; starting this blob again"); |
| 252 | } |
| 253 | Err(Failed::Restart(reason) | Failed::Fatal(reason)) => return Err(reason), |
| 254 | } |
| 255 | } |
| 256 | } |
| 257 | |
| 258 | fn resolve(&self, location: Option<&str>, fallback: &str) -> String { |
| 259 | match location { |
| 260 | Some(l) if l.starts_with("http://") || l.starts_with("https://") => l.to_owned(), |
| 261 | Some(l) if l.starts_with('/') => format!("{}{l}", self.base), |
| 262 | Some(l) if !l.is_empty() => format!("{}/{l}", self.base), |
| 263 | _ => fallback.to_owned(), |
| 264 | } |
| 265 | } |
| 266 | |
| 267 | /// Where an interrupted upload stands: the offset to continue from, |
| 268 | /// or a restart when the registry let it go. |
| 269 | fn upload_offset(&mut self, location: &mut String) -> std::result::Result<u64, Failed> { |
| 270 | let reply = self |
| 271 | .call("GET", location, &[], None) |
| 272 | .map_err(Failed::Fatal)?; |
| 273 | match reply.status { |
| 274 | 200 | 204 => { |
| 275 | *location = self.resolve(reply.location.as_deref(), location); |
| 276 | reply |
| 277 | .range |
| 278 | .as_deref() |
| 279 | .and_then(chunk::next_offset) |
| 280 | .ok_or_else(|| Failed::Restart("the upload's progress is unknown".to_owned())) |
| 281 | } |
| 282 | 404 => Err(Failed::Restart("the registry let the upload go".to_owned())), |
| 283 | _ => Err(Failed::Fatal(format!( |
| 284 | "Could not check the upload: {}", |
| 285 | explain(&reply) |
| 286 | ))), |
| 287 | } |
| 288 | } |
| 289 | |
| 290 | fn upload( |
| 291 | &mut self, |
| 292 | blob: &Blob, |
| 293 | tar: &Path, |
| 294 | chunk_size: u64, |
| 295 | ) -> std::result::Result<u64, Failed> { |
| 296 | let start = format!("{}/v2/{}/blobs/uploads/", self.base, self.name); |
| 297 | let reply = self |
| 298 | .call("POST", &start, &[], Some(&[])) |
| 299 | .map_err(Failed::Fatal)?; |
| 300 | if reply.status != 202 { |
| 301 | return Err(Failed::Fatal(format!( |
| 302 | "Could not start an upload: {}", |
| 303 | explain(&reply) |
| 304 | ))); |
| 305 | } |
| 306 | let mut location = self.resolve(reply.location.as_deref(), ""); |
| 307 | if location.is_empty() { |
| 308 | return Err(Failed::Fatal( |
| 309 | "The registry started an upload without saying where.".to_owned(), |
| 310 | )); |
| 311 | } |
| 312 | let (mut source, base_offset) = open(blob, tar).map_err(Failed::Fatal)?; |
| 313 | let total_chunks = chunk::count(blob.size, chunk_size); |
| 314 | let mut offset = 0u64; |
| 315 | let mut chunks = 0u64; |
| 316 | let mut failures = 0u32; |
| 317 | let mut buffer = Vec::new(); |
| 318 | while let Some(&(_, length)) = chunk::plan(blob.size, chunk_size, offset).first() { |
| 319 | read_at(&mut source, base_offset + offset, length, &mut buffer) |
| 320 | .map_err(Failed::Fatal)?; |
| 321 | let range = chunk::content_range(offset, length); |
| 322 | let headers = [ |
| 323 | ("content-type", "application/octet-stream"), |
| 324 | ("content-range", range.as_str()), |
| 325 | ]; |
| 326 | let sent = self.send("PATCH", &location.clone(), &headers, Some(&buffer)); |
| 327 | let retry = match &sent { |
| 328 | Ok(reply) if reply.status == 202 => None, |
| 329 | Ok(reply) => retry_wait(reply, failures), |
| 330 | Err(_) => Some(backoff(failures)), |
| 331 | }; |
| 332 | match sent { |
| 333 | Ok(reply) if reply.status == 202 => { |
| 334 | location = self.resolve(reply.location.as_deref(), &location); |
| 335 | let next = reply |
| 336 | .range |
| 337 | .as_deref() |
| 338 | .and_then(chunk::next_offset) |
| 339 | .unwrap_or(offset + length); |
| 340 | if next > blob.size { |
| 341 | return Err(Failed::Fatal(format!( |
| 342 | "The registry says it holds {next} bytes of a {}-byte blob.", |
| 343 | blob.size |
| 344 | ))); |
| 345 | } |
| 346 | offset = if next == 0 { offset + length } else { next }; |
| 347 | chunks += 1; |
| 348 | failures = 0; |
| 349 | if total_chunks > 1 { |
| 350 | eprintln!( |
| 351 | " chunk {}/{total_chunks} {} {}", |
| 352 | chunks.min(total_chunks), |
| 353 | chunk::human(length), |
| 354 | percent(offset, blob.size) |
| 355 | ); |
| 356 | } |
| 357 | } |
| 358 | Ok(reply) if reply.status == 413 => { |
| 359 | return Err(Failed::Fatal(format!( |
| 360 | "A {} chunk was refused as too large ({}). Try a smaller --chunk-size.", |
| 361 | chunk::human(length), |
| 362 | explain(&reply) |
| 363 | ))); |
| 364 | } |
| 365 | Ok(reply) if reply.status == 404 || reply.status == 416 => { |
| 366 | offset = self.upload_offset(&mut location)?; |
| 367 | } |
| 368 | Ok(reply) if retry.is_none() => { |
| 369 | return Err(Failed::Fatal(format!( |
| 370 | "A chunk was refused: {}", |
| 371 | explain(&reply) |
| 372 | ))); |
| 373 | } |
| 374 | sent => { |
| 375 | failures += 1; |
| 376 | let why = match sent { |
| 377 | Ok(reply) => explain(&reply), |
| 378 | Err(error) => error, |
| 379 | }; |
| 380 | if failures >= ATTEMPTS { |
| 381 | return Err(Failed::Fatal(format!( |
| 382 | "A chunk failed {ATTEMPTS} times; the last: {why}" |
| 383 | ))); |
| 384 | } |
| 385 | eprintln!( |
| 386 | " chunk at {} failed ({why}); trying again", |
| 387 | chunk::human(offset) |
| 388 | ); |
| 389 | wait(retry); |
| 390 | offset = self.upload_offset(&mut location)?; |
| 391 | } |
| 392 | } |
| 393 | } |
| 394 | // Every byte is in: close the upload with its digest. |
| 395 | let separator = if location.contains('?') { '&' } else { '?' }; |
| 396 | let finish = format!( |
| 397 | "{location}{separator}digest={}", |
| 398 | blob.digest.replace(':', "%3A") |
| 399 | ); |
| 400 | let mut failures = 0u32; |
| 401 | loop { |
| 402 | let sent = self.send( |
| 403 | "PUT", |
| 404 | &finish, |
| 405 | &[("content-type", "application/octet-stream")], |
| 406 | Some(&[]), |
| 407 | ); |
| 408 | let why = match sent { |
| 409 | Ok(reply) if reply.status == 201 => return Ok(chunks.max(1)), |
| 410 | Ok(reply) if reply.status != 404 && retry_wait(&reply, failures).is_none() => { |
| 411 | return Err(Failed::Fatal(format!( |
| 412 | "The registry did not take {}: {}", |
| 413 | blob.digest, |
| 414 | explain(&reply) |
| 415 | ))); |
| 416 | } |
| 417 | Ok(reply) => explain(&reply), |
| 418 | Err(error) => error, |
| 419 | }; |
| 420 | failures += 1; |
| 421 | if failures >= ATTEMPTS { |
| 422 | return Err(Failed::Fatal(format!( |
| 423 | "Closing the upload failed {ATTEMPTS} times; the last: {why}" |
| 424 | ))); |
| 425 | } |
| 426 | wait(Some(backoff(failures))); |
| 427 | // The close may have landed before the answer was lost. |
| 428 | if self.blob_exists(&blob.digest).map_err(Failed::Fatal)? { |
| 429 | return Ok(chunks.max(1)); |
| 430 | } |
| 431 | if self.upload_offset(&mut location.clone())? != blob.size { |
| 432 | return Err(Failed::Restart("the upload lost bytes".to_owned())); |
| 433 | } |
| 434 | } |
| 435 | } |
| 436 | |
| 437 | /// Puts the manifest under the tag. Returns the registry's digest. |
| 438 | pub fn push_manifest( |
| 439 | &mut self, |
| 440 | tag: &str, |
| 441 | media_type: &str, |
| 442 | manifest: &[u8], |
| 443 | ) -> Result<String> { |
| 444 | let url = format!("{}/v2/{}/manifests/{tag}", self.base, self.name); |
| 445 | let reply = self.call("PUT", &url, &[("content-type", media_type)], Some(manifest))?; |
| 446 | if reply.status != 201 && reply.status != 200 { |
| 447 | return Err(format!( |
| 448 | "The registry did not take the manifest: {}", |
| 449 | explain(&reply) |
| 450 | )); |
| 451 | } |
| 452 | Ok(reply.digest.unwrap_or_default()) |
| 453 | } |
| 454 | } |
| 455 | |
| 456 | fn open(blob: &Blob, tar: &Path) -> Result<(File, u64)> { |
| 457 | let (path, offset) = match &blob.source { |
| 458 | Source::Tar { offset } => (tar, *offset), |
| 459 | Source::File(path) => (path.as_path(), 0), |
| 460 | }; |
| 461 | let file = File::open(path).map_err(|e| format!("Could not open {}: {e}", path.display()))?; |
| 462 | Ok((file, offset)) |
| 463 | } |
| 464 | |
| 465 | fn read_at(file: &mut File, offset: u64, length: u64, buffer: &mut Vec<u8>) -> Result<()> { |
| 466 | buffer.resize(length as usize, 0); |
| 467 | file.seek(SeekFrom::Start(offset)) |
| 468 | .map_err(|e| e.to_string())?; |
| 469 | file.read_exact(buffer) |
| 470 | .map_err(|e| format!("Could not read the image: {e}")) |
| 471 | } |
| 472 | |
| 473 | fn percent(done: u64, total: u64) -> String { |
| 474 | format!( |
| 475 | "{:.0}%", |
| 476 | if total == 0 { |
| 477 | 100.0 |
| 478 | } else { |
| 479 | done as f64 * 100.0 / total as f64 |
| 480 | } |
| 481 | ) |
| 482 | } |
| 483 | |
| 484 | /// `Bearer realm="https://g1t.sh/v2/token",service="g1t.sh",scope="…"`. |
| 485 | fn parse_challenge(header: &str) -> Option<(String, String)> { |
| 486 | let rest = header |
| 487 | .trim() |
| 488 | .strip_prefix("Bearer ") |
| 489 | .or_else(|| header.trim().strip_prefix("bearer "))?; |
| 490 | let mut realm = None; |
| 491 | let mut service = None; |
| 492 | for part in rest.split(',') { |
| 493 | let (key, value) = part.trim().split_once('=')?; |
| 494 | let value = value.trim().trim_matches('"').to_owned(); |
| 495 | match key.trim() { |
| 496 | "realm" => realm = Some(value), |
| 497 | "service" => service = Some(value), |
| 498 | _ => {} |
| 499 | } |
| 500 | } |
| 501 | Some((realm?, service.unwrap_or_default())) |
| 502 | } |
| 503 | |
| 504 | /// How long to wait before trying again, when the answer is worth |
| 505 | /// another try: `429` and `5xx`. |
| 506 | fn retry_wait(reply: &Reply, attempt: u32) -> Option<Duration> { |
| 507 | if reply.status != 429 && reply.status < 500 { |
| 508 | return None; |
| 509 | } |
| 510 | Some(match reply.retry_after { |
| 511 | Some(seconds) => Duration::from_secs(seconds.min(300)), |
| 512 | None => backoff(attempt), |
| 513 | }) |
| 514 | } |
| 515 | |
| 516 | /// 1s, 2s, 4s … at most 30s. |
| 517 | pub fn backoff(attempt: u32) -> Duration { |
| 518 | Duration::from_secs((1u64 << attempt.min(5)).min(30)) |
| 519 | } |
| 520 | |
| 521 | fn wait(delay: Option<Duration>) { |
| 522 | if let Some(delay) = delay { |
| 523 | std::thread::sleep(delay); |
| 524 | } |
| 525 | } |
| 526 | |
| 527 | /// An error reply for people: the registry's own code and message when it |
| 528 | /// sent them, the status otherwise. |
| 529 | fn explain(reply: &Reply) -> String { |
| 530 | if let Ok(body) = serde_json::from_str::<Value>(&reply.body) |
| 531 | && let Some(error) = body |
| 532 | .get("errors") |
| 533 | .and_then(Value::as_array) |
| 534 | .and_then(|e| e.first()) |
| 535 | { |
| 536 | let code = error.get("code").and_then(Value::as_str).unwrap_or("ERROR"); |
| 537 | let message = error.get("message").and_then(Value::as_str).unwrap_or(""); |
| 538 | return format!("{code}: {message} ({})", reply.status); |
| 539 | } |
| 540 | let text = reply.body.trim(); |
| 541 | if text.is_empty() || text.starts_with('<') { |
| 542 | format!("HTTP {}", reply.status) |
| 543 | } else { |
| 544 | format!( |
| 545 | "HTTP {}: {}", |
| 546 | reply.status, |
| 547 | text.chars().take(200).collect::<String>() |
| 548 | ) |
| 549 | } |
| 550 | } |
| 551 | |
| 552 | #[cfg(test)] |
| 553 | mod tests { |
| 554 | use super::*; |
| 555 | |
| 556 | fn reply(status: u16, retry_after: Option<u64>, body: &str) -> Reply { |
| 557 | Reply { |
| 558 | status, |
| 559 | location: None, |
| 560 | range: None, |
| 561 | digest: None, |
| 562 | retry_after, |
| 563 | challenge: None, |
| 564 | body: body.to_owned(), |
| 565 | } |
| 566 | } |
| 567 | |
| 568 | #[test] |
| 569 | fn challenges_name_the_token_endpoint() { |
| 570 | let header = r#"Bearer realm="https://g1t.sh/v2/token",service="g1t.sh",scope="repository:acme/web:pull""#; |
| 571 | assert_eq!( |
| 572 | parse_challenge(header), |
| 573 | Some(("https://g1t.sh/v2/token".into(), "g1t.sh".into())) |
| 574 | ); |
| 575 | assert_eq!(parse_challenge("Basic realm=\"x\""), None); |
| 576 | } |
| 577 | |
| 578 | #[test] |
| 579 | fn only_429_and_5xx_are_tried_again() { |
| 580 | assert_eq!( |
| 581 | retry_wait(&reply(429, Some(7), ""), 0), |
| 582 | Some(Duration::from_secs(7)) |
| 583 | ); |
| 584 | assert_eq!( |
| 585 | retry_wait(&reply(503, None, ""), 2), |
| 586 | Some(Duration::from_secs(4)) |
| 587 | ); |
| 588 | assert_eq!(retry_wait(&reply(400, None, ""), 0), None); |
| 589 | assert_eq!(retry_wait(&reply(413, None, ""), 0), None); |
| 590 | } |
| 591 | |
| 592 | #[test] |
| 593 | fn backoff_doubles_up_to_30s() { |
| 594 | let waits: Vec<u64> = (0..8).map(|n| backoff(n).as_secs()).collect(); |
| 595 | assert_eq!(waits, [1, 2, 4, 8, 16, 30, 30, 30]); |
| 596 | } |
| 597 | |
| 598 | #[test] |
| 599 | fn errors_read_as_the_registry_wrote_them() { |
| 600 | let body = r#"{"errors":[{"code":"DENIED","message":"Pushing needs Write."}]}"#; |
| 601 | assert_eq!( |
| 602 | explain(&reply(403, None, body)), |
| 603 | "DENIED: Pushing needs Write. (403)" |
| 604 | ); |
| 605 | assert_eq!( |
| 606 | explain(&reply(413, None, "<html>Too large</html>")), |
| 607 | "HTTP 413" |
| 608 | ); |
| 609 | } |
| 610 | } |