diff --git a/Cargo.lock b/Cargo.lock index d23e9396..e055c033 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -2368,6 +2368,7 @@ dependencies = [ "google-cloud-auth", "hex", "http 1.4.1", + "pin-project-lite", "regex-lite", "reqwest", "rustls", @@ -2387,16 +2388,19 @@ version = "0.5.3" dependencies = [ "bytes", "crc32fast", + "fastrand", "futures-util", "globset", "http 1.4.1", "ocync-distribution", "postcard", + "rcgen", "reqwest", "schemars 1.2.1", "serde", "serde_json", "tempfile", + "testcontainers", "thiserror", "tokio", "tracing", diff --git a/Cargo.toml b/Cargo.toml index 68c36332..b6ba3571 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -17,10 +17,12 @@ aws-sdk-ecrpublic = { version = "1", default-features = false, features = ["beha clap = { version = "4", default-features = false, features = ["derive", "std", "help", "usage", "error-context"] } google-cloud-auth = { version = "1.10", default-features = false } bytes = { version = "1", default-features = false } +fastrand = { version = "2", default-features = false, features = ["std"] } hex = { version = "0.4", default-features = false, features = ["alloc"] } http = { version = "1", default-features = false } schemars = { version = "1.2", default-features = false, features = ["derive", "std"] } futures-util = { version = "0.3", default-features = false } +pin-project-lite = { version = "0.2", default-features = false } serde = { version = "1", default-features = false, features = ["derive", "alloc"] } serde_json = { version = "1", default-features = false, features = ["alloc"] } serde_yaml = { package = "serde_yaml_ng", version = "0.10", default-features = false } @@ -102,3 +104,10 @@ strip = true lto = true codegen-units = 1 panic = "abort" + +# Release-equivalent codegen with symbols preserved for `samply` / +# flamegraph work. Use via `cargo {test,build} --profile profiling`. +[profile.profiling] +inherits = "release" +debug = "full" +strip = false diff --git a/crates/ocync-distribution/CLAUDE.md b/crates/ocync-distribution/CLAUDE.md index d223bd63..4feaeb6f 100644 --- a/crates/ocync-distribution/CLAUDE.md +++ b/crates/ocync-distribution/CLAUDE.md @@ -57,10 +57,10 @@ When `auth_type` IS set in config, it overrides detection. Valid values: `ecr`, ## Upload protocol quirks -- Default: POST + streaming PUT with `Transfer-Encoding: chunked` (2 requests/blob). -- GHCR: multi-PATCH chunked broken (last PATCH overwrites previous). Client falls back to POST + single PATCH + PUT (3 requests/blob). -- GAR: no chunked uploads. Client buffers full blob, monolithic PUT. -- ACR: known ~20 MB streaming PUT body limit. Chunked PATCH fallback not yet implemented. +- Default: POST + streaming PUT with `Transfer-Encoding: chunked` (2 requests/blob). Streaming PUT body is gated by a per-`RegistryClient` semaphore (`streaming_blob_sem`, default cap 64) to stay under the per-h2-connection `SETTINGS_MAX_CONCURRENT_STREAMS` budget (100-128 across major registries probed 2026-06-01). +- GHCR: multi-PATCH chunked broken (last PATCH overwrites previous). Client falls back to POST + single PATCH + PUT (3 requests/blob), `blob_push_stream_ghcr`. +- GAR: no chunked uploads. Client buffers full blob, monolithic PUT, `blob_push_stream_gar`. +- ACR: ~20 MB streaming PUT body limit. Client buffers the full blob, verifies digest, then uploads in 16 MB PATCH chunks (OCI `{start}-{end}` Content-Range, NOT RFC 7233) followed by a finalize PUT, `blob_push_stream_acr`. Zero-byte blobs (e.g. signature empty-config) skip the PATCH loop and go straight to finalize PUT. Each PATCH response's `Location` header is checked against the initiate host to prevent cross-host credential forwarding via a compromised proxy. ## Cross-repo mount @@ -96,6 +96,18 @@ Provider names: `build_registry_client` (`src/cli/mod.rs`) calls `ocync_distribution::install_crypto_provider()` at the top -- production main does this too, but the dispatch entry point is also reached from tests that bypass main, so the install must be idempotent there. +## Perf-harness test hooks on `RegistryClientBuilder` + +Three `#[doc(hidden)] pub` builder methods exist on `RegistryClientBuilder` solely for in-repo perf and A/B harnesses (primarily `crates/ocync-sync/tests/perf_profile.rs`): + +- `allow_invalid_certs(bool)` -- terminates TLS at a `testcontainers` `registry:2` with a self-signed cert. +- `force_http1(bool)` -- compares HTTP/2 vs HTTP/1.1 throughput. +- `http2_adaptive_window(bool)` -- A/Bs HTTP/2 adaptive flow-control window sizing. + +These are **intentionally retained**. They look unused from a production-code grep because production code never reaches them (no CLI flag, no env var, no config setting). Removing them as "dead code" deletes load-bearing diagnostic infrastructure: any future perf investigation that needs to reproduce the HTTP/2 stall investigation (or any successor) would have to re-introduce the same plumbing from scratch. + +If you're tempted to delete them, search `tests/perf_profile.rs` first -- it references all three by name. + ## Commands ```bash diff --git a/crates/ocync-distribution/Cargo.toml b/crates/ocync-distribution/Cargo.toml index 41555d17..942e606e 100644 --- a/crates/ocync-distribution/Cargo.toml +++ b/crates/ocync-distribution/Cargo.toml @@ -30,6 +30,7 @@ futures-util = { version = "0.3", default-features = false } google-cloud-auth.workspace = true http.workspace = true hex.workspace = true +pin-project-lite.workspace = true regex-lite = { version = "0.1", default-features = false } reqwest = { workspace = true, features = ["http2", "rustls-no-provider", "stream", "json"] } rustls = { version = "0.23", default-features = false, features = ["aws-lc-rs"] } diff --git a/crates/ocync-distribution/src/blob.rs b/crates/ocync-distribution/src/blob.rs index 530ff242..71f78f84 100644 --- a/crates/ocync-distribution/src/blob.rs +++ b/crates/ocync-distribution/src/blob.rs @@ -1,9 +1,14 @@ //! Blob operations - existence checks, pull, push, mount, and upload management. +use std::pin::Pin; +use std::task::{Context, Poll}; + use bytes::Bytes; use futures_util::{Stream, StreamExt}; use http::StatusCode; -use reqwest::header::{CONTENT_LENGTH, CONTENT_TYPE, HeaderValue, LOCATION}; +use pin_project_lite::pin_project; +use reqwest::header::{CONTENT_LENGTH, CONTENT_RANGE, CONTENT_TYPE, HeaderValue, LOCATION}; +use tokio::sync::OwnedSemaphorePermit; use tracing::{debug, warn}; use crate::aimd::RegistryAction; @@ -15,9 +20,56 @@ use crate::error::Error; use crate::sha256::Sha256; use crate::spec::RepositoryName; +pin_project! { + /// Stream wrapper that holds a streaming-blob semaphore permit for the + /// lifetime of the inner stream. Used by [`RegistryClient::blob_pull`] to + /// release the permit when the caller has fully consumed the response + /// body, not when `blob_pull` itself returns. + /// + /// `pin_project_lite` lets the projection work for any `S: Stream`, + /// including streams that are not `Unpin` (a future change in + /// reqwest's stream impl would otherwise silently bypass this + /// wrapper). + struct PermitStream { + #[pin] + inner: S, + _permit: OwnedSemaphorePermit, + } +} + +impl Stream for PermitStream { + type Item = S::Item; + + fn poll_next(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll> { + self.project().inner.poll_next(cx) + } + + fn size_hint(&self) -> (usize, Option) { + // Forward the inner stream's hint so consumers that pre-allocate + // (e.g. `reqwest::Body::wrap_stream` setting Content-Length, or + // any `collect::>()`) see the same information as the raw + // `bytes_stream()`. + self.inner.size_hint() + } +} + /// Content type for raw blob data in OCI upload requests. const OCTET_STREAM: &str = "application/octet-stream"; +/// Per-PATCH chunk size for the ACR upload fallback. ACR rejects +/// streaming PUT bodies above ~20 MB; 16 MB stays comfortably under that +/// while keeping the round-trip count low. +const ACR_PATCH_CHUNK_SIZE: usize = 16 * 1024 * 1024; + +/// Upper bound on the `Vec::with_capacity` hint used by [`buffer_stream`]. +/// +/// The hint comes from a manifest's declared blob size, which is +/// attacker-controllable. Without a cap, a single malicious manifest can +/// trigger a multi-GB allocation per concurrent fallback upload. Capping +/// at 16 MB keeps the up-front allocation modest; larger streams grow the +/// Vec organically through amortised doubling. +const MAX_BUFFER_PREALLOC: usize = 16 * 1024 * 1024; + /// Result of a cross-repository blob mount attempt. #[derive(Debug)] pub enum MountResult { @@ -63,17 +115,24 @@ async fn expect_status( /// Buffer an entire byte stream into a `Vec`. /// -/// When `capacity_hint` is `Some(n)`, the buffer is pre-allocated to `n` bytes -/// to avoid reallocations. Used by fallback upload paths (GAR, GHCR, monolithic -/// threshold) that cannot stream chunks to the registry. +/// When `capacity_hint` is `Some(n)`, the buffer is pre-allocated to +/// `min(n, MAX_BUFFER_PREALLOC)` bytes -- the manifest-declared size +/// is capped at [`MAX_BUFFER_PREALLOC`] because the hint is +/// attacker-controllable; larger streams grow the `Vec` organically via +/// amortised doubling. Used by fallback upload paths (GAR, GHCR, ACR) +/// that cannot stream chunks to the registry. async fn buffer_stream( stream: impl Stream>, capacity_hint: Option, ) -> Result, Error> { - let mut body = match capacity_hint { - Some(s) => Vec::with_capacity(s as usize), - None => Vec::new(), - }; + // Cap the pre-allocation. capacity_hint comes from the manifest's + // declared blob size, which is attacker-controllable; an unbounded + // `with_capacity` plus the per-target streaming-blob concurrency + // budget multiplies into arbitrary memory pressure. + let cap = capacity_hint + .map(|s| (s as usize).min(MAX_BUFFER_PREALLOC)) + .unwrap_or(0); + let mut body = Vec::with_capacity(cap); futures_util::pin_mut!(stream); while let Some(chunk) = stream.next().await { body.extend_from_slice(&chunk?); @@ -119,17 +178,23 @@ impl RegistryClient { /// Pull a blob as a streaming response. /// /// Issues a GET request to `/v2/{repository}/blobs/{digest}` and returns - /// a byte stream for the response body. + /// a byte stream for the response body. The streaming-blob semaphore + /// is acquired before the GET and held by the returned stream until it + /// is fully consumed, bounding concurrent long-lived h2 streams. pub async fn blob_pull( &self, repository: &RepositoryName, digest: &Digest, ) -> Result> + 'static, Error> { + let permit = self.acquire_streaming_blob_permit().await; let path = blob_path(digest); let resp = self .get(repository, &path, None, RegistryAction::BlobRead) .await?; - Ok(resp.bytes_stream()) + Ok(PermitStream { + inner: resp.bytes_stream(), + _permit: permit, + }) } /// Attempt a cross-repository blob mount. @@ -182,19 +247,10 @@ impl RegistryClient { let hash = Sha256::digest(data); let digest = Digest::from_sha256(hash); - let url = build_url(&self.base_url, repository, "blobs/uploads/")?; let scopes = [Scope::pull_push(repository.as_str())]; - - let resp = self - .send_with_aimd( - RegistryAction::BlobUploadInit, - &scopes, - "blob push initiate", - |headers| self.http.post(url.clone()).headers(headers), - ) + let put_url = self + .initiate_blob_upload(repository, &scopes, "blob push initiate") .await?; - let resp = expect_status(resp, StatusCode::ACCEPTED, &self.base_url, repository).await?; - let put_url = extract_location(&resp, &self.base_url)?; let digest_str = digest.to_string(); let resp = self @@ -233,6 +289,11 @@ impl RegistryClient { /// **GAR fallback**: Google Artifact Registry does not support chunked /// uploads, so hosts ending in `-docker.pkg.dev` buffer the entire stream /// and delegate to [`blob_push`](Self::blob_push). + /// + /// **ACR fallback**: Azure Container Registry rejects streaming PUT bodies + /// above ~20 MB. Hosts on `*.azurecr.io` buffer the stream and upload via + /// multiple chunked `PATCHes` under [`ACR_PATCH_CHUNK_SIZE`] each, then PUT + /// to finalize. pub async fn blob_push_stream( &self, repository: &RepositoryName, @@ -249,23 +310,49 @@ impl RegistryClient { // Registry-specific upload fallbacks, detected via the canonical // provider detection (handles case-insensitivity, ports, trailing dots). + // + // A `match` (rather than an `if`-chain) so that adding a new + // `ProviderKind` with its own upload quirks fails to compile until + // the dispatch is updated, instead of silently falling into the + // default streaming-PUT path. let provider = self.base_url.host_str().and_then(detect_provider_kind); - - // GHCR fallback: single PATCH (no Content-Range) to avoid the - // multi-PATCH corruption bug. - if provider == Some(ProviderKind::Ghcr) { - return self - .blob_push_stream_ghcr(repository, expected_digest, known_size, stream) - .await; - } - - // GAR fallback: buffer entire stream and use monolithic push. - // GAR (Artifact Registry) does not support chunked uploads; legacy - // gcr.io hosts support it fine, so only Gar triggers this path. - if provider == Some(ProviderKind::Gar) { - return self - .blob_push_stream_gar_fallback(repository, expected_digest, known_size, stream) - .await; + match provider { + // GHCR: single PATCH (no Content-Range) to avoid the + // multi-PATCH corruption bug. + Some(ProviderKind::Ghcr) => { + return self + .blob_push_stream_ghcr(repository, expected_digest, known_size, stream) + .await; + } + // GAR: buffer entire stream and use monolithic push. + // GAR (Artifact Registry) does not support chunked uploads; + // legacy gcr.io hosts support it fine, so only Gar triggers + // this path. + Some(ProviderKind::Gar) => { + return self + .blob_push_stream_gar(repository, expected_digest, known_size, stream) + .await; + } + // ACR: chunked PATCH under ACR's ~20 MB streaming-PUT body + // limit. Buffers the stream so the digest can be verified + // before any PATCH fires (avoids wasted upload bandwidth on + // corruption), then splits into ACR_PATCH_CHUNK_SIZE chunks. + Some(ProviderKind::Acr) => { + return self + .blob_push_stream_acr(repository, expected_digest, known_size, stream) + .await; + } + // Default: streaming PUT below. Every other ProviderKind + // (Ecr, EcrPublic, Gcr, DockerHub, Chainguard) uses the + // default path; new variants must be added here explicitly. + Some( + ProviderKind::Ecr + | ProviderKind::EcrPublic + | ProviderKind::Gcr + | ProviderKind::DockerHub + | ProviderKind::Chainguard, + ) + | None => {} } debug!( @@ -274,20 +361,14 @@ impl RegistryClient { "starting streaming blob upload" ); - let url = build_url(&self.base_url, repository, "blobs/uploads/")?; let scopes = [Scope::pull_push(repository.as_str())]; - - // POST to initiate the upload session. - let resp = self - .send_with_aimd( - RegistryAction::BlobUploadInit, - &scopes, - "blob push stream initiate", - |headers| self.http.post(url.clone()).headers(headers), - ) + let upload_url = self + .initiate_blob_upload(repository, &scopes, "blob push stream initiate") .await?; - let resp = expect_status(resp, StatusCode::ACCEPTED, &self.base_url, repository).await?; - let upload_url = extract_location(&resp, &self.base_url)?; + + // Bound concurrent long-lived h2 streams. The streaming PUT body + // is the long-lived part; the POST above completes quickly. + let _stream_permit = self.acquire_streaming_blob_permit().await; // Streaming PUT - send the blob body through a single HTTP request. // Uses Transfer-Encoding: chunked (no Content-Length), so the body @@ -318,13 +399,35 @@ impl RegistryClient { Ok(expected_digest.clone()) } + /// POST `/v2/{repository}/blobs/uploads/` to initiate an upload session. + /// + /// Returns the upload URL extracted from the response's `Location` header. + /// Shared by every blob upload path (monolithic, streaming, GHCR/GAR/ACR + /// fallbacks) so the response-classification, expected-status, and + /// Location-extraction sequence has a single implementation. + async fn initiate_blob_upload( + &self, + repository: &RepositoryName, + scopes: &[Scope], + log_context: &str, + ) -> Result { + let url = build_url(&self.base_url, repository, "blobs/uploads/")?; + let resp = self + .send_with_aimd(RegistryAction::BlobUploadInit, scopes, log_context, |h| { + self.http.post(url.clone()).headers(h) + }) + .await?; + let resp = expect_status(resp, StatusCode::ACCEPTED, &self.base_url, repository).await?; + extract_location(&resp, &self.base_url) + } + /// GAR fallback: buffer the entire stream and delegate to monolithic push. /// /// Google Artifact Registry does not support chunked uploads, so the /// entire stream is buffered in memory and sent as a monolithic upload. /// The digest returned by the monolithic push is verified against the /// caller's expected digest to catch data corruption. - async fn blob_push_stream_gar_fallback( + async fn blob_push_stream_gar( &self, repository: &RepositoryName, expected_digest: &Digest, @@ -336,6 +439,14 @@ impl RegistryClient { host = self.base_url.host_str().unwrap_or("unknown"), "GAR does not support chunked uploads; buffering entire blob in memory" ); + // Two complementary caps for buffered fallback uploads: + // 1. `streaming_blob_sem` bounds the *count* of in-flight + // buffered uploads (also doubles as h2-stream cap). + // 2. `buffered_blob_bytes_sem` bounds total *bytes* held + // across in-flight buffered uploads. + // The helper acquires both in the right order; the returned + // tuple drops bytes_permit before stream_permit on scope exit. + let _permits = self.acquire_buffered_upload_permits(known_size).await; let body = buffer_stream(stream, known_size).await?; let actual_digest = self.blob_push(repository, &body).await?; @@ -367,20 +478,13 @@ impl RegistryClient { "GHCR multi-PATCH chunked upload is broken; buffering blob for single-PATCH upload" ); - let url = build_url(&self.base_url, repository, "blobs/uploads/")?; + // Count cap (h2-stream) + byte cap (cross-call memory budget). + // See `blob_push_stream_gar` for the layering rationale. + let _permits = self.acquire_buffered_upload_permits(known_size).await; let scopes = [Scope::pull_push(repository.as_str())]; - - // Initiate upload. - let resp = self - .send_with_aimd( - RegistryAction::BlobUploadInit, - &scopes, - "blob push ghcr initiate", - |headers| self.http.post(url.clone()).headers(headers), - ) + let upload_url = self + .initiate_blob_upload(repository, &scopes, "blob push ghcr initiate") .await?; - let resp = expect_status(resp, StatusCode::ACCEPTED, &self.base_url, repository).await?; - let upload_url = extract_location(&resp, &self.base_url)?; // Buffer entire stream and verify digest before uploading. let raw = buffer_stream(stream, known_size).await?; @@ -434,9 +538,119 @@ impl RegistryClient { Ok(expected_digest.clone()) } + + /// ACR fallback: chunked PATCH under Azure Container Registry's + /// ~20 MB streaming-PUT body limit. + /// + /// Buffers the full stream so the digest can be verified before any + /// PATCH is sent (avoiding wasted bandwidth on corruption), then + /// uploads via [`ACR_PATCH_CHUNK_SIZE`]-byte PATCH requests with the + /// OCI Content-Range format `{start}-{end}` (NOT RFC 7233), followed + /// by a PUT to finalize with the digest query param. Each PATCH + /// response carries a fresh `Location` header that is used as the + /// upload URL for the next PATCH (or the finalize PUT). + async fn blob_push_stream_acr( + &self, + repository: &RepositoryName, + expected_digest: &Digest, + known_size: Option, + stream: impl Stream>, + ) -> Result { + warn!( + repository = repository.as_str(), + "ACR rejects streaming PUT above ~20 MB; buffering blob for chunked PATCH upload" + ); + + // Count cap (h2-stream) + byte cap (cross-call memory budget). + // See `blob_push_stream_gar` for the layering rationale. + let _permits = self.acquire_buffered_upload_permits(known_size).await; + + let raw = buffer_stream(stream, known_size).await?; + let actual_digest = Digest::from_sha256(Sha256::digest(&raw)); + if &actual_digest != expected_digest { + return Err(Error::DigestMismatch { + expected: expected_digest.clone(), + actual: actual_digest, + }); + } + let total_len = raw.len() as u64; + let body = Bytes::from(raw); + + let scopes = [Scope::pull_push(repository.as_str())]; + let mut upload_url = self + .initiate_blob_upload(repository, &scopes, "blob push acr initiate") + .await?; + // `extract_location` enforces same-origin against base_url on + // every upload-chain response (POST initiate, each PATCH, and + // the finalize PUT), so no separate per-PATCH host check is + // needed here. + + // Chunked PATCH. For a zero-byte blob the loop never executes; the + // finalize PUT below carries the digest of the empty blob and + // completes the upload session per the OCI distribution spec + // (PUT-without-prior-PATCH on an empty blob is valid). + let mut offset: u64 = 0; + while offset < total_len { + let chunk_len = (ACR_PATCH_CHUNK_SIZE as u64).min(total_len - offset); + let chunk = body.slice((offset as usize)..((offset + chunk_len) as usize)); + let range_end = offset + chunk_len - 1; + // OCI spec Content-Range format: `{start}-{end}` (NOT RFC 7233). + let range_header = format!("{offset}-{range_end}"); + let chunk_len_str = chunk_len.to_string(); + + let resp = self + .send_with_aimd( + RegistryAction::BlobUploadChunk, + &scopes, + "blob push acr patch", + |headers| { + self.http + .patch(&upload_url) + .headers(headers) + .header(CONTENT_LENGTH, &chunk_len_str) + .header(CONTENT_RANGE, &range_header) + .header(CONTENT_TYPE, HeaderValue::from_static(OCTET_STREAM)) + .body(chunk.clone()) + }, + ) + .await?; + let resp = + expect_status(resp, StatusCode::ACCEPTED, &self.base_url, repository).await?; + upload_url = extract_location(&resp, &self.base_url)?; + offset += chunk_len; + } + + // Finalize. + let digest_str = expected_digest.to_string(); + let resp = self + .send_with_aimd( + RegistryAction::BlobUploadComplete, + &scopes, + "blob push acr finalize", + |headers| { + self.http + .put(&upload_url) + .headers(headers) + .query(&[("digest", &digest_str)]) + .header(CONTENT_LENGTH, "0") + .header(CONTENT_TYPE, HeaderValue::from_static(OCTET_STREAM)) + }, + ) + .await?; + expect_status(resp, StatusCode::CREATED, &self.base_url, repository).await?; + + Ok(expected_digest.clone()) + } } -/// Extract and resolve the Location header from an upload response. +/// Extract and resolve the `Location` header from an upload response, +/// enforcing same-origin against `base_url`. +/// +/// Used at every step of the blob-upload chain (POST initiate, PATCH +/// chunk, finalize PUT). The chain is authenticated with the registry's +/// bearer token; a `Location` pointing to a different host would leak +/// the bearer on the next request. Reject any cross-host hand-off +/// regardless of which step returned it. fn extract_location(resp: &reqwest::Response, base_url: &url::Url) -> Result { let raw = resp .headers() @@ -446,16 +660,31 @@ fn extract_location(resp: &reqwest::Response, base_url: &url::Url) -> Result impl Stream> { + ) -> impl Stream> + 'static + use<> { let chunks: Vec> = data .chunks(chunk_size) .map(|c| Ok(Bytes::copy_from_slice(c))) @@ -723,6 +952,365 @@ mod tests { assert_eq!(result, digest); } + /// ACR: small blob (under chunk size) takes the ACR path with one + /// PATCH carrying an OCI-format `Content-Range` header + /// (`{start}-{end}`, not RFC 7233). + #[tokio::test] + async fn blob_push_stream_acr_small_blob_single_patch_with_content_range() { + let server = wiremock::MockServer::start().await; + let data = b"acr blob content fits in one patch"; + let digest = test_digest(data); + let port = url::Url::parse(&server.uri()).unwrap().port().unwrap(); + + wiremock::Mock::given(wiremock::matchers::method("POST")) + .and(wiremock::matchers::path("/v2/myreg/myimg/blobs/uploads/")) + .respond_with( + wiremock::ResponseTemplate::new(202) + .append_header("Location", "/v2/myreg/myimg/blobs/uploads/acr-uuid"), + ) + .expect(1) + .mount(&server) + .await; + + // PATCH must carry a Content-Range header in OCI `{start}-{end}` + // format. Exact match catches off-by-one (matches the precise + // `0-{len-1}` value -- a regex like `^0-\d+$` would silently accept + // an off-by-one in `range_end`). + let expected_range = format!("0-{}", data.len() - 1); + wiremock::Mock::given(wiremock::matchers::method("PATCH")) + .and(wiremock::matchers::path( + "/v2/myreg/myimg/blobs/uploads/acr-uuid", + )) + .and(wiremock::matchers::header( + "content-range", + expected_range.as_str(), + )) + .respond_with(wiremock::ResponseTemplate::new(202).append_header( + "Location", + "/v2/myreg/myimg/blobs/uploads/acr-uuid?after-patch", + )) + .expect(1) + .mount(&server) + .await; + + wiremock::Mock::given(wiremock::matchers::method("PUT")) + .and(wiremock::matchers::query_param( + "digest", + digest.to_string(), + )) + .respond_with(wiremock::ResponseTemplate::new(201)) + .expect(1) + .mount(&server) + .await; + + let client = build_test_client("myacr.azurecr.io", port); + let repo = RepositoryName::new("myreg/myimg").unwrap(); + let result = client + .blob_push_stream(&repo, &digest, None, data_stream(data, 4)) + .await + .unwrap(); + assert_eq!(result, digest); + } + + /// ACR: blob larger than `ACR_PATCH_CHUNK_SIZE` splits into multiple + /// `PATCHes`, each under the 20 MB ceiling, with increasing + /// `Content-Range` offsets. + #[tokio::test] + async fn blob_push_stream_acr_large_blob_splits_into_chunks() { + // 17 MB -- just over the 16 MB chunk size, so we get 2 PATCHes + // (one full 16 MB chunk + one 1 MB tail). + let data = vec![0xABu8; 17 * 1024 * 1024]; + let total_len = data.len() as u64; + let server = wiremock::MockServer::start().await; + let digest = test_digest(&data); + let port = url::Url::parse(&server.uri()).unwrap().port().unwrap(); + + wiremock::Mock::given(wiremock::matchers::method("POST")) + .and(wiremock::matchers::path("/v2/proj/img/blobs/uploads/")) + .respond_with( + wiremock::ResponseTemplate::new(202) + .append_header("Location", "/v2/proj/img/blobs/uploads/acr-1"), + ) + .expect(1) + .mount(&server) + .await; + + // First chunk: 16 MB at offset 0. Content-Range: 0-16777215. + let first_end = ACR_PATCH_CHUNK_SIZE as u64 - 1; + wiremock::Mock::given(wiremock::matchers::method("PATCH")) + .and(wiremock::matchers::path("/v2/proj/img/blobs/uploads/acr-1")) + .and(wiremock::matchers::header( + "content-range", + format!("0-{first_end}").as_str(), + )) + .respond_with( + wiremock::ResponseTemplate::new(202) + .append_header("Location", "/v2/proj/img/blobs/uploads/acr-2"), + ) + .expect(1) + .mount(&server) + .await; + + // Second chunk: 1 MB at offset 16 MB. + let total_end = total_len - 1; + wiremock::Mock::given(wiremock::matchers::method("PATCH")) + .and(wiremock::matchers::path("/v2/proj/img/blobs/uploads/acr-2")) + .and(wiremock::matchers::header( + "content-range", + format!("{}-{total_end}", ACR_PATCH_CHUNK_SIZE).as_str(), + )) + .respond_with( + wiremock::ResponseTemplate::new(202) + .append_header("Location", "/v2/proj/img/blobs/uploads/acr-3"), + ) + .expect(1) + .mount(&server) + .await; + + wiremock::Mock::given(wiremock::matchers::method("PUT")) + .and(wiremock::matchers::path("/v2/proj/img/blobs/uploads/acr-3")) + .and(wiremock::matchers::query_param( + "digest", + digest.to_string(), + )) + .respond_with(wiremock::ResponseTemplate::new(201)) + .expect(1) + .mount(&server) + .await; + + let client = build_test_client("myacr.azurecr.io", port); + let repo = RepositoryName::new("proj/img").unwrap(); + let result = client + .blob_push_stream(&repo, &digest, Some(total_len), data_stream(&data, 65536)) + .await + .unwrap(); + assert_eq!(result, digest); + } + + /// ACR: zero-byte blob (e.g. the OCI empty config used by signature + /// referrers) skips the PATCH loop and goes straight to finalize PUT. + #[tokio::test] + async fn blob_push_stream_acr_zero_byte_blob_skips_patch_loop() { + let server = wiremock::MockServer::start().await; + let data: &[u8] = b""; + let digest = test_digest(data); + let port = url::Url::parse(&server.uri()).unwrap().port().unwrap(); + + wiremock::Mock::given(wiremock::matchers::method("POST")) + .and(wiremock::matchers::path("/v2/repo/blobs/uploads/")) + .respond_with( + wiremock::ResponseTemplate::new(202) + .append_header("Location", "/v2/repo/blobs/uploads/empty-id"), + ) + .expect(1) + .mount(&server) + .await; + + // PATCH must NOT be issued for a zero-byte blob. + wiremock::Mock::given(wiremock::matchers::method("PATCH")) + .respond_with(wiremock::ResponseTemplate::new(500)) + .expect(0) + .mount(&server) + .await; + + // Finalize PUT carries the empty-blob digest and Content-Length: 0. + wiremock::Mock::given(wiremock::matchers::method("PUT")) + .and(wiremock::matchers::path("/v2/repo/blobs/uploads/empty-id")) + .and(wiremock::matchers::query_param( + "digest", + digest.to_string(), + )) + .and(wiremock::matchers::header("content-length", "0")) + .respond_with(wiremock::ResponseTemplate::new(201)) + .expect(1) + .mount(&server) + .await; + + let client = build_test_client("myacr.azurecr.io", port); + let repo = RepositoryName::new("repo").unwrap(); + let result = client + .blob_push_stream(&repo, &digest, Some(0), data_stream(data, 4)) + .await + .unwrap(); + assert_eq!(result, digest); + } + + /// Cross-host `Location` returned at any step of the upload chain (POST + /// initiate, PATCH chunk, finalize PUT) must be rejected by + /// `extract_location`'s same-origin check. The registry bearer would + /// otherwise be forwarded to an attacker host on the next request. + #[tokio::test] + async fn blob_push_stream_acr_cross_host_patch_location_is_rejected() { + let registry = wiremock::MockServer::start().await; + let attacker = wiremock::MockServer::start().await; + let data = b"acr cross-host rejection test bytes"; + let digest = test_digest(data); + let registry_port = url::Url::parse(®istry.uri()).unwrap().port().unwrap(); + + wiremock::Mock::given(wiremock::matchers::method("POST")) + .and(wiremock::matchers::path("/v2/r/i/blobs/uploads/")) + .respond_with( + wiremock::ResponseTemplate::new(202) + .append_header("Location", "/v2/r/i/blobs/uploads/acr-1"), + ) + .expect(1) + .mount(®istry) + .await; + + // First PATCH returns a Location at the attacker host. The client + // must reject this before issuing any subsequent request. + let attacker_url = format!("{}/v2/r/i/blobs/uploads/exfil", attacker.uri()); + wiremock::Mock::given(wiremock::matchers::method("PATCH")) + .and(wiremock::matchers::path("/v2/r/i/blobs/uploads/acr-1")) + .respond_with( + wiremock::ResponseTemplate::new(202).append_header("Location", attacker_url), + ) + .expect(1) + .mount(®istry) + .await; + + // Attacker host MUST receive no requests. + wiremock::Mock::given(wiremock::matchers::any()) + .respond_with(wiremock::ResponseTemplate::new(200)) + .expect(0) + .mount(&attacker) + .await; + + // Finalize PUT MUST NOT be issued at the registry either. + wiremock::Mock::given(wiremock::matchers::method("PUT")) + .respond_with(wiremock::ResponseTemplate::new(500)) + .expect(0) + .mount(®istry) + .await; + + let client = build_test_client("myacr.azurecr.io", registry_port); + let repo = RepositoryName::new("r/i").unwrap(); + let err = client + .blob_push_stream(&repo, &digest, None, data_stream(data, 4)) + .await + .expect_err("cross-host PATCH Location must be rejected"); + let msg = err.to_string(); + assert!( + msg.contains("differs from base host"), + "expected same-origin rejection error, got: {msg}" + ); + } + + /// `blob_pull` follows 3xx redirects through reqwest's default policy. + /// ECR private serves blob bytes via a 307 to a presigned S3 URL; this + /// test exercises the same redirect-following path locally. + #[tokio::test] + async fn blob_pull_follows_307_redirect_to_blob_bytes() { + let server = wiremock::MockServer::start().await; + let port = url::Url::parse(&server.uri()).unwrap().port().unwrap(); + let payload = b"final blob bytes from redirected location"; + let digest = test_digest(payload); + let blob_path_str = format!("/v2/some/repo/blobs/{digest}"); + + // First request: 307 with Location pointing at the data path. + wiremock::Mock::given(wiremock::matchers::method("GET")) + .and(wiremock::matchers::path(blob_path_str.clone())) + .respond_with( + wiremock::ResponseTemplate::new(307).append_header("Location", "/blob-data"), + ) + .expect(1) + .mount(&server) + .await; + + // Redirected request: serves the actual bytes. + wiremock::Mock::given(wiremock::matchers::method("GET")) + .and(wiremock::matchers::path("/blob-data")) + .respond_with(wiremock::ResponseTemplate::new(200).set_body_bytes(payload.to_vec())) + .expect(1) + .mount(&server) + .await; + + let client = build_test_client("localhost", port); + let repo = RepositoryName::new("some/repo").unwrap(); + let stream = client.blob_pull(&repo, &digest).await.unwrap(); + let body: Vec = StreamExt::collect::>(stream) + .await + .into_iter() + .map(|r| r.unwrap()) + .flat_map(|b| b.to_vec()) + .collect(); + assert_eq!(body, payload); + } + + /// `blob_pull` follows cross-host 307 (e.g. ECR -> S3 presigned URL) and + /// reqwest strips the `Authorization` header on the redirected request. + /// The S3 URL's signature lives in the query string, so missing Authorization + /// is the correct outcome -- forwarding registry credentials to a third-party + /// host would leak them. + #[tokio::test] + async fn blob_pull_cross_host_redirect_strips_authorization() { + use std::sync::Arc; + use std::sync::atomic::{AtomicBool, Ordering}; + + let registry = wiremock::MockServer::start().await; + let backend = wiremock::MockServer::start().await; + let registry_port = url::Url::parse(®istry.uri()).unwrap().port().unwrap(); + + let payload = b"cross-host bytes"; + let digest = test_digest(payload); + let blob_path_str = format!("/v2/repo/blobs/{digest}"); + + // Registry: 307 redirect to the backend's full URL. + let backend_url = format!("{}/blob-data", backend.uri()); + wiremock::Mock::given(wiremock::matchers::method("GET")) + .and(wiremock::matchers::path(blob_path_str.clone())) + .respond_with( + wiremock::ResponseTemplate::new(307).append_header("Location", backend_url), + ) + .expect(1) + .mount(®istry) + .await; + + // Backend: assert NO Authorization header present, serve the bytes. + let saw_auth = Arc::new(AtomicBool::new(false)); + let saw_auth_clone = Arc::clone(&saw_auth); + wiremock::Mock::given(wiremock::matchers::method("GET")) + .and(wiremock::matchers::path("/blob-data")) + .respond_with(move |req: &wiremock::Request| { + if req.headers.contains_key("authorization") { + saw_auth_clone.store(true, Ordering::SeqCst); + } + wiremock::ResponseTemplate::new(200).set_body_bytes(payload.to_vec()) + }) + .expect(1) + .mount(&backend) + .await; + + // Use a client with a fake bearer to confirm the header would have + // been on the original request had reqwest forwarded it. + let base = url::Url::parse(&format!("http://localhost:{registry_port}")).unwrap(); + let client = crate::client::RegistryClientBuilder::new(base) + .resolve( + "localhost", + std::net::SocketAddr::from(([127, 0, 0, 1], registry_port)), + ) + .auth(crate::auth::static_token::StaticTokenAuth::new( + "localhost", + "fake-leak-canary-token", + )) + .build() + .unwrap(); + + let repo = RepositoryName::new("repo").unwrap(); + let stream = client.blob_pull(&repo, &digest).await.unwrap(); + let body: Vec = StreamExt::collect::>(stream) + .await + .into_iter() + .map(|r| r.unwrap()) + .flat_map(|b| b.to_vec()) + .collect(); + assert_eq!(body, payload); + assert!( + !saw_auth.load(Ordering::SeqCst), + "Authorization header must NOT be forwarded across hosts -- would leak registry credentials to the redirect target" + ); + } + /// GHCR: exactly one PATCH regardless of blob size vs `chunk_size`. #[tokio::test] async fn blob_push_stream_ghcr_single_patch_large_blob() { @@ -905,4 +1493,154 @@ mod tests { assert_eq!(result, digest); } + + /// Pins the byte-budget queueing contract for the buffered fallback + /// paths (GHCR / GAR / ACR). + /// + /// Two concurrent GHCR uploads, each declaring half the budget plus + /// a few bytes, would together exceed the byte budget if they ran + /// fully overlapped. The first upload is delayed by the wiremock + /// mock (200 ms PATCH/PUT response delay) so the second must wait + /// on the byte-budget permit before its own POST initiates. + /// + /// Asserts: the second upload's POST start time is at least + /// `delay` after the first upload's POST start time. This is the + /// observable signal that the byte-budget semaphore actually queued + /// the second upload behind the first. + #[tokio::test] + async fn buffered_upload_byte_budget_queues_concurrent_uploads() { + use std::sync::Arc; + use std::sync::Mutex as StdMutex; + use std::time::Instant; + use tokio::time::Duration; + use url::Url; + use wiremock::{Mock, MockServer, ResponseTemplate, matchers}; + + let server = MockServer::start().await; + let port = Url::parse(&server.uri()).unwrap().port().unwrap(); + let post_times: Arc>> = Arc::new(StdMutex::new(Vec::new())); + let post_times_clone = Arc::clone(&post_times); + + // Two distinct repos so the POST paths don't collide; record the + // POST timestamps so we can verify the ordering after the fact. + Mock::given(matchers::method("POST")) + .and(matchers::path_regex(r"^/v2/repo-[ab]/blobs/uploads/$")) + .respond_with(move |req: &wiremock::Request| { + post_times_clone.lock().unwrap().push(Instant::now()); + // Return a Location pointing back at the same upload path so + // the PATCH+PUT chain can complete. + let uri_path = req.url.path(); + let upload_path = format!("{uri_path}upload-id"); + ResponseTemplate::new(202).append_header("Location", upload_path) + }) + .mount(&server) + .await; + + // PATCH delays 200ms so the first upload holds its byte permit + // long enough for the second to definitively wait. PUT is fast. + let patch_delay = Duration::from_millis(200); + Mock::given(matchers::method("PATCH")) + .respond_with( + ResponseTemplate::new(202) + .append_header("Location", "/v2/done") + .set_delay(patch_delay), + ) + .mount(&server) + .await; + Mock::given(matchers::method("PUT")) + .respond_with(ResponseTemplate::new(201)) + .mount(&server) + .await; + + // 6 MB blobs: large enough that two together (12 MB) exceed an + // 8 MB budget. Use distinct contents per upload so the digests + // differ. + let blob_a = vec![0xAAu8; 6 * 1024 * 1024]; + let blob_b = vec![0xBBu8; 6 * 1024 * 1024]; + let digest_a = test_digest(&blob_a); + let digest_b = test_digest(&blob_b); + + // Use the GHCR fallback path (single-PATCH then PUT, buffered), + // with a small 8 MB byte budget. Two 6 MB uploads must serialize. + let base_url = Url::parse(&format!("http://ghcr.io:{port}")).unwrap(); + let client = Arc::new( + crate::client::RegistryClientBuilder::new(base_url) + .resolve( + "ghcr.io", + std::net::SocketAddr::from(([127, 0, 0, 1], port)), + ) + .buffered_blob_bytes(8 * 1024 * 1024) + .build() + .unwrap(), + ); + + let repo_a = RepositoryName::new("repo-a").unwrap(); + let repo_b = RepositoryName::new("repo-b").unwrap(); + + let client_1 = Arc::clone(&client); + let blob_a_clone = blob_a.clone(); + let digest_a_clone = digest_a.clone(); + let h1 = tokio::spawn(async move { + client_1 + .blob_push_stream( + &repo_a, + &digest_a_clone, + Some(blob_a_clone.len() as u64), + data_stream(&blob_a_clone, 65536), + ) + .await + }); + + // Deterministic synchronization (no head-start race): wait until + // task 1's POST has been recorded -- by that point task 1 has + // already acquired the byte permit, so task 2 is guaranteed to + // see the budget pressure when it tries. Polls a shared + // `post_times` Mutex on a short cadence and bails out after a + // generous deadline if task 1 never makes progress (treating + // that as a separate test failure). + let wait_for_first_post = async { + loop { + if !post_times.lock().unwrap().is_empty() { + return; + } + tokio::time::sleep(Duration::from_millis(5)).await; + } + }; + tokio::time::timeout(Duration::from_secs(5), wait_for_first_post) + .await + .expect("first upload POST must fire within 5s (task 1 likely stuck)"); + + let client_2 = Arc::clone(&client); + let blob_b_clone = blob_b.clone(); + let digest_b_clone = digest_b.clone(); + let h2 = tokio::spawn(async move { + client_2 + .blob_push_stream( + &repo_b, + &digest_b_clone, + Some(blob_b_clone.len() as u64), + data_stream(&blob_b_clone, 65536), + ) + .await + }); + + let r1 = h1.await.unwrap(); + let r2 = h2.await.unwrap(); + assert!(r1.is_ok(), "first upload should succeed: {:?}", r1.err()); + assert!(r2.is_ok(), "second upload should succeed: {:?}", r2.err()); + + let times = post_times.lock().unwrap(); + assert_eq!(times.len(), 2, "both POSTs must have fired"); + let gap = times[1].duration_since(times[0]); + // Task 2 cannot POST until task 1 releases the byte permit, + // which happens when task 1's `blob_push_stream` returns + // (after PATCH-delay + fast PUT). The lower bound is + // patch_delay minus a generous CI margin (100 ms) since we + // already eliminated the head-start race above. + let lower_bound = patch_delay.saturating_sub(Duration::from_millis(100)); + assert!( + gap >= lower_bound, + "second POST started after only {gap:?}; expected the byte budget to queue it for at least ~{lower_bound:?} behind the first (patch_delay = {patch_delay:?})" + ); + } } diff --git a/crates/ocync-distribution/src/client.rs b/crates/ocync-distribution/src/client.rs index b11ea7fe..5e5e5462 100644 --- a/crates/ocync-distribution/src/client.rs +++ b/crates/ocync-distribution/src/client.rs @@ -1,9 +1,11 @@ //! HTTP client for a single OCI registry endpoint. +use std::sync::Arc; use std::sync::atomic::{AtomicU64, Ordering}; use http::StatusCode; use reqwest::header::{ACCEPT, AUTHORIZATION, HeaderMap, HeaderValue}; +use tokio::sync::{OwnedSemaphorePermit, Semaphore}; use url::Url; use crate::aimd::{AimdController, RegistryAction}; @@ -14,6 +16,45 @@ use crate::spec::{RegistryAuthority, RepositoryName}; const DEFAULT_MAX_CONCURRENT_REQUESTS: usize = 50; const USER_AGENT_VALUE: &str = concat!("ocync/", env!("CARGO_PKG_VERSION")); +/// Default cap on concurrent long-lived blob streams per registry. +/// +/// reqwest/hyper opens one HTTP/2 connection per origin and multiplexes +/// streams over it. Each blob upload (POST + streaming PUT) or pull +/// (streaming GET) holds an h2 stream open for the duration of the +/// transfer. Major container registries advertise +/// `SETTINGS_MAX_CONCURRENT_STREAMS` in the range 100-128 +/// (registry-1.docker.io: 128, ghcr.io: 100, cgr.dev: 100, +/// us-docker.pkg.dev: 100, public.ecr.aws: 128, quay.io: 128, +/// mcr.microsoft.com: 100, probed 2026-06-01). +/// +/// Capping concurrent streaming blobs at 64 leaves 36-64 stream slots +/// for short-lived metadata operations (manifest HEAD/GET, blob mount POST, +/// auth refresh) on every registry we've probed. Without this cap, default +/// `max_concurrent_transfers * BLOB_CONCURRENCY` of 50 * 6 = 300 streams +/// overshoots the budget by 3x and the connection stalls indefinitely. +const DEFAULT_STREAMING_BLOB_CONCURRENCY: usize = 64; + +/// Default per-target byte budget for in-flight buffered blob bodies. +/// +/// The fallback upload paths (GHCR single-PATCH, GAR monolithic PUT, ACR +/// chunked PATCH) all buffer the entire blob in memory because the target +/// registry cannot accept a streaming body. With +/// [`DEFAULT_STREAMING_BLOB_CONCURRENCY`] = 64 and no byte cap, 64 +/// concurrent 1 GB layers (common for CUDA/ML images) would hold ~64 GB +/// resident -- enough to OOM most hosts. +/// +/// A 512 MB budget gives the engine ~32x the steady-state working set of +/// a typical 16 MB layer image while keeping worst-case memory bounded. +/// Sized large enough that uploads of multiple modest layers do not +/// serialise on the semaphore, small enough that a single oversized layer +/// (e.g. 2 GB GPU runtime) cannot run alongside any other buffered upload. +const DEFAULT_BUFFERED_BLOB_BYTES: usize = 512 * 1024 * 1024; + +/// Fallback acquisition size when a fallback upload has no `known_size` +/// hint. Equals `blob::MAX_BUFFER_PREALLOC` (the same defensive cap +/// applied to the initial buffer allocation in `blob::buffer_stream`). +const BUFFERED_BLOB_BYTES_FALLBACK_RESERVE: u32 = 16 * 1024 * 1024; + /// Sentinel value stored in [`RegistryClient::rate_limit_remaining`] when no /// rate-limit header has been observed yet. const RATE_LIMIT_UNKNOWN: u64 = u64::MAX; @@ -25,12 +66,30 @@ pub struct RegistryClientBuilder { url: Url, auth: Option>, max_concurrent: usize, + /// Cap on concurrent long-lived blob streams (push PUTs, pull GETs). + /// Defaults to [`DEFAULT_STREAMING_BLOB_CONCURRENCY`]. + streaming_blob_concurrency: usize, + /// Byte budget for in-flight buffered blob bodies on fallback upload + /// paths (GHCR / GAR / ACR). Defaults to [`DEFAULT_BUFFERED_BLOB_BYTES`]. + buffered_blob_bytes: usize, /// Static DNS overrides applied to the internal reqwest client. /// Each entry maps a hostname to a fixed socket address, bypassing /// DNS resolution. Used by integration tests to route ECR-hostname /// requests at a local mock server without leaking `reqwest::Client` /// into the public API. dns_overrides: Vec<(String, std::net::SocketAddr)>, + /// When `true`, the internal reqwest client accepts any server + /// certificate without validation. Test-only -- see + /// [`RegistryClientBuilder::allow_invalid_certs`]. + accept_invalid_certs: bool, + /// When `true`, the internal reqwest client refuses to negotiate + /// HTTP/2 via ALPN. Test-only -- see + /// [`RegistryClientBuilder::force_http1`]. + force_http1: bool, + /// When `true`, the internal reqwest client enables HTTP/2 + /// adaptive-window sizing. Defaults to `true` -- see the comment in + /// [`RegistryClientBuilder::build`] for the motivation. + h2_adaptive_window: bool, } impl std::fmt::Debug for RegistryClientBuilder { @@ -52,7 +111,14 @@ impl RegistryClientBuilder { url, auth: None, max_concurrent: DEFAULT_MAX_CONCURRENT_REQUESTS, + streaming_blob_concurrency: DEFAULT_STREAMING_BLOB_CONCURRENCY, + buffered_blob_bytes: DEFAULT_BUFFERED_BLOB_BYTES, dns_overrides: Vec::new(), + accept_invalid_certs: false, + force_http1: false, + // Default-on: see note in `build`. The builder method below + // can be used to disable for A/B testing. + h2_adaptive_window: true, } } @@ -63,8 +129,106 @@ impl RegistryClientBuilder { } /// Set the maximum number of concurrent requests. + /// + /// Values of `0` are clamped to `1` so the resulting semaphore can + /// always make forward progress. Pass at least `1` to opt out of the + /// clamp. pub fn max_concurrent(mut self, n: usize) -> Self { - self.max_concurrent = n; + self.max_concurrent = n.max(1); + self + } + + /// Set the cap on concurrent long-lived blob streams (push / pull). + /// + /// Each blob upload PUT and blob pull GET holds an HTTP/2 stream open + /// for the duration of the transfer. Reqwest/hyper uses one h2 + /// connection per origin, so this cap must stay under the server's + /// advertised `SETTINGS_MAX_CONCURRENT_STREAMS` (typically 100-128 on + /// major registries) minus headroom for metadata operations + /// (manifest HEAD/GET, blob mount, auth refresh). + /// + /// Default: [`DEFAULT_STREAMING_BLOB_CONCURRENCY`] (64). + pub fn streaming_blob_concurrency(mut self, n: usize) -> Self { + self.streaming_blob_concurrency = n.max(1); + self + } + + /// Set the byte budget for in-flight buffered blob bodies. + /// + /// Caps total resident memory across the fallback upload paths + /// (GHCR / GAR / ACR) that must buffer a blob before sending. Each + /// fallback upload acquires permits equal to its declared blob size + /// (capped at the budget) and releases them when the upload + /// completes; concurrent uploads that would exceed the budget queue. + /// + /// The streaming PUT default path does not buffer and is unaffected. + /// + /// `n` is clamped to a minimum of 1 (so the semaphore can always + /// make forward progress) but is otherwise honored verbatim. Values + /// smaller than [`BUFFERED_BLOB_BYTES_FALLBACK_RESERVE`] are valid: + /// the permit acquisition logic caps requests at the configured + /// budget, so unknown-size uploads (which would otherwise request + /// the full fallback reserve) still admit on a small-budget client. + /// + /// Default: [`DEFAULT_BUFFERED_BLOB_BYTES`] (512 MB). + pub fn buffered_blob_bytes(mut self, n: usize) -> Self { + self.buffered_blob_bytes = n.max(1); + self + } + + // ----------------------------------------------------------------- + // Diagnostic / perf-harness test hooks + // + // The three builder methods below (`allow_invalid_certs`, + // `force_http1`, `http2_adaptive_window`) are deliberately kept on + // the public builder surface even though no production code path + // reaches them. They exist for in-repo perf and A/B harnesses -- + // primarily `crates/ocync-sync/tests/perf_profile.rs`, which: + // - terminates TLS at a `testcontainers` `registry:2` instance + // with a generated self-signed cert (needs `allow_invalid_certs`), + // - compares HTTP/2 vs HTTP/1.1 throughput by toggling + // `force_http1` between runs, + // - and A/Bs the HTTP/2 adaptive flow-control window via + // `http2_adaptive_window(false)`. + // + // Each is `#[doc(hidden)]` so it never appears in published rustdoc, + // and the production CLI deliberately does NOT expose any flag or + // env var that toggles them. They are **intentionally retained** so + // future perf investigations can reproduce the experiments without + // re-introducing the plumbing from scratch. DO NOT delete as + // "unused" -- they are referenced from the perf harness and are + // load-bearing for repeatability. + // ----------------------------------------------------------------- + + /// Accept any server certificate without validation. + /// + /// Test-only escape hatch for the perf profile harness, which + /// terminates TLS at a local `registry:2` instance with a generated + /// self-signed cert. Hidden from public docs; never use in + /// production code. + #[doc(hidden)] + pub fn allow_invalid_certs(mut self, allow: bool) -> Self { + self.accept_invalid_certs = allow; + self + } + + /// Refuse to negotiate HTTP/2 via ALPN. Test-only escape hatch for the + /// perf profile harness, used to isolate HTTP/2-multiplexing stalls + /// from the rest of the streaming path. Hidden from public docs; + /// never use in production code. + #[doc(hidden)] + pub fn force_http1(mut self, force: bool) -> Self { + self.force_http1 = force; + self + } + + /// Configure HTTP/2 adaptive-window sizing via reqwest's + /// `http2_adaptive_window`. Defaults to enabled -- see the note in + /// `build` for motivation. This setter is for A/B testing the + /// disabled path; production callers should not need to touch it. + #[doc(hidden)] + pub fn http2_adaptive_window(mut self, enable: bool) -> Self { + self.h2_adaptive_window = enable; self } @@ -87,6 +251,26 @@ impl RegistryClientBuilder { for (host, addr) in &self.dns_overrides { http_builder = http_builder.resolve(host, *addr); } + if self.accept_invalid_certs { + http_builder = http_builder.danger_accept_invalid_certs(true); + } + if self.force_http1 { + http_builder = http_builder.http1_only(); + } + // Enable HTTP/2 adaptive flow-control window sizing. Hyper's + // default starts each stream at a 64KB receive window and grows + // it only after `WINDOW_UPDATE` frames; at high stream + // concurrency (e.g. `max_concurrent_transfers` of 50 streaming + // large blobs) this starves throughput and against some + // registries surfaces as a stall. Adaptive window sizes the + // window dynamically based on observed throughput. Off by + // default in reqwest; we want it on by default for the registry + // client. + // + // Test hook: `http2_adaptive_window(false)` disables for A/B. + if self.h2_adaptive_window { + http_builder = http_builder.http2_adaptive_window(true); + } let http = http_builder.build()?; let aimd = AimdController::new( @@ -98,6 +282,9 @@ impl RegistryClientBuilder { http, auth: self.auth, aimd, + streaming_blob_sem: Arc::new(Semaphore::new(self.streaming_blob_concurrency)), + buffered_blob_bytes_sem: Arc::new(Semaphore::new(self.buffered_blob_bytes)), + buffered_blob_bytes_budget: self.buffered_blob_bytes, rate_limit_remaining: AtomicU64::new(RATE_LIMIT_UNKNOWN), }) } @@ -112,6 +299,27 @@ pub struct RegistryClient { pub(crate) http: reqwest::Client, pub(crate) auth: Option>, pub(crate) aimd: AimdController, + /// Cap on concurrent long-lived blob streams (push / pull). Acquired + /// at the start of [`Self::blob_push_stream`] and [`Self::blob_pull`] + /// to keep the per-connection HTTP/2 stream usage below the server's + /// advertised `SETTINGS_MAX_CONCURRENT_STREAMS`. See the constant + /// `DEFAULT_STREAMING_BLOB_CONCURRENCY` for the rationale and probed + /// numbers per registry. + pub(crate) streaming_blob_sem: Arc, + /// Byte-budget semaphore for in-flight buffered blob bodies. One + /// permit == one byte. Acquired by the GHCR / GAR / ACR fallback + /// upload paths (which buffer the entire blob) before + /// `buffer_stream`, released when the buffered upload completes. The + /// total permit count caps cross-call resident bytes regardless of + /// per-call concurrency. + pub(crate) buffered_blob_bytes_sem: Arc, + /// Configured byte budget passed to `buffered_blob_bytes_sem`. Held + /// so [`Self::acquire_buffered_bytes_permit`] can cap a single + /// oversized blob at the entire budget instead of permanently + /// blocking; without this cap, a blob larger than the configured + /// budget would request more permits than the semaphore can ever + /// hold. + pub(crate) buffered_blob_bytes_budget: usize, /// Last observed rate-limit remaining value from response headers. /// /// Updated atomically on every response that carries a `ratelimit-remaining` @@ -121,6 +329,85 @@ pub struct RegistryClient { rate_limit_remaining: AtomicU64, } +impl RegistryClient { + /// Acquire a permit gating concurrent long-lived blob streams. + /// + /// Held by blob push / pull paths for the duration of the streaming + /// HTTP/2 body. Released automatically when the returned permit is + /// dropped. See [`DEFAULT_STREAMING_BLOB_CONCURRENCY`] for the + /// rationale. + pub(crate) async fn acquire_streaming_blob_permit(&self) -> OwnedSemaphorePermit { + // Sema is never closed for the lifetime of the client; unwrap is + // sound. + Arc::clone(&self.streaming_blob_sem) + .acquire_owned() + .await + .expect("streaming blob semaphore closed") + } + + /// Acquire both the per-target count permit and the byte-budget + /// permit for a fallback (buffered) blob upload, in the right order. + /// + /// Returns `(bytes_permit, stream_permit)`. Both are held by the + /// caller for the duration of the buffered upload; Rust tuples drop + /// fields in declaration order, so `bytes_permit` is released + /// FIRST on scope exit. Releasing the contested byte budget ahead + /// of the count cap is the right order: a fallback that completes + /// freeing 6 MB lets queued buffered uploads admit immediately, + /// while the stream-count cap (default 64) is rarely the bottleneck + /// at typical workloads. See the per-cap docs on + /// [`Self::acquire_streaming_blob_permit`] and + /// [`Self::acquire_buffered_bytes_permit`] for the semantics of + /// each. + pub(crate) async fn acquire_buffered_upload_permits( + &self, + known_size: Option, + ) -> (OwnedSemaphorePermit, OwnedSemaphorePermit) { + let stream_permit = self.acquire_streaming_blob_permit().await; + let bytes_permit = self.acquire_buffered_bytes_permit(known_size).await; + (bytes_permit, stream_permit) + } + + /// Acquire a byte-budget permit for a fallback (buffered) blob upload. + /// + /// `expected` is the manifest-declared blob size when known, or + /// `None` when the upload is driven by an opaque stream (rare in + /// practice; manifests always declare layer size). Acquisition rules: + /// + /// - If `expected` is `Some(n)`: acquire `min(n, budget, u32::MAX)` + /// permits. The cap at `budget` ensures a single oversized blob + /// never requests more permits than the semaphore can hold + /// (which would deadlock the entire fallback path). When this + /// cap fires, the oversized blob occupies the entire budget and + /// other fallback uploads queue until it completes -- correct + /// semantics, even though the actual memory used exceeds the + /// budget for that one upload. + /// - If `expected` is `None`: acquire + /// [`BUFFERED_BLOB_BYTES_FALLBACK_RESERVE`] permits as a defensive + /// reserve. This matches `blob::MAX_BUFFER_PREALLOC` so we + /// conservatively budget the same up-front allocation the buffer + /// itself reserves; actual buffered bytes may exceed this if the + /// blob is larger than the reserve, but the count cap + /// (`streaming_blob_sem`) still bounds total in-flight count. + pub(crate) async fn acquire_buffered_bytes_permit( + &self, + expected: Option, + ) -> OwnedSemaphorePermit { + let request = expected.unwrap_or(BUFFERED_BLOB_BYTES_FALLBACK_RESERVE as u64); + let budget = self.buffered_blob_bytes_budget as u64; + // Cap at the configured budget so a single oversized blob never + // requests more permits than the semaphore can hold (which + // would deadlock the fallback path). Cap at u32::MAX because + // `acquire_many_owned` takes a u32. `.max(1)` guarantees forward + // progress when both request and budget round to zero. + let permits: u32 = request.min(budget).min(u32::MAX as u64).max(1) as u32; + Arc::clone(&self.buffered_blob_bytes_sem) + .acquire_many_owned(permits) + .await + .expect("buffered blob bytes semaphore closed") + } +} + impl std::fmt::Debug for RegistryClient { fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { let remaining = self.rate_limit_remaining.load(Ordering::Relaxed); diff --git a/crates/ocync-distribution/src/ecr.rs b/crates/ocync-distribution/src/ecr.rs index 48079706..8bb985df 100644 --- a/crates/ocync-distribution/src/ecr.rs +++ b/crates/ocync-distribution/src/ecr.rs @@ -13,6 +13,7 @@ use std::collections::HashSet; use std::future::Future; use std::pin::Pin; +use std::time::Duration; use aws_config::BehaviorVersion; use aws_sdk_ecr::types::LayerAvailability; @@ -89,12 +90,26 @@ const MAX_DIGESTS_PER_BATCH: usize = 100; /// Boxed future returned by [`BatchBlobChecker::check_blob_existence`]. type CheckFuture<'a> = Pin, Error>> + 'a>>; +/// Boxed future returned by [`BatchBlobChecker::wait_for_blobs_available`]. +type WaitFuture<'a> = Pin> + 'a>>; + +/// Initial poll interval when waiting for an ECR consistency view to +/// converge. Doubles on each subsequent poll up to [`MAX_POLL_INTERVAL`]. +const INITIAL_POLL_INTERVAL: Duration = Duration::from_millis(200); + +/// Upper bound on the poll interval used while waiting on the ECR +/// consistency view. Keeps the worst-case stale period bounded so a +/// blob that becomes available right after a long sleep is picked up +/// quickly. +const MAX_POLL_INTERVAL: Duration = Duration::from_secs(5); + /// Async trait for batch blob existence checking. /// /// Used by the sync engine to efficiently determine which blobs already exist -/// at an ECR target registry before initiating transfers. Implementations are -/// intended to be held as `Rc` on a single-threaded -/// tokio runtime, so no `Send` or `Sync` bounds are required. +/// at an ECR target registry before initiating transfers, and to gate manifest +/// commits on blob visibility (see [`Self::wait_for_blobs_available`]). +/// Implementations are intended to be held as `Rc` on a +/// single-threaded tokio runtime, so no `Send` or `Sync` bounds are required. pub trait BatchBlobChecker { /// Check which blobs exist in the given repository. /// @@ -106,6 +121,112 @@ pub trait BatchBlobChecker { repo: &'a RepositoryName, digests: &'a [Digest], ) -> CheckFuture<'a>; + + /// Wait until all `digests` are reported available by the target. + /// + /// Polls [`Self::check_blob_existence`] with exponential backoff + /// (starting at [`INITIAL_POLL_INTERVAL`], doubling up to + /// [`MAX_POLL_INTERVAL`]) until either every digest is reported + /// available or `deadline` is reached. + /// + /// # Why this exists + /// + /// ECR's manifest-validation index is eventually consistent with + /// blob upload state. A blob `PUT /v2/.../blobs/...` returns + /// 201 Created while the validator's view can lag for hundreds of + /// milliseconds to several seconds at high concurrency. A manifest + /// `PUT` issued during this window fails with HTTP 404 carrying + /// `BLOB_UPLOAD_UNKNOWN`. Calling this method before manifest push + /// gates the commit on the consistency view directly rather than + /// relying on retry budget alone. + /// + /// # Error semantics + /// + /// Returns `Ok(())` once every digest is available. Returns an + /// `Err` with [`Error::EcrApi`] carrying the count of still-missing + /// digests when `deadline` expires; the caller is expected to log + /// and proceed with the standard retry path. Individual poll + /// failures (transient ECR API errors) are logged as warnings and + /// treated as "no progress this iteration" -- the loop continues + /// until `deadline`. + /// + /// # Default implementation + /// + /// The default polls `check_blob_existence` and returns early on + /// full availability. Tests may override to avoid real sleeping. + fn wait_for_blobs_available<'a>( + &'a self, + repo: &'a RepositoryName, + digests: &'a [Digest], + deadline: Duration, + ) -> WaitFuture<'a> { + Box::pin(default_wait_for_blobs_available( + self, repo, digests, deadline, + )) + } +} + +/// Default polling loop shared by every [`BatchBlobChecker`] impl. +/// +/// Free function so the trait remains object-safe under `Rc` +/// while still factoring out the polling logic. +async fn default_wait_for_blobs_available( + checker: &T, + repo: &RepositoryName, + digests: &[Digest], + deadline: Duration, +) -> Result<(), Error> +where + T: BatchBlobChecker + ?Sized, +{ + if digests.is_empty() { + return Ok(()); + } + + let start = tokio::time::Instant::now(); + let mut interval = INITIAL_POLL_INTERVAL; + // Track the surface of "still missing" so we only re-query the + // shrinking remainder, not the full input on every poll. ECR + // BatchCheckLayerAvailability counts toward an account-level quota + // (10 TPS), so trimming the request size reduces blast radius when + // the wait spans several seconds. + let mut remaining: Vec = digests.to_vec(); + + loop { + match checker.check_blob_existence(repo, &remaining).await { + Ok(available) => { + remaining.retain(|d| !available.contains(d)); + if remaining.is_empty() { + return Ok(()); + } + } + Err(e) => { + warn!( + repo = %repo, + error = %e, + "BatchCheckLayerAvailability poll failed; retrying until deadline" + ); + } + } + + let elapsed = start.elapsed(); + if elapsed >= deadline { + return Err(Error::EcrApi { + reason: format!( + "BatchCheckLayerAvailability timed out for {repo} after {:?} with {} digest(s) still missing", + elapsed, + remaining.len() + ), + }); + } + + // Cap sleep so we never overshoot the deadline by more than one + // interval, which would be observable as a wait noticeably + // longer than requested. + let sleep_for = interval.min(deadline.saturating_sub(elapsed)); + tokio::time::sleep(sleep_for).await; + interval = interval.saturating_mul(2).min(MAX_POLL_INTERVAL); + } } /// Abstraction over ECR batch API calls for testability. @@ -754,6 +875,164 @@ mod tests { fn _assert_object_safe(_: std::rc::Rc) {} } + // --- wait_for_blobs_available tests --- + + /// Empty input must short-circuit without any API call. + #[tokio::test] + async fn wait_empty_digests_short_circuits() { + let counts = CallCounts::default(); + let mock = MockEcrBatchApi::new("repo", counts.clone()); + let checker = BatchChecker::with_api(mock); + let result = checker + .wait_for_blobs_available( + &RepositoryName::new("repo").unwrap(), + &[], + Duration::from_secs(1), + ) + .await; + assert!(result.is_ok()); + assert_eq!( + counts.check.load(Ordering::Relaxed), + 0, + "empty input must NOT trigger an ECR API call" + ); + } + + /// First poll reports all blobs available -- single call, no polling. + #[tokio::test] + async fn wait_all_available_on_first_poll() { + let d1 = test_digest(1); + let d2 = test_digest(2); + let counts = CallCounts::default(); + let mock = MockEcrBatchApi::new("repo", counts.clone()).with_check_responses(vec![Ok( + BatchCheckResponse { + layers: vec![(d1.to_string(), true), (d2.to_string(), true)], + failures: vec![], + }, + )]); + let checker = BatchChecker::with_api(mock); + let result = checker + .wait_for_blobs_available( + &RepositoryName::new("repo").unwrap(), + &[d1, d2], + Duration::from_secs(10), + ) + .await; + assert!(result.is_ok()); + assert_eq!( + counts.check.load(Ordering::Relaxed), + 1, + "all-available on first poll must NOT loop" + ); + } + + /// First poll reports partial availability, second reports the rest. + /// The second call must request only the previously-missing digest + /// (request trimming optimisation -- otherwise we re-query the + /// already-available digest on every poll). + #[tokio::test(start_paused = true)] + async fn wait_polls_until_all_available_and_trims_requests() { + let d1 = test_digest(1); + let d2 = test_digest(2); + + let counts = CallCounts::default(); + let mock = MockEcrBatchApi::new("repo", counts.clone()).with_check_responses(vec![ + // Poll 1: only d1 is available. + Ok(BatchCheckResponse { + layers: vec![(d1.to_string(), true), (d2.to_string(), false)], + failures: vec![], + }), + // Poll 2: d2 now available. + // The trimming optimisation means we expect only d2 to be + // requested; if d1 were re-requested, the mock would still + // accept it but the count test below pins the intent. + Ok(BatchCheckResponse { + layers: vec![(d2.to_string(), true)], + failures: vec![], + }), + ]); + let checker = BatchChecker::with_api(mock); + let result = checker + .wait_for_blobs_available( + &RepositoryName::new("repo").unwrap(), + &[d1.clone(), d2.clone()], + Duration::from_secs(10), + ) + .await; + assert!(result.is_ok(), "{:?}", result.err()); + assert_eq!(counts.check.load(Ordering::Relaxed), 2); + } + + /// Polls keep reporting blob missing until deadline expires. + /// Returns `Err(EcrApi)` carrying a useful diagnostic. + #[tokio::test(start_paused = true)] + async fn wait_times_out_when_blob_never_appears() { + let d1 = test_digest(1); + // Mock returns "missing" forever. Use a long response queue so + // the test isn't flaky on number of polls. + let counts = CallCounts::default(); + let responses: Vec> = (0..50) + .map(|_| { + Ok(BatchCheckResponse { + layers: vec![(d1.to_string(), false)], + failures: vec![], + }) + }) + .collect(); + let mock = MockEcrBatchApi::new("repo", counts.clone()).with_check_responses(responses); + let checker = BatchChecker::with_api(mock); + + let result = checker + .wait_for_blobs_available( + &RepositoryName::new("repo").unwrap(), + &[d1], + Duration::from_secs(2), + ) + .await; + match result { + Err(Error::EcrApi { reason }) => { + assert!( + reason.contains("timed out") && reason.contains("1 digest"), + "unexpected timeout reason: {reason}" + ); + } + other => panic!("expected EcrApi timeout, got {other:?}"), + } + } + + /// Poll failure does NOT abort the wait -- the loop continues until + /// the deadline expires. + #[tokio::test(start_paused = true)] + async fn wait_continues_through_transient_api_errors() { + let d1 = test_digest(1); + let counts = CallCounts::default(); + let mock = MockEcrBatchApi::new("repo", counts.clone()).with_check_responses(vec![ + // Transient API failure on first poll. + Err(Error::EcrApi { + reason: "throttled".into(), + }), + // Second poll succeeds with blob now available. + Ok(BatchCheckResponse { + layers: vec![(d1.to_string(), true)], + failures: vec![], + }), + ]); + let checker = BatchChecker::with_api(mock); + let result = checker + .wait_for_blobs_available( + &RepositoryName::new("repo").unwrap(), + &[d1], + Duration::from_secs(10), + ) + .await; + assert!( + result.is_ok(), + "transient errors must not abort the wait: {:?}", + result.err() + ); + assert_eq!(counts.check.load(Ordering::Relaxed), 2); + } + #[tokio::test] async fn batch_checker_from_hostname_accepts_named_profile() { // Pins the BatchChecker / EcrAuth symmetry: both must thread the same diff --git a/crates/ocync-sync/CLAUDE.md b/crates/ocync-sync/CLAUDE.md index 01f137bf..59eef08a 100644 --- a/crates/ocync-sync/CLAUDE.md +++ b/crates/ocync-sync/CLAUDE.md @@ -28,6 +28,18 @@ Sync orchestration engine - pipelined discovery/execution, leader-follower blob 1. Per-blob `Notify` via `ClaimAction::Wait`: followers wait for leader's blob upload. 2. Per-repo `watch` via `repo_committed_watch`: followers wait for leader's manifest commit before mounting. ECR requires a committed manifest in the source repo for mount to succeed (201); without this wait, mounts hit Tier 3 and get 202 (Not Fulfilled). Uses `watch` (not `Notify`) because committed status is boolean state, not an event -- `watch` retains the last value so late subscribers always see it. +## Manifest commit gating (ECR consistency window) + +ECR's manifest-validation index is eventually consistent with blob upload state. A blob `PUT-201` can land before the validator's view catches up; a manifest `PUT` issued during that window fails with HTTP 404 carrying `BLOB_UPLOAD_UNKNOWN`. The retry path in `with_retry` catches this code as a defence-in-depth, but the structural fix is to gate the manifest commit on the authoritative blob-visibility API. + +`push_manifests` calls `BatchBlobChecker::wait_for_blobs_available` before issuing the manifest `PUT` when both: +1. The target has a configured `batch_checker` (production: only ECR), AND +2. `RetryConfig::manifest_commit_wait` is non-zero. + +The wait polls `BatchCheckLayerAvailability` with exponential backoff (`200ms -> 400ms -> ... -> 5s`) until every referenced layer is visible, or `manifest_commit_wait` is reached. On timeout the engine falls through to the standard manifest PUT and any residual `BLOB_UPLOAD_UNKNOWN` is caught by `with_retry`. + +Production default: 30s. Tests default to `Duration::ZERO` via `fast_retry()` so the wait does not fire (mock checkers don't reflect engine uploads). Tests that need to exercise the wait construct their own `RetryConfig` with `manifest_commit_wait > 0`. + ## Synchronization contracts (critical) - Per-blob `Notify::notify_waiters()` does NOT store permits. Every code path that transitions a blob out of `InProgress` MUST call `notify_blob`. Same applies to `BlobStage::notify_staged` / `notify_failed` for source-pull dedup. Missing notify = deadlock for concurrent waiters. diff --git a/crates/ocync-sync/Cargo.toml b/crates/ocync-sync/Cargo.toml index bc34f12f..6ef5b462 100644 --- a/crates/ocync-sync/Cargo.toml +++ b/crates/ocync-sync/Cargo.toml @@ -17,6 +17,7 @@ workspace = true [dependencies] bytes.workspace = true crc32fast = { version = "1", default-features = false } +fastrand.workspace = true futures-util.workspace = true globset = { version = "0.4", default-features = false } http.workspace = true @@ -32,7 +33,9 @@ tracing.workspace = true uuid.workspace = true [dev-dependencies] +rcgen = { version = "0.14", default-features = false, features = ["aws_lc_rs", "crypto", "pem"] } tempfile = { version = "3", default-features = false } +testcontainers = { version = "0.27", default-features = false } tokio = { workspace = true, features = ["macros", "rt", "test-util"] } url.workspace = true wiremock = { version = "0.6", default-features = false } diff --git a/crates/ocync-sync/src/engine.rs b/crates/ocync-sync/src/engine.rs index 3e604605..837845f2 100644 --- a/crates/ocync-sync/src/engine.rs +++ b/crates/ocync-sync/src/engine.rs @@ -717,7 +717,20 @@ struct PromoteContext<'a> { } /// Default cap for concurrent image transfers (Level 1: global image semaphore). -pub const DEFAULT_MAX_CONCURRENT_TRANSFERS: usize = 50; +/// +/// Derivation: per-target `RegistryClient::streaming_blob_sem` caps long-lived +/// HTTP/2 blob streams at 64 (the lowest-common-denominator under registry +/// `SETTINGS_MAX_CONCURRENT_STREAMS` of 100-128 minus metadata headroom). +/// With [`BLOB_CONCURRENCY`] = 6 per image, an image-level cap of 10 gives +/// 10 * 6 = 60 streaming attempts at saturation -- fits the 64-stream budget +/// without thrashing on the semaphore. Higher image counts only help when +/// most images skip / mount (no streaming), which is a workload-specific +/// override the user can set in config. +/// +/// Prior value of 50 was picked before we measured registry stream limits; +/// at 50 * 6 = 300 streams it deadlocks HTTP/2 against every registry we +/// probed (see `DEFAULT_STREAMING_BLOB_CONCURRENCY` in `ocync-distribution`). +pub const DEFAULT_MAX_CONCURRENT_TRANSFERS: usize = 10; /// Maximum concurrent blob transfers within a single image. /// @@ -726,9 +739,8 @@ pub const DEFAULT_MAX_CONCURRENT_TRANSFERS: usize = 50; /// simultaneous blob uploads/downloads per image while allowing the global /// semaphore to independently control total in-flight image tasks. /// -/// Matches skopeo's default (6). Higher values risk approaching ECR's -/// `InitiateLayerUpload` limit (100 TPS shared across all images). -/// Candidate for `SyncEngine` builder configuration if workloads need tuning. +/// Matches skopeo's default (6). The cross-image streaming budget is bounded +/// independently by `RegistryClient::streaming_blob_sem` (default 64). const BLOB_CONCURRENCY: usize = 6; /// Default shutdown drain deadline in seconds. @@ -1687,6 +1699,7 @@ async fn execute_item( &item.target.repo, &item.target.tag, &item.source_data, + item.batch_checker.as_deref(), ) .await { @@ -1708,6 +1721,7 @@ async fn execute_item( &item.artifacts_config, retry, referrers_cache, + item.batch_checker.as_deref(), ) .await { @@ -2599,13 +2613,69 @@ async fn push_staged_blob(io: &BlobIoContext<'_>, digest: &Digest) -> Result<(), } /// Push all manifests (children for indexes, then top-level) to one target. +/// +/// When `batch_checker` is `Some` AND `retry.manifest_commit_wait` is +/// non-zero, calls +/// [`BatchBlobChecker::wait_for_blobs_available`] on every blob the +/// manifest tree references BEFORE issuing any manifest `PUT`. This +/// eliminates the `BLOB_UPLOAD_UNKNOWN` race for targets whose consistency +/// view lags blob upload (ECR specifically) -- it gates the commit on +/// the authoritative blob-visibility API rather than relying on retry +/// budget alone. Targets without a batch checker (everything except +/// ECR) skip the wait entirely; setting `manifest_commit_wait` to +/// `Duration::ZERO` also skips the wait. async fn push_manifests( retry: &RetryConfig, target_client: &RegistryClient, target_repo: &RepositoryName, target_tag: &str, source_data: &PulledManifest, + batch_checker: Option<&dyn BatchBlobChecker>, ) -> Result<(), crate::Error> { + if let Some(checker) = batch_checker + && !retry.manifest_commit_wait.is_zero() + { + // Extract every leaf-blob digest the manifest references + // (config + layers across all children for indexes). Collect + // through a HashSet to deduplicate (one clone per digest); + // BatchCheckLayerAvailability does not care about order. + let digests: Vec = blobs_from_manifest(source_data) + .iter() + .map(|d| d.digest.clone()) + .collect::>() + .into_iter() + .collect(); + + if !digests.is_empty() { + match checker + .wait_for_blobs_available(target_repo, &digests, retry.manifest_commit_wait) + .await + { + Ok(()) => { + debug!( + target_repo = %target_repo, + blob_count = digests.len(), + "BatchCheckLayerAvailability confirms all blobs visible; proceeding with manifest push" + ); + } + Err(e) => { + // Falling through to the retry path is the design + // choice: a stuck consistency view is a real + // condition we want surfaced via the same retry + // accounting as any other transient failure, not a + // hard error here. with_retry will catch any + // residual BLOB_UPLOAD_UNKNOWN below. + warn!( + target_repo = %target_repo, + blob_count = digests.len(), + error = %e, + "BatchCheckLayerAvailability wait did not converge before deadline; proceeding (retry path will catch any BLOB_UPLOAD_UNKNOWN)" + ); + } + } + } + } + // For index manifests, push all children concurrently by digest. // The target registry's AIMD controller gates concurrency naturally. // Uses join_all (not try_join_all) so that a transient failure on one @@ -2666,6 +2736,19 @@ async fn push_manifests( /// For each matching artifact: push blobs, then push manifest (preserving /// the `subject` reference to the parent). /// +/// When `batch_checker` is present (production: ECR targets), each artifact +/// runs the same two-phase batch-check protocol as `transfer_image_blobs` +/// and `push_manifests`: +/// - Pre-blob: `check_blob_existence` over all artifact blob digests to +/// skip per-blob HEAD round-trips. +/// - Pre-manifest: `wait_for_blobs_available` to gate the artifact +/// manifest PUT on the same ECR consistency view that protects the +/// main image manifest PUT. +/// +/// Within an artifact, blobs are pulled+pushed concurrently capped at +/// [`BLOB_CONCURRENCY`] so multi-blob artifacts (config + layers) don't +/// serialise unnecessarily. +/// /// Returns `Ok(true)` when artifact discovery was skipped due to a transient /// error (the image synced but artifacts may be missing at the target). #[allow(clippy::too_many_arguments)] @@ -2678,6 +2761,7 @@ async fn discover_and_sync_artifacts( artifacts_config: &ResolvedArtifacts, retry: &RetryConfig, referrers_cache: &ReferrersCache, + batch_checker: Option<&dyn BatchBlobChecker>, ) -> Result { if !artifacts_config.enabled { return Ok(false); @@ -2766,38 +2850,126 @@ async fn discover_and_sync_artifacts( })?; // Push artifact blobs. + let mut blob_digests: Vec = Vec::new(); if let ManifestKind::Image(ref manifest) = artifact_pull.manifest { - let blobs = collect_image_blobs(manifest); - for blob in blobs { - // HEAD check target first. - let exists = target_client - .blob_exists(target_repo, &blob.digest) + let blobs: Vec<&Descriptor> = collect_image_blobs(manifest); + blob_digests = blobs.iter().map(|b| b.digest.clone()).collect(); + + // Batch-check pre-population (mirrors `transfer_image_blobs`). + // When a batch checker is configured (production: ECR), + // skip per-blob HEAD calls for blobs the batch API already + // reports present at the target. + let already_present: HashSet = match batch_checker { + Some(checker) => match checker + .check_blob_existence(target_repo, &blob_digests) .await - .unwrap_or(None); - - if exists.is_some() { - continue; - } + { + Ok(existing) => existing, + Err(e) => { + warn!( + target_repo = %target_repo, + artifact_digest = %artifact_digest_str, + error = %e, + "artifact batch check failed, falling back to per-blob HEAD" + ); + HashSet::new() + } + }, + None => HashSet::new(), + }; - // Pull from source, push to target. + // Pull+push missing blobs concurrently, capped by BLOB_CONCURRENCY. + let blob_sem = Semaphore::new(BLOB_CONCURRENCY); + let mut blob_futures = FuturesUnordered::new(); + for blob in blobs { let blob_digest = blob.digest.clone(); let blob_size = blob.size; - let stream = with_retry(retry, "artifact blob pull", || { - source_client.blob_pull(source_repo, &blob_digest) - }) - .await - .map_err(|e| crate::Error::ArtifactSync { - reference: artifact_digest_str.clone(), - reason: format!("blob pull failed for {blob_digest}: {e}"), - })?; + let blob_already_present = already_present.contains(&blob.digest); + let sem = &blob_sem; + let artifact_digest_str = artifact_digest_str.clone(); + blob_futures.push(async move { + // For batch-already-present blobs, skip the HEAD and + // any pull/push: the batch API already confirmed + // visibility. Otherwise do a HEAD check (covers + // non-batch-checker targets) and only pull+push on + // 404. + if !blob_already_present { + let exists = target_client + .blob_exists(target_repo, &blob_digest) + .await + .unwrap_or(None); + if exists.is_none() { + let _permit = sem.acquire().await.expect("sem closed"); + let stream = with_retry(retry, "artifact blob pull", || { + source_client.blob_pull(source_repo, &blob_digest) + }) + .await + .map_err(|e| { + crate::Error::ArtifactSync { + reference: artifact_digest_str.clone(), + reason: format!("blob pull failed for {blob_digest}: {e}"), + } + })?; + target_client + .blob_push_stream( + target_repo, + &blob_digest, + Some(blob_size), + stream, + ) + .await + .map_err(|e| crate::Error::ArtifactSync { + reference: artifact_digest_str.clone(), + reason: format!("blob push failed for {blob_digest}: {e}"), + })?; + } + } + Ok::<(), crate::Error>(()) + }); + } + // Fail-fast on first error: dropping `blob_futures` cancels + // in-flight blob HTTP. This is deliberately asymmetric with + // `transfer_image_blobs` (which uses a `cancel: Cell` + // flag so all blobs finish for stats / notify-failed + // bookkeeping). Artifacts have no per-blob stats or + // cross-image dedup to maintain, and the caller at + // `process_one_target` already treats artifact-sync errors + // as a non-fatal warning -- so cancelling the rest is + // cheaper than draining them after we've already decided + // to fail the artifact. + while let Some(result) = blob_futures.next().await { + result?; + } + } - target_client - .blob_push_stream(target_repo, &blob_digest, Some(blob_size), stream) - .await - .map_err(|e| crate::Error::ArtifactSync { - reference: artifact_digest_str.clone(), - reason: format!("blob push failed for {blob_digest}: {e}"), - })?; + // Gate the artifact manifest PUT on the target's blob-visibility + // view, mirroring `push_manifests`. Same rationale: ECR's + // manifest validator can lag blob PUT-201s by hundreds of ms. + if let Some(checker) = batch_checker + && !retry.manifest_commit_wait.is_zero() + && !blob_digests.is_empty() + { + match checker + .wait_for_blobs_available(target_repo, &blob_digests, retry.manifest_commit_wait) + .await + { + Ok(()) => { + debug!( + target_repo = %target_repo, + artifact_digest = %artifact_digest_str, + blob_count = blob_digests.len(), + "BatchCheckLayerAvailability confirms artifact blobs visible; proceeding with manifest push" + ); + } + Err(e) => { + warn!( + target_repo = %target_repo, + artifact_digest = %artifact_digest_str, + blob_count = blob_digests.len(), + error = %e, + "BatchCheckLayerAvailability wait did not converge before deadline; proceeding (retry path will catch any BLOB_UPLOAD_UNKNOWN)" + ); + } } } @@ -2949,8 +3121,11 @@ fn file_read_stream( /// Retry an async operation with exponential backoff on transient errors. /// -/// Calls `f()` in a loop. Retries on HTTP 408/429/5xx status codes and on -/// transport-level errors (connection refused, DNS failure, request timeout). +/// Calls `f()` in a loop. Retries on HTTP 408/429/5xx, transport-level +/// errors (connection refused, DNS failure, request timeout), and the +/// ECR-specific 404 `BLOB_UPLOAD_UNKNOWN` body marker via +/// [`retry::is_blob_upload_unknown`] (ECR returns this code on manifest +/// push when blob upload PUT-201s haven't fully propagated). /// Waits with jittered exponential backoff up to `config.max_retries` times. /// Returns the first `Ok` or the final `Err`. async fn with_retry( @@ -2969,6 +3144,7 @@ where Err(e) => { let retryable = if let Some(status) = e.status_code() { retry::should_retry(status, attempt, config.max_retries) + || (attempt < config.max_retries && retry::is_blob_upload_unknown(&e)) } else { // Transport-level errors (connection refused, DNS failure, // request timeout) are retryable when attempts remain. diff --git a/crates/ocync-sync/src/retry.rs b/crates/ocync-sync/src/retry.rs index 9fa4124e..9f011e2f 100644 --- a/crates/ocync-sync/src/retry.rs +++ b/crates/ocync-sync/src/retry.rs @@ -1,7 +1,5 @@ //! Retry configuration and backoff logic for transient failures. -use std::collections::hash_map::RandomState; -use std::hash::{BuildHasher, Hasher}; use std::time::Duration; use http::StatusCode; @@ -17,6 +15,16 @@ pub struct RetryConfig { pub max_backoff: Duration, /// Multiplier applied to backoff on each successive attempt. pub backoff_multiplier: u32, + /// Maximum time the engine waits for the target's blob-availability + /// view to converge before issuing a manifest `PUT`, when a + /// [`BatchBlobChecker`](ocync_distribution::ecr::BatchBlobChecker) + /// is configured (production: ECR targets). + /// + /// Production default: 30s -- larger than typical ECR consistency + /// windows but bounded so a stuck view does not hold the engine + /// open. Set to `Duration::ZERO` to disable the wait entirely; the + /// standard retry path will still catch any `BLOB_UPLOAD_UNKNOWN`. + pub manifest_commit_wait: Duration, } impl Default for RetryConfig { @@ -26,6 +34,7 @@ impl Default for RetryConfig { initial_backoff: Duration::from_secs(1), max_backoff: Duration::from_secs(300), backoff_multiplier: 2, + manifest_commit_wait: Duration::from_secs(30), } } } @@ -39,7 +48,7 @@ impl RetryConfig { pub fn backoff_for(&self, attempt: u32) -> Duration { let multiplier = self.backoff_multiplier.saturating_pow(attempt); let backoff = self.initial_backoff.saturating_mul(multiplier); - let capped = std::cmp::min(backoff, self.max_backoff); + let capped = backoff.min(self.max_backoff); jitter(capped) } } @@ -57,6 +66,57 @@ pub fn should_retry(status: StatusCode, current_attempt: u32, max_retries: u32) || status.is_server_error() } +/// ECR (and possibly other registries) can return 404 with the OCI error +/// code `BLOB_UPLOAD_UNKNOWN` when a manifest push references blobs +/// whose PUT-201 came back but haven't been promoted to the +/// manifest-validation index yet. The OCI distribution spec describes +/// `BLOB_UPLOAD_UNKNOWN` as a state that "may be returned" for upload +/// sessions in flux, leaving room for transient interpretation. +/// +/// Returns `true` only when the error is a `RegistryError` with status +/// 404 whose body is structured-OCI-error JSON and contains an +/// `errors[].code == "BLOB_UPLOAD_UNKNOWN"` entry. A free-text mention +/// of the string in some other shape (or a body that fails to parse) is +/// NOT classified as retryable -- substring matching would otherwise +/// false-positive on bodies that reference the code in prose. +/// +/// # Interaction with the manifest-commit blob-visibility wait +/// +/// For ECR targets, [`crate::engine::push_manifests`] now gates the +/// manifest `PUT` on +/// [`BatchBlobChecker::wait_for_blobs_available`](ocync_distribution::ecr::BatchBlobChecker::wait_for_blobs_available), +/// which polls the authoritative `BatchCheckLayerAvailability` API +/// until every referenced layer is visible (deadline: +/// [`RetryConfig::manifest_commit_wait`]). This eliminates the +/// `BLOB_UPLOAD_UNKNOWN` race for the common case. Retrying on this +/// error remains as a defence-in-depth fallback for the wait-deadline +/// case (a stuck consistency view that exceeds +/// `manifest_commit_wait`) and for non-ECR targets that don't have a +/// batch checker configured. +pub fn is_blob_upload_unknown(error: &ocync_distribution::Error) -> bool { + let ocync_distribution::Error::RegistryError { status, message } = error else { + return false; + }; + if *status != StatusCode::NOT_FOUND { + return false; + } + let Ok(body) = serde_json::from_str::(message) else { + return false; + }; + body.errors.iter().any(|e| e.code == "BLOB_UPLOAD_UNKNOWN") +} + +/// OCI distribution-spec error response body shape. +#[derive(serde::Deserialize)] +struct OciErrorBody { + errors: Vec, +} + +#[derive(serde::Deserialize)] +struct OciError { + code: String, +} + /// Determine whether a transport-level (non-HTTP) error should be retried. /// /// Returns `true` for connection failures, request timeouts, mid-stream @@ -67,6 +127,19 @@ pub fn should_retry(status: StatusCode, current_attempt: u32, max_retries: u32) /// failures (e.g. malformed registry responses) are bounded by /// `max_retries`. /// +/// # Predicate set +/// +/// `reqwest::Error::is_request()` covers errors raised during request +/// dispatch (including connect failures and timeouts on the async hyper +/// path - both are wrapped as `Kind::Request` by reqwest). `is_body()` +/// and `is_decode()` cover the response-body and decoder phases, which +/// are NOT classified as `Kind::Request`. Together these three predicates +/// cover the transient-network surface without overlap. +/// +/// The `should_retry_transport_*` tests below verify that connection +/// failures and timeouts both classify as retryable through `is_request()` +/// alone, pinning the equivalence. +/// /// # Known limitation /// /// Only inspects `ocync_distribution::Error::Http(reqwest::Error)`. Transport @@ -79,16 +152,7 @@ pub fn should_retry(status: StatusCode, current_attempt: u32, max_retries: u32) /// show which variant was encountered so the match can be extended. pub fn should_retry_transport(error: &ocync_distribution::Error) -> bool { if let ocync_distribution::Error::Http(reqwest_err) = error { - // `is_request()` is a superset of `is_connect()` and `is_timeout()` - // for the async hyper path (all errors from `Client::send_request` - // are wrapped as Kind::Request). The narrower predicates are kept - // for documentation: they make the intended coverage explicit and - // guard against future reqwest taxonomy changes. - reqwest_err.is_connect() - || reqwest_err.is_timeout() - || reqwest_err.is_body() - || reqwest_err.is_decode() - || reqwest_err.is_request() + reqwest_err.is_request() || reqwest_err.is_body() || reqwest_err.is_decode() } else { tracing::debug!( error = %error, @@ -101,13 +165,11 @@ pub fn should_retry_transport(error: &ocync_distribution::Error) -> bool { /// Apply multiplicative jitter to a backoff duration. /// /// Scales the base duration by a random factor in \[0.75, 1.25) to -/// decorrelate concurrent retries. Uses [`RandomState`] for per-process -/// entropy without requiring a `rand` dependency. +/// decorrelate concurrent retries. Uses `fastrand`'s thread-local PRNG +/// (auto-seeded from OS entropy on first use) so each call is a single +/// `f64` draw with no per-call syscall. fn jitter(base: Duration) -> Duration { - let mut hasher = RandomState::new().build_hasher(); - hasher.write_u64(base.as_nanos() as u64); - let hash = hasher.finish(); - let factor = 0.75 + (hash % 500) as f64 / 1000.0; + let factor = 0.75 + fastrand::f64() * 0.5; base.mul_f64(factor) } @@ -122,6 +184,7 @@ mod tests { assert_eq!(cfg.initial_backoff, Duration::from_secs(1)); assert_eq!(cfg.max_backoff, Duration::from_secs(300)); assert_eq!(cfg.backoff_multiplier, 2); + assert_eq!(cfg.manifest_commit_wait, Duration::from_secs(30)); } /// Helper: assert a duration falls within the jitter range [base*0.75, base*1.25]. @@ -238,6 +301,69 @@ mod tests { assert!(!should_retry(StatusCode::NOT_FOUND, 0, 3)); } + #[test] + fn is_blob_upload_unknown_matches_404_with_marker() { + let err = ocync_distribution::Error::RegistryError { + status: StatusCode::NOT_FOUND, + message: r#"{"errors":[{"code":"BLOB_UPLOAD_UNKNOWN","message":"Layers with digests do not exist"}]}"#.into(), + }; + assert!(is_blob_upload_unknown(&err)); + } + + #[test] + fn is_blob_upload_unknown_rejects_other_404() { + let err = ocync_distribution::Error::RegistryError { + status: StatusCode::NOT_FOUND, + message: r#"{"errors":[{"code":"NAME_UNKNOWN"}]}"#.into(), + }; + assert!(!is_blob_upload_unknown(&err)); + } + + /// Free-text mention of the code outside of `errors[].code` must NOT + /// classify as retryable -- the old substring match would have + /// false-positived here. + #[test] + fn is_blob_upload_unknown_rejects_free_text_mention_of_code() { + let err = ocync_distribution::Error::RegistryError { + status: StatusCode::NOT_FOUND, + message: r#"{"errors":[{"code":"NAME_UNKNOWN","message":"related to BLOB_UPLOAD_UNKNOWN flow"}]}"#.into(), + }; + assert!(!is_blob_upload_unknown(&err)); + } + + /// A malformed body (not parseable as the OCI error shape) is NOT + /// retried. Substring matching would have been ambiguous here. + #[test] + fn is_blob_upload_unknown_rejects_malformed_body() { + let err = ocync_distribution::Error::RegistryError { + status: StatusCode::NOT_FOUND, + message: "BLOB_UPLOAD_UNKNOWN".into(), + }; + assert!(!is_blob_upload_unknown(&err)); + } + + #[test] + fn is_blob_upload_unknown_rejects_non_404() { + let err = ocync_distribution::Error::RegistryError { + status: StatusCode::BAD_REQUEST, + message: r#"{"errors":[{"code":"BLOB_UPLOAD_UNKNOWN"}]}"#.into(), + }; + assert!(!is_blob_upload_unknown(&err)); + } + + #[test] + fn is_blob_upload_unknown_rejects_non_registry_error() { + let err = ocync_distribution::Error::DigestMismatch { + expected: "sha256:0000000000000000000000000000000000000000000000000000000000000000" + .parse() + .unwrap(), + actual: "sha256:1111111111111111111111111111111111111111111111111111111111111111" + .parse() + .unwrap(), + }; + assert!(!is_blob_upload_unknown(&err)); + } + #[test] fn should_not_retry_on_success() { assert!(!should_retry(StatusCode::OK, 0, 3)); @@ -265,8 +391,13 @@ mod tests { } /// Positive-path test: a real reqwest connection failure (refused port) - /// must be classified as retryable. This exercises the `is_connect()` - /// and `is_request()` predicates on a genuine `reqwest::Error`. + /// must be classified as retryable through `is_request()` alone. + /// + /// Pins the comment-claimed equivalence: connection failures on the + /// async hyper path surface as `Kind::Request`, so `is_request()` is + /// sufficient to catch them. If a future reqwest version changes this + /// wrapping, this test will fail and the predicate set must be + /// re-evaluated. #[tokio::test] async fn should_retry_transport_on_connect_failure() { ocync_distribution::install_crypto_provider(); @@ -282,10 +413,64 @@ mod tests { reqwest_err.is_connect(), "expected is_connect(), got: {reqwest_err}" ); + assert!( + reqwest_err.is_request(), + "is_request() must cover connect failures (predicate-set invariant); got: {reqwest_err}" + ); let err = ocync_distribution::Error::Http(reqwest_err); assert!( should_retry_transport(&err), "connection refused should be retryable" ); } + + /// Positive-path test: a real reqwest request timeout must be + /// classified as retryable through `is_request()` alone. + /// + /// Pins the comment-claimed equivalence: timeouts on the async hyper + /// path surface as `Kind::Request`, so `is_request()` is sufficient. + /// Drives the timeout by binding a `TcpListener` that accepts + /// connections but never sends data, then issuing a reqwest GET with + /// a 50ms request timeout. + #[tokio::test] + async fn should_retry_transport_on_request_timeout() { + ocync_distribution::install_crypto_provider(); + + // Bind an accepting-but-silent TCP listener on an ephemeral port. + let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap(); + let port = listener.local_addr().unwrap().port(); + tokio::spawn(async move { + // Hold each accepted connection open without writing so the + // client's read times out. + loop { + let Ok((_sock, _addr)) = listener.accept().await else { + break; + }; + tokio::time::sleep(Duration::from_secs(60)).await; + } + }); + + let client = reqwest::Client::builder() + .timeout(Duration::from_millis(50)) + .build() + .unwrap(); + let reqwest_err = client + .get(format!("http://127.0.0.1:{port}/v2/")) + .send() + .await + .expect_err("request must time out against silent listener"); + assert!( + reqwest_err.is_timeout(), + "expected is_timeout(), got: {reqwest_err}" + ); + assert!( + reqwest_err.is_request(), + "is_request() must cover timeouts (predicate-set invariant); got: {reqwest_err}" + ); + let err = ocync_distribution::Error::Http(reqwest_err); + assert!( + should_retry_transport(&err), + "request timeout should be retryable" + ); + } } diff --git a/crates/ocync-sync/tests/helpers/fixtures.rs b/crates/ocync-sync/tests/helpers/fixtures.rs index 8f50c3ec..56be7848 100644 --- a/crates/ocync-sync/tests/helpers/fixtures.rs +++ b/crates/ocync-sync/tests/helpers/fixtures.rs @@ -59,6 +59,11 @@ pub fn fast_retry() -> RetryConfig { initial_backoff: std::time::Duration::from_millis(1), max_backoff: std::time::Duration::from_millis(10), backoff_multiplier: 2, + // Disable the manifest-commit blob-visibility wait in tests so + // the engine's behaviour matches what existed before the wait + // was introduced. Tests targeting the wait itself construct + // their own RetryConfig with a non-zero value. + manifest_commit_wait: std::time::Duration::ZERO, } } diff --git a/crates/ocync-sync/tests/perf_profile.rs b/crates/ocync-sync/tests/perf_profile.rs new file mode 100644 index 00000000..ccf788f9 --- /dev/null +++ b/crates/ocync-sync/tests/perf_profile.rs @@ -0,0 +1,262 @@ +//! CPU profile harness for the sync engine. +//! +//! Pushes a deterministic corpus of small images to one local `registry:2`, +//! then syncs it to a second local `registry:2` so a profiler can sample +//! `SyncEngine::run`. Tests are `#[ignore]`'d -- they spin up Docker +//! containers and run in the seconds-range, intended for ad-hoc profiling. +//! +//! ## Tests +//! +//! - `profile_small_images` -- HTTP-only. Baseline. Useful for SHA-256, +//! JSON parsing, and task-scheduling cost. +//! - `profile_small_images_tls` -- TLS-terminated via `registry:2`'s +//! built-in HTTPS support with a self-signed cert generated at test +//! start. Surfaces rustls handshake + AEAD cost. +//! +//! ## Usage +//! +//! ```bash +//! samply record --output /tmp/sync.profile -- \ +//! cargo test --profile profiling --test perf_profile -- \ +//! --ignored --nocapture --exact profile_small_images +//! ``` +//! +//! Then `samply load /tmp/sync.profile` to view the trace. Look for the +//! `[profile] PROFILE BEGIN` / `[profile] PROFILE END` stderr markers to +//! scope the analysis window in the UI. +//! +//! ## Tunables +//! +//! - `OCYNC_PROFILE_IMAGES` number of images (default: 200) +//! - `OCYNC_PROFILE_LAYERS` layers per image (default: 2) +//! - `OCYNC_PROFILE_BYTES` bytes per layer (default: 16384). Bump to +//! ~5 MB to amplify SHA-256 cost like a real container layer. +//! - `OCYNC_PROFILE_WORKERS` `max_concurrent_transfers` (default: 50; +//! note the engine default is 10 -- the harness deliberately oversubscribes +//! to stress h2 stream budgeting). +//! - `OCYNC_PROFILE_HTTP1=1` refuse to negotiate HTTP/2 via ALPN. +//! Used to isolate HTTP/2 multiplexing stalls from the rest of the path. +//! - `OCYNC_PROFILE_H2_ADAPTIVE=0` disable HTTP/2 adaptive-window sizing +//! for A/B testing the pre-fix behavior. +//! - `OCYNC_PROFILE_STREAM_CAP` override per-`RegistryClient` +//! `streaming_blob_concurrency` (default: 64). Lower to verify the +//! per-target stream cap binds. + +mod helpers; + +use std::sync::Arc; +use std::time::Instant; + +use bytes::Bytes; +use futures_util::stream; +use ocync_distribution::sha256::Sha256; +use ocync_distribution::spec::{MediaType, RepositoryName}; +use ocync_distribution::{Digest, RegistryClient, RegistryClientBuilder}; +use ocync_sync::engine::{SyncEngine, TagPair}; +use ocync_sync::progress::NullProgress; +use ocync_sync::staging::BlobStage; +use rcgen::{CertificateParams, DnType, KeyPair}; +use testcontainers::core::WaitFor; +use testcontainers::runners::AsyncRunner; +use testcontainers::{ContainerAsync, GenericImage, ImageExt}; +use url::Url; + +use helpers::*; + +fn env_usize(key: &str, default: usize) -> usize { + std::env::var(key) + .ok() + .and_then(|v| v.parse().ok()) + .unwrap_or(default) +} + +async fn start_registry_http() -> (ContainerAsync, Url) { + let container = GenericImage::new("registry", "2") + .with_exposed_port(5000.into()) + .with_wait_for(WaitFor::message_on_stderr("listening on")) + .start() + .await + .expect("registry:2 container failed to start"); + let port = container + .get_host_port_ipv4(5000) + .await + .expect("get_host_port_ipv4 failed"); + let url = Url::parse(&format!("http://127.0.0.1:{port}")).unwrap(); + (container, url) +} + +/// Self-signed cert covering `localhost` + `127.0.0.1`, valid for the +/// run of the test. Generated per test invocation so there's no +/// on-disk artifact to manage. +fn self_signed_localhost() -> (Vec, Vec) { + let key = KeyPair::generate().expect("rcgen KeyPair::generate"); + let mut params = CertificateParams::new(vec!["localhost".to_string(), "127.0.0.1".to_string()]) + .expect("rcgen CertificateParams::new"); + params + .distinguished_name + .push(DnType::CommonName, "ocync-perf-profile"); + let cert = params.self_signed(&key).expect("rcgen self_signed"); + (cert.pem().into_bytes(), key.serialize_pem().into_bytes()) +} + +async fn start_registry_tls() -> (ContainerAsync, Url) { + let (cert_pem, key_pem) = self_signed_localhost(); + let container = GenericImage::new("registry", "2") + .with_exposed_port(5000.into()) + .with_wait_for(WaitFor::message_on_stderr("listening on")) + .with_env_var("REGISTRY_HTTP_TLS_CERTIFICATE", "/certs/tls.crt") + .with_env_var("REGISTRY_HTTP_TLS_KEY", "/certs/tls.key") + .with_copy_to("/certs/tls.crt", cert_pem) + .with_copy_to("/certs/tls.key", key_pem) + .start() + .await + .expect("registry:2 (TLS) container failed to start"); + let port = container + .get_host_port_ipv4(5000) + .await + .expect("get_host_port_ipv4 failed"); + let url = Url::parse(&format!("https://127.0.0.1:{port}")).unwrap(); + (container, url) +} + +fn make_client(url: Url, accept_invalid_certs: bool) -> Arc { + let force_http1 = std::env::var("OCYNC_PROFILE_HTTP1").ok().as_deref() == Some("1"); + // Adaptive window is on by default in the builder. Set + // OCYNC_PROFILE_H2_ADAPTIVE=0 to disable for A/B testing the + // pre-fix behavior. + let mut builder = RegistryClientBuilder::new(url) + .allow_invalid_certs(accept_invalid_certs) + .force_http1(force_http1); + if std::env::var("OCYNC_PROFILE_H2_ADAPTIVE").ok().as_deref() == Some("0") { + builder = builder.http2_adaptive_window(false); + } + if let Some(cap) = std::env::var("OCYNC_PROFILE_STREAM_CAP") + .ok() + .and_then(|v| v.parse().ok()) + { + builder = builder.streaming_blob_concurrency(cap); + } + Arc::new(builder.build().expect("RegistryClient")) +} + +async fn run_profile(src: Arc, dst: Arc, label: &str) { + let n_images = env_usize("OCYNC_PROFILE_IMAGES", 200); + let n_layers = env_usize("OCYNC_PROFILE_LAYERS", 2); + let layer_bytes = env_usize("OCYNC_PROFILE_BYTES", 16 * 1024); + let workers = env_usize("OCYNC_PROFILE_WORKERS", 50); + + eprintln!( + "[profile] {label}: populating source: {n_images} images x {n_layers} layers x {layer_bytes} B" + ); + let pop_start = Instant::now(); + let repos = populate_source(&src, n_images, n_layers, layer_bytes).await; + eprintln!("[profile] {label}: populate took {:?}", pop_start.elapsed()); + + let mappings = repos + .iter() + .map(|repo| { + resolved_mapping( + Arc::clone(&src), + repo.as_str(), + repo.as_str(), + vec![target_entry("target", Arc::clone(&dst))], + vec![TagPair::same("v1")], + ) + }) + .collect::>(); + + let engine = SyncEngine::new(fast_retry(), workers); + + eprintln!("[profile] PROFILE BEGIN {label} workers={workers}"); + let sync_start = Instant::now(); + let report = engine + .run( + mappings, + empty_cache(), + BlobStage::disabled(), + &NullProgress, + None, + ) + .await; + let sync_elapsed = sync_start.elapsed(); + eprintln!("[profile] PROFILE END {label} elapsed={sync_elapsed:?}"); + + let synced = report + .images + .iter() + .filter(|r| matches!(r.status, ocync_sync::ImageStatus::Synced)) + .count(); + eprintln!( + "[profile] {label}: images={n_images} synced={synced} blobs_transferred={} bytes={}", + report.stats.blobs_transferred, report.stats.bytes_transferred, + ); + assert_eq!( + synced, n_images, + "expected all {n_images} images to sync; got {synced}", + ); +} + +#[tokio::test(flavor = "current_thread")] +#[ignore = "spins up Docker containers; run via samply, not in CI"] +async fn profile_small_images() { + let (_src_ctr, src_url) = start_registry_http().await; + let (_dst_ctr, dst_url) = start_registry_http().await; + let src = make_client(src_url, false); + let dst = make_client(dst_url, false); + run_profile(src, dst, "http").await; +} + +#[tokio::test(flavor = "current_thread")] +#[ignore = "spins up Docker containers; run via samply, not in CI"] +async fn profile_small_images_tls() { + let (_src_ctr, src_url) = start_registry_tls().await; + let (_dst_ctr, dst_url) = start_registry_tls().await; + let src = make_client(src_url, true); + let dst = make_client(dst_url, true); + run_profile(src, dst, "tls").await; +} + +async fn push_blob_stream(client: &RegistryClient, repo: &RepositoryName, data: Vec) { + let digest = Digest::from_sha256(Sha256::digest(&data)); + let size = data.len() as u64; + let body = Bytes::from(data); + let s = stream::once(async move { Ok::<_, ocync_distribution::Error>(body) }); + client + .blob_push_stream(repo, &digest, Some(size), s) + .await + .expect("blob_push_stream"); +} + +async fn populate_source( + client: &RegistryClient, + n_images: usize, + n_layers: usize, + layer_bytes: usize, +) -> Vec { + let mut repos = Vec::with_capacity(n_images); + for i in 0..n_images { + let repo = RepositoryName::new(format!("perf/img-{i:04}")).unwrap(); + let config_data = format!("{{\"image\":{i}}}").into_bytes(); + // Stream the corpus into the source registry the same way the + // engine does during sync; otherwise the harness's setup-time SHA + // dominates the profile and the actual streaming path is invisible. + push_blob_stream(client, &repo, config_data.clone()).await; + + let mut builder = ManifestBuilder::new(&config_data); + for l in 0..n_layers { + let mut layer = vec![0u8; layer_bytes]; + // Vary content per (image, layer) so digests are unique. + layer[0] = (i & 0xff) as u8; + layer[1] = (l & 0xff) as u8; + push_blob_stream(client, &repo, layer.clone()).await; + builder = builder.layer(&layer); + } + let parts = builder.build(); + client + .manifest_push(&repo, "v1", &MediaType::OciManifest, &parts.bytes) + .await + .expect("manifest push"); + repos.push(repo); + } + repos +} diff --git a/crates/ocync-sync/tests/sync_cache.rs b/crates/ocync-sync/tests/sync_cache.rs index 0502491a..c84d3be2 100644 --- a/crates/ocync-sync/tests/sync_cache.rs +++ b/crates/ocync-sync/tests/sync_cache.rs @@ -121,6 +121,78 @@ impl BatchBlobChecker for FailingBatchChecker { } } +/// Mock that overrides `wait_for_blobs_available` directly so the test can +/// count *commit-gate* invocations separately from pre-discovery +/// `check_blob_existence` calls. +struct WaitTrackingChecker { + expected_repo: String, + existing: HashSet, + /// Count of pre-discovery `check_blob_existence` calls. + check_count: Arc, + /// Count of commit-gate `wait_for_blobs_available` calls. Set by the + /// engine's `push_manifests` exactly when the wait fires. + wait_count: Arc, +} + +impl WaitTrackingChecker { + fn new( + expected_repo: &str, + existing: HashSet, + ) -> (Self, Arc, Arc) { + let check_count = Arc::new(AtomicUsize::new(0)); + let wait_count = Arc::new(AtomicUsize::new(0)); + ( + Self { + expected_repo: expected_repo.to_owned(), + existing, + check_count: Arc::clone(&check_count), + wait_count: Arc::clone(&wait_count), + }, + check_count, + wait_count, + ) + } +} + +impl BatchBlobChecker for WaitTrackingChecker { + fn check_blob_existence<'a>( + &'a self, + repo: &'a RepositoryName, + digests: &'a [Digest], + ) -> Pin, ocync_distribution::Error>> + 'a>> + { + assert_eq!(repo.as_str(), self.expected_repo); + Box::pin(async move { + self.check_count.fetch_add(1, Ordering::Relaxed); + Ok(digests + .iter() + .filter(|d| self.existing.contains(d)) + .cloned() + .collect()) + }) + } + + fn wait_for_blobs_available<'a>( + &'a self, + repo: &'a RepositoryName, + digests: &'a [Digest], + _deadline: std::time::Duration, + ) -> Pin> + 'a>> { + assert_eq!(repo.as_str(), self.expected_repo); + let all_known = digests.iter().all(|d| self.existing.contains(d)); + Box::pin(async move { + self.wait_count.fetch_add(1, Ordering::Relaxed); + if all_known { + Ok(()) + } else { + Err(ocync_distribution::Error::EcrApi { + reason: "test: some digests still missing".into(), + }) + } + }) + } +} + // --------------------------------------------------------------------------- // Tests: progressive cache population, cross-repo mount, monolithic upload, // lazy invalidation, and cache persistence round-trip. @@ -1724,3 +1796,101 @@ async fn sync_batch_checker_with_prewarmed_cache() { assert_eq!(report.stats.bytes_transferred, 0); // wiremock expect(0) on blob HEADs verifies no fallback path was used. } + +// --------------------------------------------------------------------------- +// Tests: manifest-commit blob-visibility gating (ECR consistency window) +// --------------------------------------------------------------------------- + +/// With a non-zero `manifest_commit_wait`, the engine calls the target's +/// `BatchBlobChecker::wait_for_blobs_available` exactly once per manifest +/// commit (before the manifest `PUT`). With the wait disabled +/// (`Duration::ZERO`, the test-default), the wait is NOT called. +/// +/// This is the gating check for ECR's consistency window: blob `PUT-201`s +/// can land before the manifest validator's view catches up, so a +/// manifest `PUT` issued immediately after blob upload would fail with +/// `BLOB_UPLOAD_UNKNOWN`. The wait converts that race into a +/// deterministic gate on the authoritative blob-visibility API. +/// +/// The mock pre-marks all blobs as existing so blob uploads are skipped +/// (Step 1 cache hit); the manifest `PUT` still fires and is the +/// observable trigger for the wait. +#[tokio::test] +async fn manifest_commit_wait_fires_only_when_enabled() { + use ocync_sync::engine::SyncEngine; + use ocync_sync::progress::NullProgress; + use ocync_sync::retry::RetryConfig; + use ocync_sync::staging::BlobStage; + + async fn run_one(commit_wait: std::time::Duration) -> (usize, usize) { + let source_server = MockServer::start().await; + let target_server = MockServer::start().await; + + let parts = ManifestBuilder::new(b"cfg-mw").layer(b"layer-mw").build(); + // Source serves the manifest under `src/repo`. The mapping below + // pulls from `src/repo` and pushes to `tgt/repo`. + mount_source_manifest(&source_server, "src/repo", "v1", &parts.bytes).await; + mount_manifest_head_not_found(&target_server, "tgt/repo", "v1").await; + mount_manifest_push(&target_server, "tgt/repo", "v1").await; + + let existing = HashSet::from([ + parts.config_desc.digest.clone(), + parts.layer_descs[0].digest.clone(), + ]); + let (checker, check_count, wait_count) = WaitTrackingChecker::new("tgt/repo", existing); + + let mapping = resolved_mapping( + mock_client(&source_server), + "src/repo", + "tgt/repo", + vec![TargetEntry { + name: RegistryAlias::new("target"), + client: mock_client(&target_server), + batch_checker: Some(Rc::new(checker)), + existing_tags: HashSet::new(), + }], + vec![TagPair::same("v1")], + ); + + // Build the engine with the test's chosen manifest_commit_wait. + let retry = RetryConfig { + max_retries: 2, + initial_backoff: std::time::Duration::from_millis(1), + max_backoff: std::time::Duration::from_millis(10), + backoff_multiplier: 2, + manifest_commit_wait: commit_wait, + }; + let report = SyncEngine::new(retry, 4) + .run( + vec![mapping], + empty_cache(), + BlobStage::disabled(), + &NullProgress, + None, + ) + .await; + assert_status!(report, 0, ImageStatus::Synced); + + ( + check_count.load(Ordering::Relaxed), + wait_count.load(Ordering::Relaxed), + ) + } + + // Wait DISABLED: wait_for_blobs_available must NOT be invoked. + let (checks_off, waits_off) = run_one(std::time::Duration::ZERO).await; + assert!( + waits_off == 0, + "with manifest_commit_wait=0 the commit-gate wait must not fire (got {waits_off})" + ); + // Pre-discovery still uses check_blob_existence at least once. + assert!(checks_off >= 1, "pre-discovery batch check expected"); + + // Wait ENABLED: wait_for_blobs_available must fire exactly once + // per manifest commit (one image in this test). + let (_checks_on, waits_on) = run_one(std::time::Duration::from_millis(200)).await; + assert_eq!( + waits_on, 1, + "with manifest_commit_wait>0 the commit-gate wait must fire once per manifest" + ); +} diff --git a/crates/ocync-sync/tests/sync_disjoint_high_concurrency.rs b/crates/ocync-sync/tests/sync_disjoint_high_concurrency.rs new file mode 100644 index 00000000..23b36c1a --- /dev/null +++ b/crates/ocync-sync/tests/sync_disjoint_high_concurrency.rs @@ -0,0 +1,350 @@ +//! High-concurrency, mostly-disjoint mapping reproducer. +//! +//! Mirrors the production failure mode: 10 mappings to one target registry, +//! mostly disjoint blob sets, `max_concurrent_transfers=50`. One pair +//! shares a single layer so `elect_leaders` picks exactly one leader +//! (matching the `total_leaders=1 images=10` log observed against ECR). +//! +//! The engine reports `ImageStatus::Synced` only when the blob loop and +//! manifest push both succeed. If the engine silently short-circuits a blob +//! push (returning `Skipped`/`Mounted` for a blob that was never actually +//! transferred), the per-target request counts will be lower than the +//! per-image stats claim -- and these assertions fire. +//! +//! Three variants: +//! - [`sync_ten_fully_disjoint_mappings_high_concurrency`] -- baseline +//! with zero shared blobs, exercises the no-leader path. +//! - [`sync_ten_disjoint_mappings_one_shared_blob_high_concurrency`] -- +//! one shared blob, one elected leader, follower mounts. +//! - [`sync_ten_disjoint_mappings_one_shared_blob_with_batch_checker`] -- +//! same shape, plus a `BatchBlobChecker` returning `existing=[]` for +//! every repo (exercises the ECR cache-pre-population path). + +mod helpers; + +use std::collections::HashSet; +use std::future::Future; +use std::pin::Pin; +use std::rc::Rc; +use std::sync::Arc; +use std::sync::atomic::{AtomicUsize, Ordering}; +use std::time::Duration; + +use ocync_distribution::spec::RepositoryName; +use ocync_distribution::{BatchBlobChecker, Digest}; +use ocync_sync::ImageStatus; +use ocync_sync::engine::{RegistryAlias, SyncEngine, TagPair, TargetEntry}; +use ocync_sync::progress::NullProgress; +use ocync_sync::staging::BlobStage; +use wiremock::matchers::{method, path_regex, query_param}; +use wiremock::{Mock, MockServer, ResponseTemplate}; + +use helpers::*; + +/// Production-shape reproducer. Asserts every image syncs and that the +/// report's per-blob stats reconcile against the target server's actual +/// PUT and mount POST counts. +#[tokio::test(flavor = "current_thread")] +async fn sync_ten_disjoint_mappings_one_shared_blob_high_concurrency() { + let fixture = build_ten_image_fixture(/* shared_pair = */ true).await; + let report = run_engine_default(&fixture, /* with_batch_checker = */ false).await; + assert_report_matches_target_receipts(&fixture, &report).await; +} + +/// As above, but each `TargetEntry` carries an ECR-style batch checker +/// that returns `existing=[]` for every call. Confirms the cache +/// pre-population path doesn't introduce false-Skipped or false-Mounted +/// terminals at high concurrency. +#[tokio::test(flavor = "current_thread")] +async fn sync_ten_disjoint_mappings_one_shared_blob_with_batch_checker() { + let fixture = build_ten_image_fixture(/* shared_pair = */ true).await; + let report = run_engine_default(&fixture, /* with_batch_checker = */ true).await; + assert_report_matches_target_receipts(&fixture, &report).await; +} + +/// Sanity baseline: zero shared blobs. `elect_leaders` returns 0 (no +/// leader log fires). Every image processes independently. +#[tokio::test(flavor = "current_thread")] +async fn sync_ten_fully_disjoint_mappings_high_concurrency() { + let fixture = build_ten_image_fixture(/* shared_pair = */ false).await; + let report = run_engine_default(&fixture, /* with_batch_checker = */ false).await; + assert_report_matches_target_receipts(&fixture, &report).await; + // No shared blobs -> no mount candidates -> no mount POSTs at all. + assert_eq!(report.stats.blobs_mounted, 0); +} + +// --------------------------------------------------------------------------- +// Shared fixture + assertions +// --------------------------------------------------------------------------- + +struct TenImageFixture { + source_server: MockServer, + target_server: MockServer, + source_client: Arc, + target_client: Arc, + images: Vec, + // (source_repo, target_repo) per image index. + repos: Vec<(String, String)>, +} + +async fn build_ten_image_fixture(shared_pair: bool) -> TenImageFixture { + let source_server = MockServer::start().await; + let target_server = MockServer::start().await; + + let shared_layer = b"shared-layer-between-images-0-and-1".to_vec(); + + let mut images = Vec::with_capacity(10); + for i in 0..10usize { + let config = format!("config-image-{i}").into_bytes(); + let layer_a = format!("image-{i}-layer-a").into_bytes(); + let layer_b = format!("image-{i}-layer-b").into_bytes(); + + let mut builder = ManifestBuilder::new(&config) + .layer(&layer_a) + .layer(&layer_b); + if shared_pair && i < 2 { + builder = builder.layer(&shared_layer); + } + images.push(builder.build()); + } + + let repos: Vec<(String, String)> = (0..10) + .map(|i| (format!("src/img-{i}"), format!("tgt/img-{i}"))) + .collect(); + + for (parts, (src, _)) in images.iter().zip(repos.iter()) { + parts.mount_source(&source_server, src, "latest").await; + } + + for (_, tgt) in &repos { + mount_manifest_head_not_found(&target_server, tgt, "latest").await; + mount_manifest_push(&target_server, tgt, "latest").await; + // Blob HEAD 404 for every digest this repo might be asked about + // (without a batch checker, the engine HEADs every blob). + for parts in &images { + mount_blob_not_found(&target_server, tgt, &parts.config_desc.digest).await; + for desc in &parts.layer_descs { + mount_blob_not_found(&target_server, tgt, &desc.digest).await; + } + } + mount_blob_push(&target_server, tgt).await; + + // Mount mocks: 201 for any cross-repo mount of any of our digests. + for parts in &images { + for desc in &parts.layer_descs { + Mock::given(method("POST")) + .and(path_regex(format!("^/v2/{tgt}/blobs/uploads/$"))) + .and(query_param("mount", desc.digest.to_string())) + .respond_with(ResponseTemplate::new(201)) + .with_priority(1) + .mount(&target_server) + .await; + } + Mock::given(method("POST")) + .and(path_regex(format!("^/v2/{tgt}/blobs/uploads/$"))) + .and(query_param("mount", parts.config_desc.digest.to_string())) + .respond_with(ResponseTemplate::new(201)) + .with_priority(1) + .mount(&target_server) + .await; + } + } + + let source_client = mock_client(&source_server); + let target_client = mock_client(&target_server); + + TenImageFixture { + source_server, + target_server, + source_client, + target_client, + images, + repos, + } +} + +async fn run_engine_default( + fixture: &TenImageFixture, + with_batch_checker: bool, +) -> ocync_sync::SyncReport { + let mappings: Vec<_> = fixture + .repos + .iter() + .map(|(src, tgt)| { + let batch_checker: Option> = if with_batch_checker { + Some(Rc::new(EmptyExistingChecker::new(tgt))) + } else { + None + }; + let entry = TargetEntry { + name: RegistryAlias::new("target"), + client: Arc::clone(&fixture.target_client), + batch_checker, + existing_tags: HashSet::new(), + }; + resolved_mapping( + Arc::clone(&fixture.source_client), + src, + tgt, + vec![entry], + vec![TagPair::same("latest")], + ) + }) + .collect(); + + // Engine default is 10 (DEFAULT_MAX_CONCURRENT_TRANSFERS); the test + // deliberately oversubscribes at 50 to exercise the per-target + // streaming-blob semaphore (cap 64) under a workload that would + // otherwise demand 50 * BLOB_CONCURRENCY=6 = 300 concurrent streams. + let engine = SyncEngine::new(fast_retry(), 50); + tokio::time::timeout( + Duration::from_secs(30), + engine.run( + mappings, + empty_cache(), + BlobStage::disabled(), + &NullProgress, + None, + ), + ) + .await + .expect("engine hung past 30s -- leader/follower stall or auth deadlock") +} + +async fn assert_report_matches_target_receipts( + fixture: &TenImageFixture, + report: &ocync_sync::SyncReport, +) { + // Every image must claim Synced. + assert_eq!(report.images.len(), 10, "expected 10 image results"); + for r in &report.images { + assert!( + matches!(r.status, ImageStatus::Synced), + "{} -> {} not Synced: {:#?}", + r.source, + r.target, + r.status, + ); + } + + // Per-image declared blob count: config + N layers. + let declared: Vec = fixture + .images + .iter() + .map(|p| 1 + p.layer_descs.len() as u64) + .collect(); + for r in &report.images { + let i = parse_image_index(&r.source); + let handled = r.blob_stats.transferred + r.blob_stats.mounted + r.blob_stats.skipped; + assert_eq!( + handled, declared[i], + "image[{i}]: handled blob count {handled} != declared {} (transferred={}, mounted={}, skipped={})", + declared[i], r.blob_stats.transferred, r.blob_stats.mounted, r.blob_stats.skipped, + ); + } + + let target_requests = fixture.target_server.received_requests().await.unwrap(); + + let blob_puts: u64 = target_requests + .iter() + .filter(|r| { + r.method == wiremock::http::Method::PUT && r.url.path().contains("/blobs/uploads/") + }) + .count() as u64; + let blob_mounts: u64 = target_requests + .iter() + .filter(|r| { + r.method == wiremock::http::Method::POST + && r.url.path().ends_with("/blobs/uploads/") + && r.url.query_pairs().any(|(k, _)| k == "mount") + }) + .count() as u64; + let manifest_puts: u64 = target_requests + .iter() + .filter(|r| r.method == wiremock::http::Method::PUT && r.url.path().contains("/manifests/")) + .count() as u64; + + let stats_transferred: u64 = report.images.iter().map(|r| r.blob_stats.transferred).sum(); + let stats_mounted: u64 = report.images.iter().map(|r| r.blob_stats.mounted).sum(); + let stats_skipped: u64 = report.images.iter().map(|r| r.blob_stats.skipped).sum(); + + eprintln!( + "[repro] target receipts: blob_puts={blob_puts} blob_mounts={blob_mounts} \ + report.transferred={stats_transferred} report.mounted={stats_mounted} \ + report.skipped={stats_skipped}", + ); + + // No skipping: empty cache + fresh target repos. A non-zero + // `stats_skipped` would mean a blob was reported skipped without an + // actual reason (false-positive cache state or HEAD). + assert_eq!( + stats_skipped, 0, + "blobs were reported Skipped despite an empty cache and fresh target repos", + ); + + // The decisive checks: report-claimed counts must match what the + // target actually received. + assert_eq!( + blob_puts, stats_transferred, + "report claims {stats_transferred} blobs transferred but target received {blob_puts} PUT uploads -- engine is short-circuiting blob pushes", + ); + assert_eq!( + blob_mounts, stats_mounted, + "report claims {stats_mounted} blobs mounted but target received {blob_mounts} mount POSTs", + ); + assert_eq!( + manifest_puts, 10, + "expected 10 manifest PUTs, got {manifest_puts}" + ); + + // Suppress unused-field warnings for fields kept on the fixture + // for diagnostic clarity at failure time. + let _ = &fixture.source_server; +} + +fn parse_image_index(source: &str) -> usize { + source + .strip_prefix("src/img-") + .and_then(|s| s.strip_suffix(":latest")) + .and_then(|s| s.parse().ok()) + .unwrap_or_else(|| panic!("source did not match src/img-:latest: {source}")) +} + +// --------------------------------------------------------------------------- +// Test-local batch checker mirroring ECR's BatchCheckLayerAvailability +// against an empty target. Asserts the caller passes the expected repo +// (per-mock contract fidelity). +// --------------------------------------------------------------------------- + +struct EmptyExistingChecker { + expected_repo: String, + call_count: Arc, +} + +impl EmptyExistingChecker { + fn new(expected_repo: &str) -> Self { + Self { + expected_repo: expected_repo.to_owned(), + call_count: Arc::new(AtomicUsize::new(0)), + } + } +} + +impl BatchBlobChecker for EmptyExistingChecker { + fn check_blob_existence<'a>( + &'a self, + repo: &'a RepositoryName, + _digests: &'a [Digest], + ) -> Pin, ocync_distribution::Error>> + 'a>> + { + assert_eq!( + repo.as_str(), + self.expected_repo, + "batch checker called with wrong repo", + ); + Box::pin(async { + self.call_count.fetch_add(1, Ordering::Relaxed); + Ok(HashSet::new()) + }) + } +} diff --git a/docs/public/config.schema.json b/docs/public/config.schema.json index 8daf6275..8b0d79e3 100644 --- a/docs/public/config.schema.json +++ b/docs/public/config.schema.json @@ -255,10 +255,10 @@ ] }, "max_concurrent_transfers": { - "description": "Maximum concurrent image syncs (default: 50).", + "description": "Maximum concurrent image syncs (default: 10).\n\nDefault derived as `streaming_blob_concurrency / BLOB_CONCURRENCY = 64\n/ 6`, sized so the per-target HTTP/2 stream budget is not exhausted\n(typical advertised `SETTINGS_MAX_CONCURRENT_STREAMS` is 100-128\nacross major container registries).", "type": "integer", "format": "uint", - "default": 50, + "default": 10, "minimum": 0 }, "staging_size_limit": { @@ -383,7 +383,7 @@ "default": null }, "head_first": { - "description": "HEAD-check targets before pulling full source manifests on cache miss.\n\nWhen enabled, the engine issues a manifest HEAD against all targets\nbefore performing a full source manifest GET. If every target already\nholds the same digest as the source HEAD, the expensive GET is skipped.\nThis conserves rate-limit tokens on source registries with aggressive\nquotas (e.g., Docker Hub).", + "description": "HEAD-check targets before pulling full source manifests on cache miss.\n\nWhen enabled, the engine issues a manifest HEAD against all targets\nbefore performing a full source manifest GET. If every target already\nholds the same digest as the source HEAD, the expensive GET is skipped.\nThis conserves rate-limit tokens on source registries with aggressive\nquotas (e.g., Docker Hub).\n\nApplies to both `sync` (which is the original target) and `copy`\n(single-image variant); in `copy` the optimization fires when the\ndestination tag already resolves to the same source digest.", "type": "boolean", "default": false }, diff --git a/docs/src/content/configuration.md b/docs/src/content/configuration.md index 8ac6c49d..aac81fa0 100644 --- a/docs/src/content/configuration.md +++ b/docs/src/content/configuration.md @@ -30,7 +30,7 @@ mappings: ```yaml global: - max_concurrent_transfers: 50 + max_concurrent_transfers: 10 cache_ttl: "12h" staging_size_limit: "2GB" @@ -73,7 +73,7 @@ Top-level `global` section controls engine-wide behavior: ```yaml global: - max_concurrent_transfers: 50 # Maximum concurrent image syncs (default: 50) + max_concurrent_transfers: 10 # Maximum concurrent image syncs (default: 10) cache_dir: /var/cache/ocync # Cache directory (default: next to config file) cache_ttl: "12h" # Warm cache TTL (default: "12h", "0" disables) staging_size_limit: "2GB" # Disk staging limit (SI prefixes, "0" disables) @@ -81,7 +81,7 @@ global: | Field | Default | Description | |---|---|---| -| `max_concurrent_transfers` | `50` | Maximum number of images synced in parallel. Must be >= 1 | +| `max_concurrent_transfers` | `10` | Maximum number of images synced in parallel. Must be >= 1. Each image runs up to 6 concurrent blob transfers; the per-target HTTP/2 connection bounds streaming-blob concurrency at 64 by default (configurable via the registry client's `streaming_blob_concurrency`) to stay under the typical advertised `SETTINGS_MAX_CONCURRENT_STREAMS` of 100-128 across major registries | | `cache_dir` | Adjacent to config file | Directory for persistent cache and blob staging | | `cache_ttl` | `"12h"` | How long cached blob existence checks are valid. Accepts bare integers (seconds) or integers with a suffix: `s`, `m`, `h`, `d`. `"0"` disables TTL expiry (lazy invalidation only) | | `staging_size_limit` | Unlimited | Maximum disk space for blob staging. Accepts `"0"` (disabled) or an integer with a suffix: `B`, `KB`, `MB`, `GB`, `TB`. Uses SI decimal prefixes (1 GB = 1,000,000,000 bytes) | diff --git a/docs/src/content/design/engine.md b/docs/src/content/design/engine.md index a407e753..fc8a5a53 100644 --- a/docs/src/content/design/engine.md +++ b/docs/src/content/design/engine.md @@ -72,7 +72,7 @@ Concurrency is controlled at four levels that compose naturally, replacing any n ### Four-level hierarchy -**Level 1, global image semaphore** (default: 50). Bounds how many `(tag, target)` pairs are in-flight simultaneously, preventing memory explosion. This is the engine-level `max_concurrent_transfers` config. +**Level 1, global image semaphore** (default: 10). Bounds how many `(tag, target)` pairs are in-flight simultaneously, preventing memory explosion. This is the engine-level `max_concurrent_transfers` config. The default is derived as `streaming_blob_concurrency / BLOB_CONCURRENCY = 64 / 6`, sized so the per-target HTTP/2 stream budget is not exhausted (typical advertised `SETTINGS_MAX_CONCURRENT_STREAMS` is 100-128 across major registries). **Level 2, per-registry aggregate semaphore** (`max_concurrent` per registry, default: 50). Bounds total concurrent HTTP requests to a single registry host across all action types. This is a safety ceiling for connection/memory pressure, not a rate-limit mechanism. With HTTP/2 multiplexing, 100+ concurrent requests share ~6-8 TCP connections, so the aggregate cap is conservative. A request must acquire a permit from this semaphore before proceeding to the per-action AIMD check. diff --git a/docs/src/content/registries/acr.md b/docs/src/content/registries/acr.md index ba93529f..eb5f25c3 100644 --- a/docs/src/content/registries/acr.md +++ b/docs/src/content/registries/acr.md @@ -19,7 +19,7 @@ Notable behaviors: - Two-step OAuth2 exchange: AAD access token -> `POST /oauth2/exchange` -> ACR refresh token (~3h) -> `POST /oauth2/token` with scope -> ACR access token (~75min). TTLs are driven by the JWT `exp` claim. - Sovereign cloud routing: `*.azurecr.cn` uses `login.chinacloudapi.cn`; `*.azurecr.us` uses `login.microsoftonline.us`. The hostname suffix selects both the AAD authority and the ACR resource endpoint. -- **~20 MB streaming PUT body limit.** ACR rejects streaming uploads above ~20 MB with a connection reset or 413; chunked PATCH fallback is not yet implemented. Blobs exceeding this limit cannot currently be pushed to ACR. +- **~20 MB streaming PUT body limit.** ACR rejects streaming uploads above ~20 MB with a connection reset or 413. ocync detects `*.azurecr.io` / `*.azurecr.cn` / `*.azurecr.us` hosts and routes blob pushes through a chunked-PATCH fallback (`blob_push_stream_acr`): the stream is buffered, the digest is verified up front, then the blob is split into 16 MiB PATCH chunks (Content-Range in the OCI `{start}-{end}` form, not RFC 7233) and finalised with a PUT. Each PATCH response's `Location` header is checked against the original upload host so a compromised proxy cannot redirect later chunks (and the bearer-bearing finalize PUT) to an attacker. - Two rate-limit windows: ACR enforces separate ReadOps and WriteOps quotas; ocync tracks them as two AIMD windows. ## CLI example diff --git a/docs/src/content/registries/ghcr.md b/docs/src/content/registries/ghcr.md index a848c8aa..651e7cda 100644 --- a/docs/src/content/registries/ghcr.md +++ b/docs/src/content/registries/ghcr.md @@ -23,6 +23,8 @@ Notable behaviors: - Multi-PATCH chunked upload is broken on GHCR (last PATCH overwrites previous). ocync falls back to POST + single PATCH + PUT (3 requests/blob instead of 2) automatically. - Cross-repo blob mounting is fulfilled within the same org or user namespace. Mount POSTs returning 202 (not fulfilled) fall through to upload. - Single 2000 RPM aggregate cap across all reads and writes per authenticated principal. ocync uses one AIMD window so the token-bucket layer cannot exceed the cap by spending read and write budgets concurrently. +- **10-minute server-side upload timeout** per layer (community-documented, [github.com/orgs/community/discussions/77429](https://github.com/orgs/community/discussions/77429)). Because GHCR's single-PATCH path requires sending the whole blob in one request, a multi-GB layer over a slow link can be cut off mid-body. ocync's `with_retry` classifies the resulting connection reset as transport-retryable and starts a fresh upload session (POST + PATCH + PUT from scratch), but each retry has the same 10-minute budget. For very large layers on constrained networks, pre-stage to a faster mirror or set a higher `RetryConfig.max_retries`. +- 10 GB per-layer cap. Layers larger than this fail at finalize regardless of the timeout above; split the image or use a different registry. ## CLI example diff --git a/src/cli/commands/copy.rs b/src/cli/commands/copy.rs index 1ce3f8e9..56c34b5e 100644 --- a/src/cli/commands/copy.rs +++ b/src/cli/commands/copy.rs @@ -6,6 +6,8 @@ use std::rc::Rc; use std::sync::Arc; use ocync_distribution::RepositoryName; +use ocync_distribution::auth::detect::{ProviderKind, detect_provider_kind}; +use ocync_distribution::ecr::{BatchBlobChecker, BatchChecker}; use ocync_sync::cache::TransferStateCache; use ocync_sync::engine::{ DEFAULT_MAX_CONCURRENT_TRANSFERS, RegistryAlias, ResolvedArtifacts, ResolvedMapping, @@ -16,7 +18,7 @@ use ocync_sync::shutdown::ShutdownSignal; use ocync_sync::staging::BlobStage; use crate::CopyArgs; -use crate::cli::config::load_config; +use crate::cli::config::{AuthType, load_config}; use crate::cli::{CliError, ExitCode, bare_hostname, build_registry_client, endpoint_host}; /// Run the copy command: transfer a single image from source to destination. @@ -63,20 +65,57 @@ pub(crate) async fn run( .registry_authority() .map_err(|e| CliError::Input(format!("source '{}': {e}", args.source)))?; + // Build an ECR batch checker for the destination when applicable. Without + // this, single-image copy falls back to per-blob HEAD against ECR, whose + // HEAD returns false-positive 200s under concurrency; only + // `BatchCheckLayerAvailability` is authoritative. + // + // Resolution: an explicit non-Ecr `auth_type` in registry config is a + // hard opt-out -- the user has told us this destination is not ECR even + // though the hostname pattern matches. Only fall through to hostname + // detection when no `auth_type` is set. + let dst_hostname = bare_hostname(args.destination.registry()); + let dst_is_ecr = match dst_reg_config.and_then(|r| r.auth_type.as_ref()) { + Some(AuthType::Ecr) => true, + Some(_) => false, + None => detect_provider_kind(dst_hostname) == Some(ProviderKind::Ecr), + }; + let batch_checker: Option> = if dst_is_ecr { + let profile = dst_reg_config.and_then(|r| r.aws_profile.as_deref()); + let checker = BatchChecker::from_hostname(dst_hostname, profile) + .await + .map_err(|e| CliError::Input(format!("ECR batch checker for '{dst_hostname}': {e}")))?; + Some(Rc::new(checker)) + } else { + None + }; + + // head_first is a source-side optimization: HEAD-check the target + // before pulling the source manifest, skipping the source GET when + // the destination already holds the same digest. Defined on the + // source registry's config because the optimization conserves the + // source registry's rate-limit budget; applies to both `sync` and + // `copy` because the savings shape is the same. + // + // When `--config` is not supplied (one-shot copy by image ref) the + // default is false -- no per-invocation override flag yet; see + // RegistryConfig::head_first for the field doc. + let head_first = src_reg_config.map(|r| r.head_first).unwrap_or(false); + let mapping = ResolvedMapping { source_authority, source_client, source_repo: RepositoryName::new(args.source.repository())?, target_repo: RepositoryName::new(args.destination.repository())?, targets: vec![TargetEntry { - name: RegistryAlias::new(bare_hostname(args.destination.registry())), + name: RegistryAlias::new(dst_hostname), client: target_client, - batch_checker: None, + batch_checker, existing_tags: HashSet::new(), }], tags: vec![TagPair::retag(src_tag.to_owned(), dst_tag.to_owned())], platforms: None, - head_first: false, + head_first, immutable_glob: None, artifacts_config: Rc::new(ResolvedArtifacts { enabled: false, diff --git a/src/cli/commands/synchronize.rs b/src/cli/commands/synchronize.rs index 8d17865e..753be24f 100644 --- a/src/cli/commands/synchronize.rs +++ b/src/cli/commands/synchronize.rs @@ -626,8 +626,15 @@ async fn build_batch_checkers( for (name, reg) in &config.registries { let hostname = bare_hostname(®.url); - let is_ecr = reg.auth_type.as_ref().is_some_and(|a| *a == AuthType::Ecr) - || detect_provider_kind(hostname) == Some(ProviderKind::Ecr); + // Explicit non-Ecr auth_type is a hard opt-out: don't try to build an + // AWS-SDK-backed batch checker for a registry the user has declared + // is not ECR, even if the hostname pattern matches. Only fall + // through to hostname auto-detection when no `auth_type` is set. + let is_ecr = match reg.auth_type.as_ref() { + Some(AuthType::Ecr) => true, + Some(_) => false, + None => detect_provider_kind(hostname) == Some(ProviderKind::Ecr), + }; if !is_ecr { continue; diff --git a/src/cli/config.rs b/src/cli/config.rs index fded2a8f..e833113a 100644 --- a/src/cli/config.rs +++ b/src/cli/config.rs @@ -132,7 +132,12 @@ pub(crate) struct Config { /// Global engine settings that apply across all sync operations. #[derive(Debug, Clone, Deserialize, Serialize, JsonSchema)] pub(crate) struct GlobalConfig { - /// Maximum concurrent image syncs (default: 50). + /// Maximum concurrent image syncs (default: 10). + /// + /// Default derived as `streaming_blob_concurrency / BLOB_CONCURRENCY = 64 + /// / 6`, sized so the per-target HTTP/2 stream budget is not exhausted + /// (typical advertised `SETTINGS_MAX_CONCURRENT_STREAMS` is 100-128 + /// across major container registries). #[serde(default = "default_max_concurrent_transfers")] pub max_concurrent_transfers: usize, @@ -155,7 +160,12 @@ pub(crate) struct GlobalConfig { } fn default_max_concurrent_transfers() -> usize { - 50 + // Mirror the engine's DEFAULT_MAX_CONCURRENT_TRANSFERS. The previous + // value of 50 multiplied by BLOB_CONCURRENCY=6 (= 300 streams) blew + // past the per-connection HTTP/2 stream budget on every major + // registry; see DEFAULT_STREAMING_BLOB_CONCURRENCY in + // ocync-distribution for the derivation. Keep these two in lockstep. + ocync_sync::engine::DEFAULT_MAX_CONCURRENT_TRANSFERS } impl Default for GlobalConfig { @@ -230,6 +240,10 @@ pub(crate) struct RegistryConfig { /// holds the same digest as the source HEAD, the expensive GET is skipped. /// This conserves rate-limit tokens on source registries with aggressive /// quotas (e.g., Docker Hub). + /// + /// Applies to both `sync` (which is the original target) and `copy` + /// (single-image variant); in `copy` the optimization fires when the + /// destination tag already resolves to the same source digest. #[serde(default)] pub head_first: bool, @@ -1599,7 +1613,10 @@ mappings: "#; let config: Config = serde_yaml::from_str(yaml).unwrap(); let global = config.global.as_ref().unwrap(); - assert_eq!(global.max_concurrent_transfers, 50); + assert_eq!( + global.max_concurrent_transfers, + ocync_sync::engine::DEFAULT_MAX_CONCURRENT_TRANSFERS, + ); } #[test]