diff --git a/Cargo.lock b/Cargo.lock index f342b4e..97cd296 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -1211,6 +1211,7 @@ dependencies = [ "git-cache-disk", "git-cache-domain", "git-cache-git", + "git-cache-objectstore", "http 1.4.0", "reqwest", "serde", diff --git a/crates/git-cache-api/Cargo.toml b/crates/git-cache-api/Cargo.toml index dc52f50..5a028db 100644 --- a/crates/git-cache-api/Cargo.toml +++ b/crates/git-cache-api/Cargo.toml @@ -16,9 +16,11 @@ git-cache-core = { path = "../git-cache-core" } git-cache-disk = { path = "../git-cache-disk" } git-cache-domain = { path = "../git-cache-domain" } git-cache-git = { path = "../git-cache-git" } +git-cache-objectstore = { path = "../git-cache-objectstore" } http.workspace = true reqwest.workspace = true serde.workspace = true +serde_json.workspace = true sha2.workspace = true tokio.workspace = true tokio-util.workspace = true @@ -30,5 +32,4 @@ default = [] s3 = ["git-cache-domain/s3"] [dev-dependencies] -serde_json.workspace = true tempfile.workspace = true diff --git a/crates/git-cache-api/src/lib.rs b/crates/git-cache-api/src/lib.rs index 3132112..e339936 100644 --- a/crates/git-cache-api/src/lib.rs +++ b/crates/git-cache-api/src/lib.rs @@ -3,7 +3,7 @@ use axum::extract::{DefaultBodyLimit, Path, Query, State}; use axum::response::{IntoResponse, Response}; use axum::routing::{any, get, post}; use axum::{Json, Router}; -use futures::Stream; +use futures::{Stream, StreamExt}; use git_cache_core::{ AppConfig, BranchName, CommitSha, GitCacheError, MaterializeRequest, RepoKey, Result as CoreResult, Selector, UpstreamAuth, UpstreamAuthorizationMode, @@ -20,6 +20,7 @@ use git_cache_domain::{ UpstreamRefComparison, }; use git_cache_git::UploadPackProcess; +use git_cache_objectstore::lfs_object_key; use http::{header, HeaderMap, Method, StatusCode, Uri}; use serde::{Deserialize, Serialize}; use sha2::{Digest, Sha256}; @@ -30,8 +31,7 @@ use std::sync::atomic::{AtomicBool, AtomicU64, Ordering}; use std::sync::{Arc, Mutex}; use std::task::{Context, Poll}; use std::time::{Duration, Instant}; -use tokio::io::AsyncRead; -use tokio::io::AsyncWriteExt; +use tokio::io::{AsyncRead, AsyncReadExt, AsyncWriteExt}; use tokio::sync::{mpsc, OwnedSemaphorePermit, Semaphore}; use tokio::task::JoinHandle; use tokio::time::Sleep; @@ -767,6 +767,12 @@ async fn git_repo_inner(state: Arc, request: GitRepoRequest) -> Respon )) .into_response(); } + GitRequestType::LfsBatch => { + return lfs_batch_handler(state, request).await; + } + GitRequestType::LfsDownload => { + return lfs_download_handler(state, request).await; + } GitRequestType::Unsupported => { return ApiError::from(GitCacheError::Unsupported(format!( "unsupported git request: {} {path}", @@ -786,10 +792,9 @@ async fn git_repo_inner(state: Arc, request: GitRepoRequest) -> Respon match request_type { GitRequestType::UploadPackRefs => git_repo_get(state, request, started, auth).await, GitRequestType::UploadPack => git_repo_post(state, request, started, auth).await, - GitRequestType::ReceivePack => ApiError::from(GitCacheError::Unsupported( - "git-receive-pack is disabled".into(), - )) - .into_response(), + GitRequestType::ReceivePack | GitRequestType::LfsBatch | GitRequestType::LfsDownload => { + unreachable!("handled above") + } GitRequestType::Unsupported => ApiError::from(GitCacheError::Unsupported(format!( "unsupported git request: {} {path}", request.method @@ -803,6 +808,8 @@ enum GitRequestType { UploadPackRefs, UploadPack, ReceivePack, + LfsBatch, + LfsDownload, Unsupported, } @@ -818,6 +825,17 @@ fn git_request_type(request: &GitRepoRequest) -> GitRequestType { return GitRequestType::ReceivePack; } + if request.method == Method::POST && path.contains(LFS_BATCH_PATH) { + return GitRequestType::LfsBatch; + } + + if request.method == Method::GET + && path.contains(LFS_OBJECTS_PATH) + && !path.contains(LFS_BATCH_PATH) + { + return GitRequestType::LfsDownload; + } + if request.method == Method::GET && path.ends_with("/info/refs") && request @@ -2112,5 +2130,799 @@ impl Stream for ChildGuardStream { } } +// --------------------------------------------------------------------------- +// LFS batch + object download handlers +// --------------------------------------------------------------------------- + +const LFS_CONTENT_TYPE: &str = "application/vnd.git-lfs+json"; +/// Bounded request body for LFS batch POSTs (1 MiB). +const LFS_BATCH_MAX_BODY_BYTES: usize = 1024 * 1024; +/// Upper bound for upstream LFS batch JSON responses (16 MiB). +const LFS_BATCH_MAX_RESPONSE_BYTES: u64 = 16 * 1024 * 1024; +/// Path suffix for the LFS batch API endpoint. +const LFS_BATCH_PATH: &str = "/info/lfs/objects/batch"; +/// Path infix for individual LFS object downloads. +const LFS_OBJECTS_PATH: &str = "/info/lfs/objects/"; + +#[derive(Debug, Clone, Copy, PartialEq, Eq, Deserialize)] +#[serde(rename_all = "lowercase")] +enum LfsBatchOperation { + Download, + Upload, +} + +#[derive(Debug, Deserialize)] +struct LfsBatchRequest { + operation: LfsBatchOperation, + #[serde(default)] + objects: Vec, +} + +#[derive(Debug, Deserialize)] +struct LfsBatchObject { + oid: String, + size: u64, +} + +#[derive(Debug, Serialize)] +struct LfsBatchResponse { + transfer: &'static str, + objects: Vec, +} + +#[derive(Debug, Serialize)] +struct LfsBatchObjectResponse { + oid: String, + size: u64, + #[serde(skip_serializing_if = "Option::is_none")] + actions: Option, + #[serde(skip_serializing_if = "Option::is_none")] + error: Option, +} + +#[derive(Debug, Serialize)] +struct LfsBatchActions { + download: LfsBatchAction, +} + +#[derive(Debug, Serialize)] +struct LfsBatchAction { + href: String, +} + +#[derive(Debug, Serialize)] +struct LfsBatchError { + code: u16, + message: String, +} + +fn validate_lfs_oid(oid: &str) -> bool { + oid.len() == 64 && oid.bytes().all(|b| b.is_ascii_hexdigit()) +} + +fn lfs_base_url( + headers: &HeaderMap, + bind_addr: &std::net::SocketAddr, + public_path_prefix: &str, +) -> String { + let scheme = headers + .get("x-forwarded-proto") + .and_then(|v| v.to_str().ok()) + .unwrap_or("http"); + let origin = match headers.get(header::HOST).and_then(|v| v.to_str().ok()) { + Some(host) => format!("{scheme}://{host}"), + None => format!("{scheme}://{bind_addr}"), + }; + if public_path_prefix.is_empty() { + origin + } else { + format!("{origin}{public_path_prefix}") + } +} + +async fn lfs_batch_handler(state: Arc, request: GitRepoRequest) -> Response { + let request_id = request.request_id; + let repo = match repo_from_git_path(&request.repo_path) { + Ok(repo) => repo, + Err(error) => return ApiError::from(error).into_response(), + }; + + let materializer = Materializer::new(Arc::clone(&state.domain)); + if let Err(error) = materializer.validate_host(&repo) { + return ApiError::from(error).into_response(); + } + + if request.body.len() > LFS_BATCH_MAX_BODY_BYTES { + return ApiError::from(GitCacheError::Validation(format!( + "LFS batch request body exceeds {} byte limit", + LFS_BATCH_MAX_BODY_BYTES + ))) + .into_response(); + } + + let batch: LfsBatchRequest = match serde_json::from_slice(&request.body) { + Ok(batch) => batch, + Err(error) => { + return ApiError::from(GitCacheError::Validation(format!( + "invalid LFS batch request body: {error}" + ))) + .into_response(); + } + }; + + match batch.operation { + LfsBatchOperation::Download => {} + LfsBatchOperation::Upload => { + return ApiError::from(GitCacheError::Unsupported( + "LFS upload is not supported; cache is read-only".into(), + )) + .into_response(); + } + } + + let max_object_bytes = state.domain.config.lfs.max_object_bytes; + let auth = match direct_git_upstream_auth(&request.headers) { + Ok(auth) => auth, + Err(error) => return error.into_response(), + }; + + info!( + request_id, + repo = %repo, + objects = batch.objects.len(), + "LFS batch download request" + ); + + let upstream_url = match materializer.upstream_url(&repo) { + Ok(url) => url, + Err(error) => return ApiError::from(error).into_response(), + }; + + // Pre-validate objects: collect valid ones for the upstream batch while + // recording which indices failed so the final response preserves request order. + let mut valid_objects: Vec<&LfsBatchObject> = Vec::new(); + let mut pre_validation_errors: std::collections::HashMap = + std::collections::HashMap::new(); + for (i, obj) in batch.objects.iter().enumerate() { + if !validate_lfs_oid(&obj.oid) { + pre_validation_errors.insert( + i, + LfsBatchError { + code: 422, + message: "invalid LFS OID".into(), + }, + ); + } else if obj.size > max_object_bytes { + pre_validation_errors.insert( + i, + LfsBatchError { + code: 422, + message: format!("object exceeds {max_object_bytes} byte limit"), + }, + ); + } else { + valid_objects.push(obj); + } + } + + let objects_json: Vec = valid_objects + .iter() + .map(|o| serde_json::json!({"oid": o.oid, "size": o.size})) + .collect(); + + // Skip upstream call when all objects failed pre-validation. + if valid_objects.is_empty() { + let objects: Vec = batch + .objects + .iter() + .enumerate() + .map(|(i, obj)| LfsBatchObjectResponse { + oid: obj.oid.clone(), + size: obj.size, + actions: None, + error: pre_validation_errors.remove(&i), + }) + .collect(); + let resp = LfsBatchResponse { + transfer: "basic", + objects, + }; + return Response::builder() + .status(StatusCode::OK) + .header(header::CONTENT_TYPE, LFS_CONTENT_TYPE) + .body(Body::from(serde_json::to_vec(&resp).unwrap_or_default())) + .expect("LFS batch response"); + } + + let upstream_batch = match lfs_upstream_batch( + &state.upstream_http, + &upstream_url, + &auth, + &objects_json, + request_id, + &repo, + ) + .await + { + Ok(v) => v, + Err(resp) => return resp, + }; + let upstream_objects: Vec = upstream_batch["objects"] + .as_array() + .cloned() + .unwrap_or_default(); + let upstream_by_oid: std::collections::HashMap<&str, &serde_json::Value> = upstream_objects + .iter() + .filter_map(|obj| obj["oid"].as_str().map(|oid| (oid, obj))) + .collect(); + + let base_url = lfs_base_url( + &request.headers, + &state.domain.config.bind_addr, + &state.domain.config.public_path_prefix, + ); + + // Build response in request order so positional correspondence is preserved. + let store = &state.domain.store; + let mut response_objects = Vec::with_capacity(batch.objects.len()); + for (i, req_obj) in batch.objects.iter().enumerate() { + let oid = &req_obj.oid; + let size = req_obj.size; + + // Pre-validation failure — emit error and continue. + if let Some(error) = pre_validation_errors.remove(&i) { + response_objects.push(LfsBatchObjectResponse { + oid: oid.clone(), + size, + actions: None, + error: Some(error), + }); + continue; + } + + let upstream_obj = match upstream_by_oid.get(oid.as_str()) { + Some(obj) => *obj, + None => { + response_objects.push(LfsBatchObjectResponse { + oid: oid.clone(), + size, + actions: None, + error: Some(LfsBatchError { + code: 404, + message: "object not found in upstream response".into(), + }), + }); + continue; + } + }; + + // Check if the upstream returned an error for this object. + if let Some(err) = upstream_obj.get("error") { + response_objects.push(LfsBatchObjectResponse { + oid: oid.clone(), + size, + actions: None, + error: Some(LfsBatchError { + code: err["code"].as_u64().unwrap_or(404) as u16, + message: err["message"] + .as_str() + .unwrap_or("upstream error") + .to_string(), + }), + }); + continue; + } + + // Check cache. On miss, fetch from upstream and store. + let obj_key = match lfs_object_key(&repo, oid) { + Ok(key) => key, + Err(error) => { + response_objects.push(LfsBatchObjectResponse { + oid: oid.clone(), + size, + actions: None, + error: Some(LfsBatchError { + code: 500, + message: error.to_string(), + }), + }); + continue; + } + }; + + let cached = match store.exists(&obj_key).await { + Ok(exists) => exists, + Err(error) => { + warn!( + request_id, + repo = %repo, + oid = %oid, + error = %error, + "LFS cache check failed" + ); + false + } + }; + + if !cached { + // Extract the upstream download action. + let download_action = upstream_obj.get("actions").and_then(|a| a.get("download")); + let upstream_href = download_action + .and_then(|d| d.get("href")) + .and_then(|h| h.as_str()); + let action_headers = extract_lfs_action_headers(download_action); + + match upstream_href { + Some(href) => { + // Fetch from upstream and store (proxy-tee). + match fetch_and_cache_lfs_object( + &state.upstream_http, + href, + &action_headers, + store.as_ref(), + &obj_key, + max_object_bytes, + ) + .await + { + Ok(_content_length) => { + info!( + request_id, + repo = %repo, + oid = %oid, + size, + "LFS object cached from upstream" + ); + } + Err(error) => { + warn!( + request_id, + repo = %repo, + oid = %oid, + error = %error, + "LFS object cache-fill failed" + ); + response_objects.push(LfsBatchObjectResponse { + oid: oid.clone(), + size, + actions: None, + error: Some(LfsBatchError { + code: 502, + message: format!("cache-fill failed: {error}"), + }), + }); + continue; + } + } + } + None => { + response_objects.push(LfsBatchObjectResponse { + oid: oid.clone(), + size, + actions: None, + error: Some(LfsBatchError { + code: 404, + message: "object not available upstream".into(), + }), + }); + continue; + } + } + } + + let download_url = format!( + "{}/git/{}.git{}{}", + base_url, + repo.as_str(), + LFS_OBJECTS_PATH, + oid + ); + + response_objects.push(LfsBatchObjectResponse { + oid: oid.clone(), + size, + actions: Some(LfsBatchActions { + download: LfsBatchAction { href: download_url }, + }), + error: None, + }); + } + + info!( + request_id, + repo = %repo, + objects = response_objects.len(), + "LFS batch download response" + ); + + let resp = LfsBatchResponse { + transfer: "basic", + objects: response_objects, + }; + + Response::builder() + .status(StatusCode::OK) + .header(header::CONTENT_TYPE, LFS_CONTENT_TYPE) + .body(Body::from(serde_json::to_vec(&resp).unwrap_or_default())) + .expect("LFS batch response") +} + +/// Constructs the upstream LFS batch URL from an upstream repo URL. +/// Returns `None` for non-HTTP(S) upstreams (e.g. local filesystem). +fn lfs_batch_url(upstream_url: &str) -> Option { + if !(upstream_url.starts_with("https://") || upstream_url.starts_with("http://")) { + return None; + } + Some(format!( + "{}{}", + upstream_url.trim_end_matches('/'), + LFS_BATCH_PATH + )) +} + +/// Sends a download batch request to the upstream LFS server and returns the +/// parsed JSON response. On failure returns a ready-made error `Response`. +async fn lfs_upstream_batch( + http: &reqwest::Client, + upstream_url: &str, + auth: &UpstreamAuth, + objects: &[serde_json::Value], + request_id: u64, + repo: &git_cache_core::RepoKey, +) -> Result { + let url = match lfs_batch_url(upstream_url) { + Some(url) => url, + None => { + return Err(ApiError::from(GitCacheError::Unsupported( + "LFS requires an HTTP(S) upstream; local filesystem upstreams do not support LFS" + .into(), + )) + .into_response()); + } + }; + let mut req = http + .post(&url) + .header(header::CONTENT_TYPE.as_str(), LFS_CONTENT_TYPE) + .header(header::ACCEPT.as_str(), LFS_CONTENT_TYPE); + if let Some(raw_auth) = auth.raw_header() { + req = req.header(header::AUTHORIZATION.as_str(), raw_auth); + } + req = req.json(&serde_json::json!({ + "operation": "download", + "transfers": ["basic"], + "objects": objects, + })); + + let resp = match req.send().await { + Ok(r) => r, + Err(error) => { + warn!( + request_id, + repo = %repo, + error = %error, + "LFS batch upstream request failed" + ); + return Err(ApiError::from(GitCacheError::UpstreamUnavailable(format!( + "LFS batch upstream request failed: {error}" + ))) + .into_response()); + } + }; + + if !resp.status().is_success() { + let status = resp.status().as_u16(); + warn!( + request_id, + repo = %repo, + upstream_status = status, + "LFS batch upstream returned error" + ); + return Err(ApiError::from(GitCacheError::UpstreamUnavailable(format!( + "LFS batch upstream returned HTTP {status}" + ))) + .into_response()); + } + + if resp.content_length().unwrap_or(0) > LFS_BATCH_MAX_RESPONSE_BYTES { + return Err(ApiError::from(GitCacheError::Validation(format!( + "upstream LFS batch response exceeds {} byte limit", + LFS_BATCH_MAX_RESPONSE_BYTES + ))) + .into_response()); + } + + // Read the response through a bounded reader so chunked responses (no + // Content-Length) cannot consume unbounded memory. + let body_stream = resp.bytes_stream(); + let reader = tokio_util::io::StreamReader::new( + body_stream.map(|result| result.map_err(std::io::Error::other)), + ); + let mut bounded = reader.take(LFS_BATCH_MAX_RESPONSE_BYTES + 1); + let mut buf = + Vec::with_capacity(std::cmp::min(LFS_BATCH_MAX_RESPONSE_BYTES, 1024 * 1024) as usize); + match tokio::io::AsyncReadExt::read_to_end(&mut bounded, &mut buf).await { + Ok(n) if n as u64 > LFS_BATCH_MAX_RESPONSE_BYTES => { + return Err(ApiError::from(GitCacheError::Validation(format!( + "upstream LFS batch response exceeds {} byte limit", + LFS_BATCH_MAX_RESPONSE_BYTES + ))) + .into_response()); + } + Ok(_) => {} + Err(error) => { + return Err(ApiError::from(GitCacheError::UpstreamUnavailable(format!( + "failed to read LFS batch upstream response: {error}" + ))) + .into_response()); + } + } + let bytes = buf; + + serde_json::from_slice(&bytes).map_err(|error| { + ApiError::from(GitCacheError::UpstreamUnavailable(format!( + "invalid upstream LFS batch response: {error}" + ))) + .into_response() + }) +} + +fn extract_lfs_oid_from_path(path: &str) -> Option<&str> { + let idx = path.find(LFS_OBJECTS_PATH)?; + let after = &path[idx + LFS_OBJECTS_PATH.len()..]; + // Reject anything after the OID (e.g. trailing slashes or extra segments). + let oid = after.split('/').next()?; + if oid.is_empty() { + return None; + } + // Strip .git suffix if present. + let oid = oid.strip_suffix(".git").unwrap_or(oid); + Some(oid) +} + +async fn lfs_download_handler(state: Arc, request: GitRepoRequest) -> Response { + let request_id = request.request_id; + let path = request.uri.path().to_string(); + let repo = match repo_from_git_path(&request.repo_path) { + Ok(repo) => repo, + Err(error) => return ApiError::from(error).into_response(), + }; + + let materializer = Materializer::new(Arc::clone(&state.domain)); + if let Err(error) = materializer.validate_host(&repo) { + return ApiError::from(error).into_response(); + } + + let oid = match extract_lfs_oid_from_path(&path) { + Some(oid) if validate_lfs_oid(oid) => oid.to_string(), + _ => { + return ApiError::from(GitCacheError::Validation( + "invalid or missing LFS OID in download path".into(), + )) + .into_response(); + } + }; + + let obj_key = match lfs_object_key(&repo, &oid) { + Ok(key) => key, + Err(error) => return ApiError::from(error).into_response(), + }; + + let store = &state.domain.store; + let max_object_bytes = state.domain.config.lfs.max_object_bytes; + + // Try streaming from cache. + match store.get_stream(&obj_key).await { + Ok(Some((reader, len))) => { + info!( + request_id, + repo = %repo, + oid = %oid, + size = len, + "LFS object served from cache (streaming)" + ); + let capped_len = len.min(max_object_bytes); + let stream = ReaderStream::new(reader.take(max_object_bytes)); + return Response::builder() + .status(StatusCode::OK) + .header(header::CONTENT_TYPE, "application/octet-stream") + .header(header::CONTENT_LENGTH, capped_len) + .body(Body::from_stream(stream)) + .expect("LFS download response"); + } + Ok(None) => { + info!( + request_id, + repo = %repo, + oid = %oid, + "LFS object not in cache; proxying from upstream" + ); + } + Err(error) => { + warn!( + request_id, + repo = %repo, + oid = %oid, + error = %error, + "LFS cache read failed; proxying from upstream" + ); + } + } + + // Cache miss — proxy from upstream LFS server. + let auth = match direct_git_upstream_auth(&request.headers) { + Ok(auth) => auth, + Err(error) => return error.into_response(), + }; + + let upstream_url = match materializer.upstream_url(&repo) { + Ok(url) => url, + Err(error) => return ApiError::from(error).into_response(), + }; + + // Resolve upstream download URL via a batch request. + let objects_json = vec![serde_json::json!({"oid": oid, "size": 0})]; + let upstream_batch = match lfs_upstream_batch( + &state.upstream_http, + &upstream_url, + &auth, + &objects_json, + request_id, + &repo, + ) + .await + { + Ok(v) => v, + Err(resp) => return resp, + }; + + let download_action = upstream_batch["objects"] + .as_array() + .and_then(|arr| arr.first()) + .and_then(|obj| obj.get("actions")) + .and_then(|a| a.get("download")); + let upstream_href = download_action + .and_then(|d| d.get("href")) + .and_then(|h| h.as_str()); + let action_headers = extract_lfs_action_headers(download_action); + + let Some(href) = upstream_href else { + return ApiError::from(GitCacheError::NotFound(format!( + "LFS object {oid} not found upstream" + ))) + .into_response(); + }; + + // Fetch from upstream, cache (streaming), and serve. + let fetched = fetch_and_cache_lfs_object( + &state.upstream_http, + href, + &action_headers, + store.as_ref(), + &obj_key, + max_object_bytes, + ) + .await; + + match fetched { + Ok(content_length) => { + // Object is now cached — stream it back from the store. + match store.get_stream(&obj_key).await { + Ok(Some((reader, len))) => { + info!( + request_id, + repo = %repo, + oid = %oid, + size = len, + "LFS object served after proxy-tee (streaming)" + ); + let capped_len = len.min(max_object_bytes); + let stream = ReaderStream::new(reader.take(max_object_bytes)); + Response::builder() + .status(StatusCode::OK) + .header(header::CONTENT_TYPE, "application/octet-stream") + .header(header::CONTENT_LENGTH, capped_len) + .body(Body::from_stream(stream)) + .expect("LFS download response") + } + _ => { + warn!( + request_id, + repo = %repo, + oid = %oid, + "LFS cache read-back failed after proxy-tee ({content_length} bytes cached)" + ); + ApiError::from(GitCacheError::UpstreamUnavailable(format!( + "LFS object {oid} cached but read-back failed" + ))) + .into_response() + } + } + } + Err(error) => { + warn!( + request_id, + repo = %repo, + oid = %oid, + error = %error, + "LFS object fetch failed" + ); + ApiError::from(GitCacheError::UpstreamUnavailable(format!( + "LFS object {oid} could not be retrieved" + ))) + .into_response() + } + } +} + +/// Fetches an LFS object from upstream and streams it into the object store +/// without buffering the full body in memory. Returns the number of bytes +/// cached on success. +/// Extracts the optional `header` map from an LFS download action JSON object. +fn extract_lfs_action_headers(action: Option<&serde_json::Value>) -> Vec<(String, String)> { + let Some(header_obj) = action + .and_then(|a| a.get("header")) + .and_then(|h| h.as_object()) + else { + return Vec::new(); + }; + header_obj + .iter() + .filter_map(|(k, v)| v.as_str().map(|v| (k.clone(), v.to_string()))) + .collect() +} + +async fn fetch_and_cache_lfs_object( + http: &reqwest::Client, + href: &str, + action_headers: &[(String, String)], + store: &dyn git_cache_objectstore::ObjectStore, + obj_key: &str, + max_bytes: u64, +) -> CoreResult { + let mut req = http.get(href); + for (key, value) in action_headers { + req = req.header(key.as_str(), value.as_str()); + } + let response = req + .send() + .await + .map_err(|e| GitCacheError::UpstreamUnavailable(format!("LFS download failed: {e}")))?; + + if !response.status().is_success() { + return Err(GitCacheError::UpstreamUnavailable(format!( + "LFS download returned HTTP {}", + response.status() + ))); + } + + let content_length = response.content_length().unwrap_or(0); + if content_length > max_bytes { + return Err(GitCacheError::Validation(format!( + "LFS object exceeds {max_bytes} byte limit (content-length: {content_length})" + ))); + } + + // Wrap the response body as an AsyncRead with a byte-limit guard so the + // store never ingests more than max_bytes. + let body_stream = response.bytes_stream(); + let reader = tokio_util::io::StreamReader::new( + body_stream.map(|result| result.map_err(std::io::Error::other)), + ); + let reader = tokio::io::BufReader::new(reader.take(max_bytes)); + + store + .put_stream(obj_key, Box::pin(reader), content_length) + .await + .map_err(|e| GitCacheError::Internal(format!("LFS cache write failed: {e}")))?; + + // Return the actual stored size rather than the declared Content-Length, + // which may be 0 for chunked transfers. + let stored_size = match store.head(obj_key).await { + Ok(Some(meta)) => meta.len, + _ => content_length, + }; + Ok(stored_size) +} + #[cfg(test)] mod tests; diff --git a/crates/git-cache-api/src/tests.rs b/crates/git-cache-api/src/tests.rs index f00958a..5342b27 100644 --- a/crates/git-cache-api/src/tests.rs +++ b/crates/git-cache-api/src/tests.rs @@ -258,9 +258,11 @@ async fn proxy_warm_task_queues_async_generation_materialize() { }, git_remote: Default::default(), compaction: Default::default(), + lfs: Default::default(), shutdown: Default::default(), max_concurrent_git_processes: git_cache_core::default_max_concurrent_git_processes(), async_materialize_concurrency: 1, + public_path_prefix: String::new(), use_gitoxide: true, }; let state = Arc::new(ApiState::try_new(config).unwrap()); @@ -386,9 +388,11 @@ async fn proxy_warm_task_publishes_served_commit_when_branch_moves() { }, git_remote: Default::default(), compaction: Default::default(), + lfs: Default::default(), shutdown: Default::default(), max_concurrent_git_processes: git_cache_core::default_max_concurrent_git_processes(), async_materialize_concurrency: 1, + public_path_prefix: String::new(), use_gitoxide: true, }; let state = Arc::new(ApiState::try_new(config).unwrap()); @@ -476,9 +480,11 @@ async fn authenticated_resolve_is_rate_limited_before_upstream_work() { }, git_remote: Default::default(), compaction: Default::default(), + lfs: Default::default(), shutdown: Default::default(), max_concurrent_git_processes: git_cache_core::default_max_concurrent_git_processes(), async_materialize_concurrency: git_cache_core::default_async_materialize_concurrency(), + public_path_prefix: String::new(), use_gitoxide: true, }; let state = Arc::new(ApiState::try_new(config).unwrap()); @@ -524,9 +530,11 @@ async fn healthz_fails_after_shutdown_begins() { }, git_remote: Default::default(), compaction: Default::default(), + lfs: Default::default(), shutdown: Default::default(), max_concurrent_git_processes: git_cache_core::default_max_concurrent_git_processes(), async_materialize_concurrency: git_cache_core::default_async_materialize_concurrency(), + public_path_prefix: String::new(), use_gitoxide: true, }; let state = Arc::new(ApiState::try_new(config).unwrap()); diff --git a/crates/git-cache-api/tests/support/mod.rs b/crates/git-cache-api/tests/support/mod.rs index cb9478b..0eab757 100644 --- a/crates/git-cache-api/tests/support/mod.rs +++ b/crates/git-cache-api/tests/support/mod.rs @@ -36,9 +36,11 @@ pub fn test_config_with_upstream( ..Default::default() }, compaction: Default::default(), + lfs: Default::default(), shutdown: Default::default(), max_concurrent_git_processes: git_cache_core::default_max_concurrent_git_processes(), async_materialize_concurrency: git_cache_core::default_async_materialize_concurrency(), + public_path_prefix: String::new(), use_gitoxide: true, } } diff --git a/crates/git-cache-core/src/config.rs b/crates/git-cache-core/src/config.rs index ed3f387..f7b7e0a 100644 --- a/crates/git-cache-core/src/config.rs +++ b/crates/git-cache-core/src/config.rs @@ -29,11 +29,15 @@ pub struct AppConfig { #[serde(default)] pub compaction: CompactionConfig, #[serde(default)] + pub lfs: LfsConfig, + #[serde(default)] pub shutdown: ShutdownConfig, #[serde(default = "default_max_concurrent_git_processes")] pub max_concurrent_git_processes: usize, #[serde(default = "default_async_materialize_concurrency")] pub async_materialize_concurrency: usize, + #[serde(default)] + pub public_path_prefix: String, /// Use in-process gitoxide for local read-only Git operations instead of /// spawning the `git` binary. Acts as a kill switch when disabled. #[serde(default = "default_use_gitoxide")] @@ -111,6 +115,12 @@ impl AppConfig { )?, proxy_tee_import: parse_bool_env("GIT_CACHE_GIT_REMOTE_PROXY_TEE_IMPORT", true)?, }, + lfs: LfsConfig { + max_object_bytes: parse_env( + "GIT_CACHE_LFS_MAX_OBJECT_BYTES", + default_lfs_max_object_bytes(), + )?, + }, compaction: CompactionConfig { chain_depth_threshold: parse_env( "GIT_CACHE_COMPACTION_CHAIN_DEPTH_THRESHOLD", @@ -132,6 +142,7 @@ impl AppConfig { default_shutdown_drain_timeout_seconds(), )?, }, + public_path_prefix: env::var("GIT_CACHE_PUBLIC_PATH_PREFIX").unwrap_or_default(), max_concurrent_git_processes: parse_env( "GIT_CACHE_MAX_CONCURRENT_GIT_PROCESSES", default_max_concurrent_git_processes(), @@ -288,6 +299,25 @@ impl Default for GitRemoteConfig { } } +#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] +pub struct LfsConfig { + #[serde(default = "default_lfs_max_object_bytes")] + pub max_object_bytes: u64, +} + +impl Default for LfsConfig { + fn default() -> Self { + Self { + max_object_bytes: default_lfs_max_object_bytes(), + } + } +} + +fn default_lfs_max_object_bytes() -> u64 { + // 2 GiB + 2 * 1024 * 1024 * 1024 +} + fn default_true() -> bool { true } diff --git a/crates/git-cache-core/src/config/tests.rs b/crates/git-cache-core/src/config/tests.rs index 2a53c78..6657f6b 100644 --- a/crates/git-cache-core/src/config/tests.rs +++ b/crates/git-cache-core/src/config/tests.rs @@ -34,6 +34,8 @@ const ENV_KEYS: &[&str] = &[ "GIT_CACHE_MAX_CONCURRENT_GIT_PROCESSES", "GIT_CACHE_ASYNC_MATERIALIZE_CONCURRENCY", "GIT_CACHE_USE_GITOXIDE", + "GIT_CACHE_LFS_MAX_OBJECT_BYTES", + "GIT_CACHE_PUBLIC_PATH_PREFIX", ]; struct EnvGuard { diff --git a/crates/git-cache-core/src/lib.rs b/crates/git-cache-core/src/lib.rs index 2d6c315..ce59ca4 100644 --- a/crates/git-cache-core/src/lib.rs +++ b/crates/git-cache-core/src/lib.rs @@ -20,7 +20,7 @@ pub const GIT_UPLOAD_PACK_RESULT_CONTENT_TYPE: &str = "application/x-git-upload- pub use auth::{SecretString, UpstreamAuth, UpstreamAuthorizationMode}; pub use config::{ default_async_materialize_concurrency, default_max_concurrent_git_processes, AppConfig, - CompactionConfig, DiskConfig, GitRemoteConfig, ObjectStoreConfig, ShutdownConfig, + CompactionConfig, DiskConfig, GitRemoteConfig, LfsConfig, ObjectStoreConfig, ShutdownConfig, }; pub use error::{GitCacheError, Result}; pub use manifest::{ diff --git a/crates/git-cache-domain/src/materializer/tests/mod.rs b/crates/git-cache-domain/src/materializer/tests/mod.rs index 6087256..28c4ed2 100644 --- a/crates/git-cache-domain/src/materializer/tests/mod.rs +++ b/crates/git-cache-domain/src/materializer/tests/mod.rs @@ -138,9 +138,11 @@ impl GitFixture { }, git_remote: Default::default(), compaction: Default::default(), + lfs: Default::default(), shutdown: Default::default(), max_concurrent_git_processes: git_cache_core::default_max_concurrent_git_processes(), async_materialize_concurrency: 2, + public_path_prefix: String::new(), use_gitoxide: true, } } diff --git a/crates/git-cache-domain/src/state/tests.rs b/crates/git-cache-domain/src/state/tests.rs index f587242..d4a5516 100644 --- a/crates/git-cache-domain/src/state/tests.rs +++ b/crates/git-cache-domain/src/state/tests.rs @@ -64,9 +64,11 @@ fn test_config(cache_root: PathBuf, object_root: PathBuf) -> AppConfig { }, git_remote: Default::default(), compaction: Default::default(), + lfs: Default::default(), shutdown: Default::default(), max_concurrent_git_processes: 1, async_materialize_concurrency: 2, + public_path_prefix: String::new(), use_gitoxide: true, } } diff --git a/crates/git-cache-fuzz/tests/materializer_fuzz.rs b/crates/git-cache-fuzz/tests/materializer_fuzz.rs index 3b51ac6..b109711 100644 --- a/crates/git-cache-fuzz/tests/materializer_fuzz.rs +++ b/crates/git-cache-fuzz/tests/materializer_fuzz.rs @@ -54,9 +54,11 @@ impl Fixture { }, git_remote: Default::default(), compaction: Default::default(), + lfs: Default::default(), shutdown: Default::default(), max_concurrent_git_processes: 4, async_materialize_concurrency: 2, + public_path_prefix: String::new(), use_gitoxide: true, } } diff --git a/crates/git-cache-objectstore/src/lib.rs b/crates/git-cache-objectstore/src/lib.rs index f434b9e..b03ae9c 100644 --- a/crates/git-cache-objectstore/src/lib.rs +++ b/crates/git-cache-objectstore/src/lib.rs @@ -9,17 +9,19 @@ use bytes::Bytes; use chrono::{DateTime, Utc}; use git_cache_core::{GitCacheError, Result}; use std::path::{Component, Path}; +use std::pin::Pin; +use tokio::io::AsyncRead; pub use local::LocalObjectStore; pub use manifests::{ - commit_manifest_key, generation_manifest_key, generation_manifest_prefix, pack_key, - pack_prefix, read_commit_manifest, read_generation_manifest, read_json, read_ref_manifest, - read_repo_generation_head, read_repo_generation_head_versioned, ref_manifest_key, - repo_generation_head_key, write_commit_manifest, write_commit_manifest_if_absent_or_matches, - write_generation_manifest, write_generation_manifest_if_absent_or_matches, write_json, - write_json_if_absent, write_json_if_absent_or_matches, write_ref_manifest, - write_repo_generation_head, write_repo_generation_head_if_version_matches, GenerationPublish, - PublishManifests, + commit_manifest_key, generation_manifest_key, generation_manifest_prefix, lfs_object_key, + pack_key, pack_prefix, read_commit_manifest, read_generation_manifest, read_json, + read_ref_manifest, read_repo_generation_head, read_repo_generation_head_versioned, + ref_manifest_key, repo_generation_head_key, write_commit_manifest, + write_commit_manifest_if_absent_or_matches, write_generation_manifest, + write_generation_manifest_if_absent_or_matches, write_json, write_json_if_absent, + write_json_if_absent_or_matches, write_ref_manifest, write_repo_generation_head, + write_repo_generation_head_if_version_matches, GenerationPublish, PublishManifests, }; #[cfg(feature = "s3")] @@ -89,6 +91,31 @@ pub trait ObjectStore: Send + Sync { let data = tokio::fs::read(path).await?; self.put(key, Bytes::from(data)).await } + + /// Stream an object's bytes without loading the entire object into memory. + /// Returns `None` if the key does not exist. + async fn get_stream(&self, key: &str) -> Result>, u64)>> { + let Some(data) = self.get(key).await? else { + return Ok(None); + }; + let len = data.len() as u64; + Ok(Some((Box::pin(std::io::Cursor::new(data)), len))) + } + + /// Write an object from a streaming reader without accumulating the full + /// body in memory. `content_length` is the expected total size. + async fn put_stream( + &self, + key: &str, + reader: Pin>, + content_length: u64, + ) -> Result<()> { + use tokio::io::AsyncReadExt; + let mut reader = reader; + let mut buf = Vec::with_capacity(content_length.min(64 * 1024 * 1024) as usize); + reader.read_to_end(&mut buf).await?; + self.put(key, Bytes::from(buf)).await + } } pub(crate) fn validate_key(key: &str) -> Result<()> { diff --git a/crates/git-cache-objectstore/src/local.rs b/crates/git-cache-objectstore/src/local.rs index 90a67c7..6823a91 100644 --- a/crates/git-cache-objectstore/src/local.rs +++ b/crates/git-cache-objectstore/src/local.rs @@ -5,10 +5,11 @@ use chrono::{DateTime, Utc}; use git_cache_core::{GitCacheError, Result}; use std::hash::{Hash, Hasher}; use std::path::{Path, PathBuf}; +use std::pin::Pin; use std::sync::OnceLock; use std::time::{SystemTime, UNIX_EPOCH}; use tokio::fs; -use tokio::io::AsyncWriteExt; +use tokio::io::{AsyncRead, AsyncWriteExt}; #[derive(Debug, Clone)] pub struct LocalObjectStore { @@ -213,6 +214,47 @@ impl ObjectStore for LocalObjectStore { } } } + + async fn get_stream(&self, key: &str) -> Result>, u64)>> { + let path = self.object_path(key)?; + match fs::metadata(&path).await { + Ok(metadata) if metadata.is_file() => { + let file = fs::File::open(&path).await?; + Ok(Some((Box::pin(file), metadata.len()))) + } + Ok(_) => Ok(None), + Err(err) if err.kind() == std::io::ErrorKind::NotFound => Ok(None), + Err(err) => Err(err.into()), + } + } + + async fn put_stream( + &self, + key: &str, + reader: Pin>, + _content_length: u64, + ) -> Result<()> { + let dest = self.object_path(key)?; + let parent = parent_dir(&dest)?; + fs::create_dir_all(parent).await?; + + let tmp_path = allocate_temp_path(parent, &dest)?; + let mut file = fs::File::create(&tmp_path).await?; + let mut reader = reader; + tokio::io::copy(&mut reader, &mut file).await?; + file.sync_all().await?; + drop(file); + match fs::rename(&tmp_path, &dest).await { + Ok(()) => { + sync_directory(parent)?; + Ok(()) + } + Err(err) => { + let _ = fs::remove_file(&tmp_path).await; + Err(err.into()) + } + } + } } /// Serializes local read-compare-write sequences process-wide. The local diff --git a/crates/git-cache-objectstore/src/manifests.rs b/crates/git-cache-objectstore/src/manifests.rs index c7f00ea..efb6aed 100644 --- a/crates/git-cache-objectstore/src/manifests.rs +++ b/crates/git-cache-objectstore/src/manifests.rs @@ -119,6 +119,25 @@ pub fn generation_manifest_prefix(repo: &RepoKey) -> String { format!("repos/{repo}/generations/") } +pub fn lfs_object_key(repo: &RepoKey, oid: &str) -> Result { + validate_lfs_oid(oid)?; + Ok(format!( + "repos/{repo}/lfs/{}/{}/{}", + &oid[..2], + &oid[2..4], + oid + )) +} + +fn validate_lfs_oid(value: &str) -> Result<()> { + if value.len() != 64 || !value.bytes().all(|byte| byte.is_ascii_hexdigit()) { + return Err(GitCacheError::Validation(format!( + "invalid LFS OID `{value}`" + ))); + } + Ok(()) +} + pub fn pack_prefix(repo: &RepoKey) -> String { format!("repos/{repo}/packs/") } diff --git a/crates/git-cache-objectstore/src/s3.rs b/crates/git-cache-objectstore/src/s3.rs index 3dc7b83..4ccb8ec 100644 --- a/crates/git-cache-objectstore/src/s3.rs +++ b/crates/git-cache-objectstore/src/s3.rs @@ -12,7 +12,8 @@ use bytes::Bytes; use chrono::{DateTime, Utc}; use git_cache_core::{GitCacheError, Result}; use std::path::Path; -use tokio::io::AsyncWriteExt; +use std::pin::Pin; +use tokio::io::{AsyncRead, AsyncReadExt, AsyncWriteExt}; const S3_SINGLE_PUT_LIMIT_BYTES: u64 = 5 * 1024 * 1024 * 1024; const S3_MAX_OBJECT_BYTES: u64 = 5 * 1024 * 1024 * 1024 * 1024; @@ -207,6 +208,97 @@ impl S3ObjectStore { .map_err(|err| s3_error("abort_multipart_upload", s3_key, err))?; Ok(()) } + + async fn put_stream_multipart_inner( + &self, + s3_key: &str, + reader: &mut (dyn AsyncRead + Send + Unpin), + part_size: usize, + upload_id: &str, + ) -> Result<()> { + let mut parts = Vec::new(); + let mut part_number: i32 = 1; + + loop { + let mut buf = vec![0u8; part_size]; + let mut filled = 0; + // Fill the buffer up to part_size before uploading. + while filled < part_size { + match reader.read(&mut buf[filled..]).await? { + 0 => break, + n => filled += n, + } + } + if filled == 0 { + break; + } + buf.truncate(filled); + + let body = ByteStream::new(Bytes::from(buf).into()); + let output = self + .client + .upload_part() + .bucket(&self.bucket) + .key(s3_key) + .upload_id(upload_id) + .part_number(part_number) + .content_length(filled as i64) + .body(body) + .send() + .await + .map_err(|err| s3_error("upload_part", s3_key, err))?; + let e_tag = output + .e_tag() + .ok_or_else(|| { + GitCacheError::UpstreamUnavailable(format!( + "s3 upload_part `{s3_key}` part {part_number} returned no etag" + )) + })? + .to_string(); + parts.push( + CompletedPart::builder() + .part_number(part_number) + .e_tag(e_tag) + .build(), + ); + part_number += 1; + + if filled < part_size { + // Last chunk was smaller than part_size → EOF. + break; + } + } + + if parts.is_empty() { + // Empty stream — upload a zero-length object instead. + self.client + .put_object() + .bucket(&self.bucket) + .key(s3_key) + .body(ByteStream::new(Bytes::new().into())) + .send() + .await + .map_err(|err| s3_error("put_stream empty", s3_key, err))?; + // Abort the unused multipart upload. + let _ = self.abort_multipart_upload(s3_key, upload_id).await; + return Ok(()); + } + + self.client + .complete_multipart_upload() + .bucket(&self.bucket) + .key(s3_key) + .upload_id(upload_id) + .multipart_upload( + CompletedMultipartUpload::builder() + .set_parts(Some(parts)) + .build(), + ) + .send() + .await + .map_err(|err| s3_error("complete_multipart_upload", s3_key, err))?; + Ok(()) + } } #[async_trait] @@ -476,6 +568,94 @@ impl ObjectStore for S3ObjectStore { .map_err(|err| s3_error("put_file", &s3_key, err))?; Ok(()) } + + async fn get_stream(&self, key: &str) -> Result>, u64)>> { + let s3_key = self.s3_key(key)?; + let output = match self + .client + .get_object() + .bucket(&self.bucket) + .key(&s3_key) + .send() + .await + { + Ok(output) => output, + Err(err) if is_not_found(&err) => return Ok(None), + Err(err) => return Err(s3_error("get_stream", &s3_key, err)), + }; + + let len = output.content_length().unwrap_or(0) as u64; + let reader = output.body.into_async_read(); + Ok(Some((Box::pin(reader), len))) + } + + async fn put_stream( + &self, + key: &str, + reader: Pin>, + content_length: u64, + ) -> Result<()> { + let s3_key = self.s3_key(key)?; + let mut reader = reader; + + // Small objects with known size: single PutObject with in-memory buffer. + // When content_length is 0 the size is unknown (e.g. chunked upstream + // response) so we must use the multipart path to avoid buffering an + // arbitrarily large body. + if content_length > 0 && content_length <= S3_MIN_MULTIPART_PART_BYTES { + let mut buf = Vec::with_capacity(content_length as usize); + reader.read_to_end(&mut buf).await?; + let body = ByteStream::new(buf.into()); + self.client + .put_object() + .bucket(&self.bucket) + .key(&s3_key) + .body(body) + .send() + .await + .map_err(|err| s3_error("put_stream", &s3_key, err))?; + return Ok(()); + } + + // Large objects (or unknown size): multipart upload streaming from reader. + let effective_length = if content_length == 0 { + S3_DEFAULT_MULTIPART_PART_BYTES * 2 + } else { + content_length + }; + let part_size = multipart_part_size(effective_length)? as usize; + let create = self + .client + .create_multipart_upload() + .bucket(&self.bucket) + .key(&s3_key) + .send() + .await + .map_err(|err| s3_error("create_multipart_upload", &s3_key, err))?; + let upload_id = create + .upload_id() + .ok_or_else(|| { + GitCacheError::UpstreamUnavailable(format!( + "s3 create_multipart_upload `{s3_key}` returned no upload id" + )) + })? + .to_string(); + + let result = self + .put_stream_multipart_inner(&s3_key, &mut reader, part_size, &upload_id) + .await; + + if let Err(err) = result { + return match self.abort_multipart_upload(&s3_key, &upload_id).await { + Ok(()) => Err(err), + Err(abort_err) => Err(GitCacheError::UpstreamUnavailable(format!( + "{err}; additionally failed to abort multipart upload `{s3_key}`: {abort_err}" + ))), + }; + } + + Ok(()) + } } fn normalize_prefix(prefix: String) -> Result { diff --git a/integration_tests/README.md b/integration_tests/README.md index fdf9b1d..577352c 100644 --- a/integration_tests/README.md +++ b/integration_tests/README.md @@ -85,6 +85,8 @@ By default it covers `astral-sh/uv`, `astral-sh/ruff`, `torvalds/linux`, and `GITHUB_TOKEN`, `GH_TOKEN`, or `gh auth token` is available - blobless-to-full depth-1 transition checks for `uv` and `ruff` - `git-receive-pack` rejection +- LFS batch API cold/warm latency, object download cold/warm with throughput, + multi-object batch, upload rejection, invalid OID rejection (for LFS repos) `GIT_CACHE_AWS_DEV_RESET_LOCAL_CACHE=1` uses `scripts/aws/remove-cache-repo.sh` to delete only the local EBS hot-cache repo @@ -111,6 +113,29 @@ HEAD plus a bounded history walk, and also runs a blobless full checkout for `uv` and `ruff`. Add `GIT_CACHE_AWS_DEV_DIRECT_HEAVY_BASELINE=1` to also measure the same heavy shapes directly from GitHub for comparison. +## LFS smoke tests (`test_lfs_smoke`) + +```sh +RUN_GITHUB_INTEGRATION=1 python3 -m unittest -v integration_tests.test_lfs_smoke +``` + +Tests Git LFS caching through the cache: verifies the git protocol layer +works for LFS repos (pointer files clone correctly), the LFS batch API +proxies upstream and returns download URLs, LFS objects are cached in the +object store after first access, and the upstream-URL workaround lets +`git lfs pull` succeed. + +Optional overrides: + +```sh +GIT_CACHE_LFS_TEST_REPO=github.com/charmbracelet/vhs \ +RUN_GITHUB_INTEGRATION=1 \ +python3 -m unittest -v integration_tests.test_lfs_smoke +``` + +Requires `git-lfs` installed (`git lfs install`). The test class is +automatically skipped if `git-lfs` is not available. + ## Docker / MinIO object-store tests These tests use Docker Compose to run MinIO locally and exercise the S3-compatible diff --git a/integration_tests/test_aws_dev_git_matrix.py b/integration_tests/test_aws_dev_git_matrix.py index 2490c11..9ffad4f 100755 --- a/integration_tests/test_aws_dev_git_matrix.py +++ b/integration_tests/test_aws_dev_git_matrix.py @@ -17,6 +17,7 @@ import base64 import json import os +import re import shutil import subprocess import tempfile @@ -37,6 +38,13 @@ ("github.com/llvm/llvm-project", "main"), ] SMALL_REPOS = {"github.com/astral-sh/uv", "github.com/astral-sh/ruff"} +LFS_REPOS = {"github.com/charmbracelet/vhs", "github.com/SixLabors/ImageSharp"} +LFS_CONTENT_TYPE = "application/vnd.git-lfs+json" +LFS_POINTER_RE = re.compile( + r"^version https://git-lfs\.github\.com/spec/v1\n" + r"oid sha256:[0-9a-f]{64}\n" + r"size \d+\n\Z", +) @dataclass(frozen=True) @@ -170,7 +178,14 @@ def setUpClass(cls) -> None: cls.direct_heavy_baseline = env_bool("GIT_CACHE_AWS_DEV_DIRECT_HEAVY_BASELINE", False) cls.reset_local_cache = env_bool("GIT_CACHE_AWS_DEV_RESET_LOCAL_CACHE", False) cls.require_auth = env_bool("GIT_CACHE_AWS_DEV_REQUIRE_AUTH", False) + cls.skip_lfs = env_bool("GIT_CACHE_AWS_DEV_SKIP_LFS", False) cls.repos = parse_repos() + # Append LFS repos unless explicitly skipped or already present. + if not cls.skip_lfs: + existing = {rc.repo for rc in cls.repos} + for lfs_repo in sorted(LFS_REPOS): + if lfs_repo not in existing: + cls.repos.append(RepoCase(lfs_repo, "main")) cls.basic_auth_header = basic_auth_header_from_env() if cls.require_auth and not cls.basic_auth_header: raise RuntimeError("auth is required but no Basic auth token/header is configured") @@ -303,6 +318,8 @@ def run_repo_matrix(self, repo_case: RepoCase) -> None: if repo_case.repo in SMALL_REPOS and not self.skip_standard: self.blobless_then_full_depth1_transition(repo_case, upstream_head) + if repo_case.repo in LFS_REPOS and not self.skip_lfs: + self.run_lfs_matrix(repo_case) if self.tier == "heavy": if self.direct_heavy_baseline: self.heavy_github_direct_baseline(repo_case, upstream_head) @@ -806,6 +823,263 @@ def wait_after_proxy_warm(self, repo: str, label: str) -> None: } ) + # ── LFS matrix ───────────────────────────────────────────────────── + + def run_lfs_matrix(self, repo_case: RepoCase) -> None: + """LFS cache correctness and performance sub-matrix.""" + # 1. Clone with skip-smudge to discover pointer files + clone_dir = self.tmp / safe_name(f"lfs-clone-{repo_case.repo}") + shutil.rmtree(clone_dir, ignore_errors=True) + env = git_env() + env["GIT_LFS_SKIP_SMUDGE"] = "1" + t0 = time.monotonic() + clone_result = run_command( + ["git", "clone", "--depth", "1", self.git_url(repo_case.repo), str(clone_dir)], + env=env, timeout=self.command_timeout, + ) + clone_elapsed = time.monotonic() - t0 + self.record({ + "case": "lfs_clone_skip_smudge", + "repo": repo_case.repo, + "branch": repo_case.branch, + "duration_s": round(clone_elapsed, 3), + "returncode": clone_result.returncode, + "status": "passed" if clone_result.returncode == 0 else "failed", + }) + if clone_result.returncode != 0: + self.failures.append(f"LFS skip-smudge clone failed for {repo_case.repo}") + return + + # Find pointer files + pointers = self._find_lfs_pointer_files(clone_dir) + self.record({ + "case": "lfs_pointer_discovery", + "repo": repo_case.repo, + "pointer_count": len(pointers), + "status": "passed" if pointers else "failed", + }) + if not pointers: + self.failures.append(f"no LFS pointer files found in {repo_case.repo}") + return + + # Pick the largest pointer for the single-object test to exercise streaming. + best_ptr, oid, size = None, None, 0 + for p in pointers: + text = p.read_text(errors="replace") + o = self._extract_oid(text) + sm = re.search(r"size (\d+)", text) + s = int(sm.group(1)) if sm else 0 + if o and s > size: + best_ptr, oid, size = p, o, s + if not oid: + self.failures.append(f"failed to extract OID from pointer in {repo_case.repo}") + return + + # 2. LFS batch API — cold (first request, cache miss) + cold_batch = self._lfs_batch_timed(repo_case, [{"oid": oid, "size": size}], "lfs_batch_cold") + if not cold_batch: + return + + cold_href = cold_batch.get("href") + + # 3. LFS object download — cold + if cold_href: + self._lfs_download_timed(repo_case, cold_href, oid, size, "lfs_download_cold") + + # 4. LFS batch API — warm (second request, cache hit) + warm_batch = self._lfs_batch_timed(repo_case, [{"oid": oid, "size": size}], "lfs_batch_warm") + + # 5. LFS object download — warm + warm_href = warm_batch.get("href") if warm_batch else cold_href + if warm_href: + self._lfs_download_timed(repo_case, warm_href, oid, size, "lfs_download_warm") + + # 6. Multi-object batch + multi_objects = [] + for p in pointers[:3]: + text = p.read_text(errors="replace") + o = self._extract_oid(text) + sm = re.search(r"size (\d+)", text) + if o and sm: + multi_objects.append({"oid": o, "size": int(sm.group(1))}) + if len(multi_objects) > 1: + self._lfs_batch_timed(repo_case, multi_objects, "lfs_batch_multi_cold", expect_count=len(multi_objects)) + self._lfs_batch_timed(repo_case, multi_objects, "lfs_batch_multi_warm", expect_count=len(multi_objects)) + + # 7. Upload rejected + self._lfs_upload_rejected(repo_case) + + # 8. Invalid OID rejected + self._lfs_invalid_oid(repo_case) + + def _lfs_batch_url(self, repo: str) -> str: + return f"{self.git_url(repo)}/info/lfs/objects/batch" + + def _lfs_batch_timed( + self, + repo_case: RepoCase, + objects: list[dict[str, Any]], + label: str, + *, + expect_count: int = 1, + ) -> dict[str, Any] | None: + body = json.dumps({ + "operation": "download", + "transfers": ["basic"], + "objects": objects, + }).encode() + req = urllib.request.Request( + self._lfs_batch_url(repo_case.repo), + data=body, method="POST", + headers={"Content-Type": LFS_CONTENT_TYPE}, + ) + t0 = time.monotonic() + try: + with urllib.request.urlopen(req, timeout=60) as resp: + data = json.loads(resp.read().decode()) + except Exception as exc: + elapsed = time.monotonic() - t0 + self.record({ + "case": label, "repo": repo_case.repo, + "duration_s": round(elapsed, 3), + "status": "failed", "error": str(exc), + }) + self.failures.append(f"{label} failed for {repo_case.repo}: {exc}") + return None + elapsed = time.monotonic() - t0 + + objs = data.get("objects", []) + all_have_actions = all("actions" in o for o in objs) + passed = len(objs) == expect_count and all_have_actions + self.record({ + "case": label, + "repo": repo_case.repo, + "branch": repo_case.branch, + "duration_s": round(elapsed, 3), + "object_count": len(objs), + "all_have_actions": all_have_actions, + "status": "passed" if passed else "failed", + }) + if not passed: + self.failures.append(f"{label} failed for {repo_case.repo}") + return None + + first = objs[0] + href = first.get("actions", {}).get("download", {}).get("href") + return {"href": href, "oid": first.get("oid"), "size": first.get("size")} + + def _lfs_download_timed( + self, + repo_case: RepoCase, + href: str, + oid: str, + expected_size: int, + label: str, + ) -> None: + t0 = time.monotonic() + try: + with urllib.request.urlopen(href, timeout=60) as resp: + data = resp.read() + except Exception as exc: + elapsed = time.monotonic() - t0 + self.record({ + "case": label, "repo": repo_case.repo, + "oid": oid, "duration_s": round(elapsed, 3), + "status": "failed", "error": str(exc), + }) + self.failures.append(f"{label} failed for {repo_case.repo}: {exc}") + return + elapsed = time.monotonic() - t0 + passed = len(data) == expected_size + self.record({ + "case": label, + "repo": repo_case.repo, + "oid": oid, + "expected_size": expected_size, + "actual_size": len(data), + "duration_s": round(elapsed, 3), + "throughput_mbps": round(len(data) / elapsed / 1_000_000, 2) if elapsed > 0 else None, + "status": "passed" if passed else "failed", + }) + if not passed: + self.failures.append(f"{label} size mismatch for {repo_case.repo}: {len(data)} != {expected_size}") + + def _lfs_upload_rejected(self, repo_case: RepoCase) -> None: + body = json.dumps({ + "operation": "upload", + "transfers": ["basic"], + "objects": [{"oid": "a" * 64, "size": 100}], + }).encode() + req = urllib.request.Request( + self._lfs_batch_url(repo_case.repo), + data=body, method="POST", + headers={"Content-Type": LFS_CONTENT_TYPE}, + ) + try: + with urllib.request.urlopen(req, timeout=30) as resp: + status = resp.status + except urllib.error.HTTPError as exc: + status = exc.code + passed = status == 405 + self.record({ + "case": "lfs_upload_rejected", + "repo": repo_case.repo, + "http_status": status, + "status": "passed" if passed else "failed", + }) + if not passed: + self.failures.append(f"LFS upload not rejected for {repo_case.repo}: HTTP {status}") + + def _lfs_invalid_oid(self, repo_case: RepoCase) -> None: + body = json.dumps({ + "operation": "download", + "transfers": ["basic"], + "objects": [{"oid": "not-a-valid-oid", "size": 100}], + }).encode() + req = urllib.request.Request( + self._lfs_batch_url(repo_case.repo), + data=body, method="POST", + headers={"Content-Type": LFS_CONTENT_TYPE}, + ) + try: + with urllib.request.urlopen(req, timeout=30) as resp: + data = json.loads(resp.read().decode()) + except urllib.error.HTTPError: + self.record({"case": "lfs_invalid_oid", "repo": repo_case.repo, "status": "passed"}) + return + obj = data.get("objects", [{}])[0] + has_error = "error" in obj + self.record({ + "case": "lfs_invalid_oid", + "repo": repo_case.repo, + "has_error": has_error, + "error": obj.get("error"), + "status": "passed" if has_error else "failed", + }) + if not has_error: + self.failures.append(f"LFS invalid OID not rejected for {repo_case.repo}") + + @staticmethod + def _find_lfs_pointer_files(tree: Path) -> list[Path]: + pointers: list[Path] = [] + for path in tree.rglob("*"): + if not path.is_file() or ".git" in path.parts: + continue + try: + text = path.read_text(errors="replace") + except OSError: + continue + if LFS_POINTER_RE.match(text): + pointers.append(path) + return pointers + + @staticmethod + def _extract_oid(content: str) -> str | None: + m = re.search(r"oid sha256:([0-9a-f]{64})", content) + return m.group(1) if m else None + + # ── infrastructure helpers ───────────────────────────────────────── + def assert_receive_pack_rejected(self) -> None: url = ( f"{self.base_url}/git/github.com/astral-sh/uv.git/info/refs" diff --git a/integration_tests/test_lfs_preview_matrix.py b/integration_tests/test_lfs_preview_matrix.py new file mode 100644 index 0000000..cbefffb --- /dev/null +++ b/integration_tests/test_lfs_preview_matrix.py @@ -0,0 +1,372 @@ +#!/usr/bin/env python3 +"""LFS cache preview test matrix. + +Runs against a deployed preview instance to verify: +- LFS batch API proxies upstream and returns download URLs +- LFS object download via cache URL +- Cache hits on repeated requests +- Upload operation is rejected +- Clone with LFS resolves pointers through cache + +Usage: + + RUN_LFS_PREVIEW_MATRIX=1 \ + GIT_CACHE_PREVIEW_BASE_URL=http:// \ + python3 -m unittest -v integration_tests.test_lfs_preview_matrix +""" + +from __future__ import annotations + +import json +import os +import re +import shutil +import subprocess +import tempfile +import time +import unittest +import urllib.error +import urllib.request +from pathlib import Path +from typing import Any + +REPO_ROOT = Path(__file__).resolve().parents[1] +DEFAULT_LFS_TEST_REPO = "github.com/charmbracelet/vhs" +LFS_CONTENT_TYPE = "application/vnd.git-lfs+json" + +LFS_POINTER_RE = re.compile( + r"^version https://git-lfs\.github\.com/spec/v1\n" + r"oid sha256:[0-9a-f]{64}\n" + r"size \d+\n\Z", +) + + +def _run( + cmd: list[str], + *, + cwd: Path | None = None, + env: dict[str, str] | None = None, + check: bool = True, + timeout: int = 300, +) -> subprocess.CompletedProcess[str]: + print("+", " ".join(cmd)) + completed = subprocess.run( + cmd, cwd=cwd, env=env, text=True, + stdout=subprocess.PIPE, stderr=subprocess.PIPE, timeout=timeout, + ) + combined = (completed.stdout or "") + (completed.stderr or "") + for line in combined.strip().splitlines()[:15]: + print(line) + if check: + completed.check_returncode() + return completed + + +def _is_lfs_pointer(content: str) -> bool: + return bool(LFS_POINTER_RE.match(content)) + + +def _extract_oid(content: str) -> str | None: + m = re.search(r"oid sha256:([0-9a-f]{64})", content) + return m.group(1) if m else None + + +def _find_lfs_pointer_files(tree: Path) -> list[Path]: + pointers: list[Path] = [] + for path in tree.rglob("*"): + if not path.is_file() or ".git" in path.parts: + continue + try: + text = path.read_text(errors="replace") + except OSError: + continue + if _is_lfs_pointer(text): + pointers.append(path) + return pointers + + +@unittest.skipUnless( + os.environ.get("RUN_LFS_PREVIEW_MATRIX") == "1", + "set RUN_LFS_PREVIEW_MATRIX=1 to run", +) +class LfsPreviewMatrix(unittest.TestCase): + """LFS cache test matrix against a deployed preview.""" + + @classmethod + def setUpClass(cls) -> None: + cls.base_url = os.environ.get("GIT_CACHE_PREVIEW_BASE_URL", "").rstrip("/") + if not cls.base_url: + raise unittest.SkipTest("set GIT_CACHE_PREVIEW_BASE_URL") + + cls.repo = os.environ.get("GIT_CACHE_LFS_TEST_REPO", DEFAULT_LFS_TEST_REPO) + cls.owner_repo = cls.repo.removeprefix("github.com/") + cls.command_timeout = int(os.environ.get("GIT_CACHE_LFS_COMMAND_TIMEOUT", "300")) + + tmp_base = Path(os.environ.get("TEST_TMPDIR", tempfile.gettempdir())) + tmp_base.mkdir(parents=True, exist_ok=True) + cls.tmp = Path(tempfile.mkdtemp(prefix="lfs-preview-matrix-", dir=tmp_base)) + + results_path = os.environ.get("GIT_CACHE_LFS_RESULTS") + cls.results_path = Path(results_path) if results_path else cls.tmp / "results.jsonl" + cls.results_path.parent.mkdir(parents=True, exist_ok=True) + cls.results: list[dict[str, Any]] = [] + cls.failures: list[str] = [] + + cls.record({ + "case": "suite_start", + "base_url": cls.base_url, + "repo": cls.repo, + }) + + # Verify server health + health_url = f"{cls.base_url}/healthz" + try: + with urllib.request.urlopen(health_url, timeout=10) as resp: + if resp.status != 200: + raise RuntimeError(f"healthz returned {resp.status}") + except Exception as e: + raise RuntimeError(f"preview not healthy: {e}") from e + + @classmethod + def tearDownClass(cls) -> None: + if hasattr(cls, "results_path"): + cls.record({ + "case": "suite_end", + "failures": cls.failures, + "total_cases": len(cls.results), + }) + print(f"LFS preview matrix results: {cls.results_path}") + tmp = getattr(cls, "tmp", None) + if tmp is not None: + shutil.rmtree(tmp, ignore_errors=True) + + @classmethod + def record(cls, event: dict[str, Any]) -> None: + event.setdefault("timestamp", time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime())) + cls.results.append(event) + with cls.results_path.open("a") as handle: + handle.write(json.dumps(event, sort_keys=True) + "\n") + + def _git_url(self) -> str: + return f"{self.base_url}/git/{self.repo}.git" + + def _lfs_batch(self, objects: list[dict[str, Any]]) -> dict[str, Any]: + url = f"{self._git_url()}/info/lfs/objects/batch" + body = json.dumps({ + "operation": "download", + "transfers": ["basic"], + "objects": objects, + }).encode() + req = urllib.request.Request( + url, data=body, method="POST", + headers={"Content-Type": LFS_CONTENT_TYPE}, + ) + with urllib.request.urlopen(req, timeout=60) as resp: + return json.loads(resp.read().decode()) + + # ── matrix cases ────────────────────────────────────────────────── + + def test_01_clone_skip_smudge(self) -> None: + """Clone with GIT_LFS_SKIP_SMUDGE=1 and verify pointer files.""" + t0 = time.monotonic() + clone_dir = self.tmp / "clone-skip-smudge" + env = os.environ.copy() + env["GIT_LFS_SKIP_SMUDGE"] = "1" + result = _run( + ["git", "clone", "--depth", "1", self._git_url(), str(clone_dir)], + env=env, check=False, timeout=self.command_timeout, + ) + elapsed = time.monotonic() - t0 + passed = result.returncode == 0 + pointers = _find_lfs_pointer_files(clone_dir) if passed else [] + self.record({ + "case": "clone_skip_smudge", + "status": "passed" if passed and pointers else "failed", + "returncode": result.returncode, + "pointer_count": len(pointers), + "elapsed_s": round(elapsed, 2), + }) + self.assertEqual(result.returncode, 0) + self.assertGreater(len(pointers), 0, "expected LFS pointer files") + + def test_02_batch_download_cold(self) -> None: + """First batch request: cold cache, proxies from upstream.""" + clone_dir = self.tmp / "clone-for-oid" + env = os.environ.copy() + env["GIT_LFS_SKIP_SMUDGE"] = "1" + _run( + ["git", "clone", "--depth", "1", self._git_url(), str(clone_dir)], + env=env, check=False, + ) + pointers = _find_lfs_pointer_files(clone_dir) + self.assertGreater(len(pointers), 0) + + ptr_text = pointers[0].read_text(errors="replace") + oid = _extract_oid(ptr_text) + self.assertIsNotNone(oid) + size_m = re.search(r"size (\d+)", ptr_text) + size = int(size_m.group(1)) if size_m else 0 + + t0 = time.monotonic() + resp = self._lfs_batch([{"oid": oid, "size": size}]) + elapsed = time.monotonic() - t0 + obj = resp["objects"][0] + has_actions = "actions" in obj + + self.record({ + "case": "batch_download_cold", + "status": "passed" if has_actions else "failed", + "oid": oid, + "size": size, + "elapsed_s": round(elapsed, 2), + "has_download_url": has_actions, + }) + self.assertTrue(has_actions, f"expected actions, got: {obj}") + + # Save OID for later tests + self.__class__._cached_oid = oid + self.__class__._cached_size = size + self.__class__._cached_href = obj["actions"]["download"]["href"] + + def test_03_object_download(self) -> None: + """Download the LFS object via the cache URL.""" + oid = getattr(self.__class__, "_cached_oid", None) + size = getattr(self.__class__, "_cached_size", 0) + href = getattr(self.__class__, "_cached_href", None) + if not href: + self.skipTest("no cached href from test_02") + + t0 = time.monotonic() + with urllib.request.urlopen(href, timeout=60) as resp: + data = resp.read() + elapsed = time.monotonic() - t0 + passed = len(data) == size + + self.record({ + "case": "object_download", + "status": "passed" if passed else "failed", + "oid": oid, + "expected_size": size, + "actual_size": len(data), + "elapsed_s": round(elapsed, 2), + }) + self.assertEqual(len(data), size) + + def test_04_batch_download_warm(self) -> None: + """Second batch request for same OID: should be a cache hit.""" + oid = getattr(self.__class__, "_cached_oid", None) + size = getattr(self.__class__, "_cached_size", 0) + if not oid: + self.skipTest("no cached OID from test_02") + + t0 = time.monotonic() + resp = self._lfs_batch([{"oid": oid, "size": size}]) + elapsed = time.monotonic() - t0 + obj = resp["objects"][0] + has_actions = "actions" in obj + + self.record({ + "case": "batch_download_warm", + "status": "passed" if has_actions else "failed", + "oid": oid, + "elapsed_s": round(elapsed, 2), + }) + self.assertTrue(has_actions) + + def test_05_object_download_warm(self) -> None: + """Download cached object again — should be fast.""" + href = getattr(self.__class__, "_cached_href", None) + size = getattr(self.__class__, "_cached_size", 0) + if not href: + self.skipTest("no cached href from test_02") + + t0 = time.monotonic() + with urllib.request.urlopen(href, timeout=60) as resp: + data = resp.read() + elapsed = time.monotonic() - t0 + + self.record({ + "case": "object_download_warm", + "status": "passed" if len(data) == size else "failed", + "expected_size": size, + "actual_size": len(data), + "elapsed_s": round(elapsed, 2), + }) + self.assertEqual(len(data), size) + + def test_06_upload_rejected(self) -> None: + """LFS upload operation should be rejected.""" + url = f"{self._git_url()}/info/lfs/objects/batch" + body = json.dumps({ + "operation": "upload", + "transfers": ["basic"], + "objects": [{"oid": "a" * 64, "size": 100}], + }).encode() + req = urllib.request.Request( + url, data=body, method="POST", + headers={"Content-Type": LFS_CONTENT_TYPE}, + ) + try: + with urllib.request.urlopen(req, timeout=30) as resp: + status = resp.status + except urllib.error.HTTPError as exc: + status = exc.code + + passed = status == 405 + self.record({ + "case": "upload_rejected", + "status": "passed" if passed else "failed", + "http_status": status, + }) + self.assertEqual(status, 405) + + def test_07_invalid_oid_rejected(self) -> None: + """Batch with invalid OID returns validation error.""" + resp = self._lfs_batch([{"oid": "not-a-valid-oid", "size": 100}]) + obj = resp["objects"][0] + has_error = "error" in obj + + self.record({ + "case": "invalid_oid_rejected", + "status": "passed" if has_error else "failed", + "error": obj.get("error"), + }) + self.assertTrue(has_error) + + def test_08_multi_object_batch(self) -> None: + """Batch with multiple objects returns results for all.""" + clone_dir = self.tmp / "clone-multi-oid" + env = os.environ.copy() + env["GIT_LFS_SKIP_SMUDGE"] = "1" + _run( + ["git", "clone", "--depth", "1", self._git_url(), str(clone_dir)], + env=env, check=False, + ) + pointers = _find_lfs_pointer_files(clone_dir) + objects = [] + for p in pointers[:3]: + text = p.read_text(errors="replace") + oid = _extract_oid(text) + size_m = re.search(r"size (\d+)", text) + if oid and size_m: + objects.append({"oid": oid, "size": int(size_m.group(1))}) + if not objects: + self.skipTest("no LFS pointer files found") + + resp = self._lfs_batch(objects) + result_count = len(resp.get("objects", [])) + all_have_actions = all("actions" in o for o in resp["objects"]) + + self.record({ + "case": "multi_object_batch", + "status": "passed" if result_count == len(objects) and all_have_actions else "failed", + "requested": len(objects), + "returned": result_count, + "all_have_actions": all_have_actions, + }) + self.assertEqual(result_count, len(objects)) + self.assertTrue(all_have_actions) + + +if __name__ == "__main__": + unittest.main(verbosity=2) diff --git a/integration_tests/test_lfs_smoke.py b/integration_tests/test_lfs_smoke.py new file mode 100644 index 0000000..092552d --- /dev/null +++ b/integration_tests/test_lfs_smoke.py @@ -0,0 +1,561 @@ +#!/usr/bin/env python3 +"""Opt-in integration tests for Git LFS caching through the cache. + +These tests use only Python's standard library and shell out to ``cargo`` +and ``git``. They are skipped unless ``RUN_GITHUB_INTEGRATION=1`` is set. + +The suite verifies: +- LFS repos clone correctly through the cache (pointer files present) +- The LFS batch API proxies upstream and returns download URLs +- LFS objects are cached in the object store after first access +- The upstream-URL workaround lets ``git lfs pull`` succeed +""" + +from __future__ import annotations + +import json +import os +import re +import shutil +import socket +import subprocess +import tempfile +import time +import unittest +import urllib.error +import urllib.request +from pathlib import Path + + +REPO_ROOT = Path(__file__).resolve().parents[1] + +DEFAULT_LFS_TEST_REPO = "github.com/charmbracelet/vhs" +LFS_TEST_REPO = os.environ.get("GIT_CACHE_LFS_TEST_REPO", DEFAULT_LFS_TEST_REPO) + +LFS_POINTER_RE = re.compile( + r"^version https://git-lfs\.github\.com/spec/v1\n" + r"oid sha256:[0-9a-f]{64}\n" + r"size \d+\n\Z", +) + + +def _run( + cmd: list[str], + *, + cwd: Path = REPO_ROOT, + env: dict[str, str] | None = None, + check: bool = True, +) -> subprocess.CompletedProcess[str]: + print("+", " ".join(cmd)) + completed = subprocess.run( + cmd, + cwd=cwd, + env=env, + text=True, + stdout=subprocess.PIPE, + stderr=subprocess.PIPE, + timeout=300, + ) + combined = (completed.stdout or "") + (completed.stderr or "") + for line in combined.strip().splitlines()[:20]: + print(line) + if combined.strip().count("\n") > 20: + print(f" ... ({combined.strip().count(chr(10)) - 20} more lines)") + if check: + completed.check_returncode() + return completed + + +def _free_port() -> int: + with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as sock: + sock.bind(("127.0.0.1", 0)) + return int(sock.getsockname()[1]) + + +def _is_lfs_pointer(content: str) -> bool: + return bool(LFS_POINTER_RE.match(content)) + + +def _find_lfs_pointer_files(tree: Path) -> list[Path]: + """Return paths of files that contain LFS pointer text.""" + pointers: list[Path] = [] + for path in tree.rglob("*"): + if not path.is_file(): + continue + if path.parts and ".git" in path.parts: + continue + try: + text = path.read_text(errors="replace") + except OSError: + continue + if _is_lfs_pointer(text): + pointers.append(path) + return pointers + + +def _extract_oid_from_pointer(content: str) -> str | None: + """Extract the sha256 OID from an LFS pointer file's text.""" + m = re.search(r"oid sha256:([0-9a-f]{64})", content) + return m.group(1) if m else None + + +@unittest.skipUnless( + os.environ.get("RUN_GITHUB_INTEGRATION") == "1", + "set RUN_GITHUB_INTEGRATION=1 to test LFS behavior through the cache", +) +class LfsSmokeTest(unittest.TestCase): + """Test Git LFS caching through the read-through /git/ endpoint.""" + + @classmethod + def setUpClass(cls) -> None: + # Gate on git-lfs being installed + try: + subprocess.run( + ["git", "lfs", "version"], + stdout=subprocess.PIPE, + stderr=subprocess.PIPE, + timeout=10, + ) + except (FileNotFoundError, subprocess.TimeoutExpired): + raise unittest.SkipTest("git-lfs is not installed") + + base = Path(os.environ.get("TEST_TMPDIR", tempfile.gettempdir())) + base.mkdir(parents=True, exist_ok=True) + cls.tmp = Path(tempfile.mkdtemp(prefix="git-cache-lfs-smoke-", dir=base)) + cls.port = _free_port() + cls.base_url = f"http://127.0.0.1:{cls.port}" + cls.cache_root = cls.tmp / "cache" + cls.object_root = cls.tmp / "object-store" + cls.repo = LFS_TEST_REPO + cls.owner_repo = cls.repo.removeprefix("github.com/") + + config_path = cls.tmp / "config.toml" + config_path.write_text( + f"""\ +bind_addr = "127.0.0.1:{cls.port}" +cache_root = "{cls.cache_root}" +git_timeout_seconds = 300 +max_git_output_bytes = 1073741824 +rate_limit_per_minute = 0 +allowed_upstream_hosts = ["github.com"] + +[object_store] +kind = "local" +root = "{cls.object_root}" + +[disk] +quota_bytes = 10737418240 +min_free_bytes = 0 + +[git_remote] +commit_read_through = true + +[lfs] +enabled = true +""" + ) + + _run(["cargo", "build", "-p", "git-cache-api"]) + + git_tmp = cls.tmp / "git-tmp" + git_tmp.mkdir(parents=True, exist_ok=True) + + env = os.environ.copy() + env["RUST_LOG"] = "info" + env["GIT_CACHE_CONFIG"] = str(config_path) + env["TMPDIR"] = str(git_tmp) + cls._server_log = open(cls.tmp / "server.log", "w") + cls.server = subprocess.Popen( + [str(REPO_ROOT / "target/debug/git-cache-api")], + cwd=REPO_ROOT, + env=env, + stdout=cls._server_log, + stderr=subprocess.STDOUT, + ) + cls._wait_for_healthz() + + @classmethod + def tearDownClass(cls) -> None: + server = getattr(cls, "server", None) + if server is not None: + server.terminate() + try: + server.wait(timeout=10) + except subprocess.TimeoutExpired: + server.kill() + server.wait(timeout=10) + log_fh = getattr(cls, "_server_log", None) + if log_fh is not None: + log_fh.close() + log_path = cls.tmp / "server.log" + if log_path.exists(): + tail = log_path.read_text().strip().splitlines()[-30:] + for line in tail: + print(line) + + tmp = getattr(cls, "tmp", None) + if tmp is not None: + shutil.rmtree(tmp, ignore_errors=True) + + @classmethod + def _wait_for_healthz(cls) -> None: + deadline = time.time() + 30 + url = f"{cls.base_url}/healthz" + last_error: Exception | None = None + + while time.time() < deadline: + if cls.server.poll() is not None: + output = cls.server.stdout.read() if cls.server.stdout is not None else "" + raise RuntimeError(f"git-cache-api exited early:\n{output}") + try: + with urllib.request.urlopen(url, timeout=1) as response: + if response.status == 200: + return + except (OSError, urllib.error.URLError) as error: + last_error = error + time.sleep(0.25) + + raise TimeoutError(f"timed out waiting for {url}: {last_error}") + + def _git_url(self, repo: str | None = None) -> str: + r = repo or self.repo + return f"{self.base_url}/git/{r}.git" + + def _clone_env(self, *, skip_smudge: bool = False) -> dict[str, str]: + env = os.environ.copy() + if skip_smudge: + env["GIT_LFS_SKIP_SMUDGE"] = "1" + else: + env.pop("GIT_LFS_SKIP_SMUDGE", None) + return env + + # ── Test cases ──────────────────────────────────────────────────── + + def test_clone_with_lfs_skip_smudge(self) -> None: + """Clone with GIT_LFS_SKIP_SMUDGE=1: pointer files should remain.""" + clone_dir = self.tmp / "clone-skip-smudge" + result = _run( + [ + "git", "clone", "--depth", "1", + self._git_url(), str(clone_dir), + ], + env=self._clone_env(skip_smudge=True), + check=False, + ) + self.assertEqual(result.returncode, 0, f"clone failed:\n{result.stderr}") + + entries = list(clone_dir.iterdir()) + self.assertTrue( + len(entries) > 1, # at least more than just .git + f"working tree is empty: {entries}", + ) + + pointers = _find_lfs_pointer_files(clone_dir) + self.assertTrue( + len(pointers) > 0, + "expected at least one LFS pointer file in the working tree", + ) + + def test_lfs_batch_api_returns_download_urls(self) -> None: + """The LFS batch endpoint should proxy upstream and return download URLs.""" + # First clone to get a real OID from the repo + clone_dir = self.tmp / "clone-for-batch-oid" + _run( + [ + "git", "clone", "--depth", "1", + self._git_url(), str(clone_dir), + ], + env=self._clone_env(skip_smudge=True), + check=False, + ) + pointers = _find_lfs_pointer_files(clone_dir) + self.assertTrue(len(pointers) > 0, "need at least one LFS pointer for batch test") + + ptr_text = pointers[0].read_text(errors="replace") + oid = _extract_oid_from_pointer(ptr_text) + self.assertIsNotNone(oid, f"could not extract OID from pointer:\n{ptr_text}") + + size_match = re.search(r"size (\d+)", ptr_text) + size = int(size_match.group(1)) if size_match else 0 + + url = f"{self._git_url()}/info/lfs/objects/batch" + body = json.dumps({ + "operation": "download", + "transfers": ["basic"], + "objects": [{"oid": oid, "size": size}], + }).encode() + req = urllib.request.Request( + url, + data=body, + method="POST", + headers={"Content-Type": "application/vnd.git-lfs+json"}, + ) + with urllib.request.urlopen(req, timeout=60) as resp: + self.assertEqual(resp.status, 200) + resp_body = json.loads(resp.read().decode()) + + self.assertIn("objects", resp_body) + self.assertEqual(len(resp_body["objects"]), 1) + obj = resp_body["objects"][0] + self.assertEqual(obj["oid"], oid) + self.assertIn("actions", obj, f"expected download actions, got error: {obj.get('error')}") + self.assertIn("download", obj["actions"]) + self.assertIn("href", obj["actions"]["download"]) + download_href = obj["actions"]["download"]["href"] + self.assertIn(f"/info/lfs/objects/{oid}", download_href) + + def test_lfs_batch_upload_rejected(self) -> None: + """LFS upload operation should be rejected (cache is read-only).""" + url = f"{self._git_url()}/info/lfs/objects/batch" + body = json.dumps({ + "operation": "upload", + "transfers": ["basic"], + "objects": [{"oid": "a" * 64, "size": 100}], + }).encode() + req = urllib.request.Request( + url, + data=body, + method="POST", + headers={"Content-Type": "application/vnd.git-lfs+json"}, + ) + try: + with urllib.request.urlopen(req, timeout=30) as resp: + self.fail(f"expected 405 but got {resp.status}") + except urllib.error.HTTPError as exc: + self.assertEqual( + exc.code, 405, + f"expected 405 for upload, got {exc.code}", + ) + + def test_lfs_object_download_from_cache(self) -> None: + """After a batch request caches an object, the download URL serves it.""" + clone_dir = self.tmp / "clone-for-download" + _run( + [ + "git", "clone", "--depth", "1", + self._git_url(), str(clone_dir), + ], + env=self._clone_env(skip_smudge=True), + check=False, + ) + pointers = _find_lfs_pointer_files(clone_dir) + self.assertTrue(len(pointers) > 0, "need at least one LFS pointer") + + ptr_text = pointers[0].read_text(errors="replace") + oid = _extract_oid_from_pointer(ptr_text) + self.assertIsNotNone(oid) + + size_match = re.search(r"size (\d+)", ptr_text) + size = int(size_match.group(1)) if size_match else 0 + + # Trigger batch to cache the object + batch_url = f"{self._git_url()}/info/lfs/objects/batch" + batch_body = json.dumps({ + "operation": "download", + "transfers": ["basic"], + "objects": [{"oid": oid, "size": size}], + }).encode() + batch_req = urllib.request.Request( + batch_url, + data=batch_body, + method="POST", + headers={"Content-Type": "application/vnd.git-lfs+json"}, + ) + with urllib.request.urlopen(batch_req, timeout=60) as resp: + batch_resp = json.loads(resp.read().decode()) + + download_href = batch_resp["objects"][0]["actions"]["download"]["href"] + + # Download the object via the cache URL + with urllib.request.urlopen(download_href, timeout=60) as resp: + self.assertEqual(resp.status, 200) + content_type = resp.headers.get("Content-Type", "") + self.assertIn("octet-stream", content_type) + data = resp.read() + + self.assertGreater(len(data), 0, "downloaded LFS object should not be empty") + self.assertEqual(len(data), size, f"expected {size} bytes, got {len(data)}") + + # Verify it's NOT a pointer file (it's real binary content) + try: + text = data.decode("utf-8") + self.assertFalse( + _is_lfs_pointer(text), + "downloaded content should be real data, not a pointer", + ) + except UnicodeDecodeError: + pass # binary content, definitely not a pointer + + def test_lfs_cache_hit_on_second_request(self) -> None: + """Second batch + download for the same OID should be served from cache.""" + clone_dir = self.tmp / "clone-for-cache-hit" + _run( + [ + "git", "clone", "--depth", "1", + self._git_url(), str(clone_dir), + ], + env=self._clone_env(skip_smudge=True), + check=False, + ) + pointers = _find_lfs_pointer_files(clone_dir) + self.assertTrue(len(pointers) > 0) + + ptr_text = pointers[0].read_text(errors="replace") + oid = _extract_oid_from_pointer(ptr_text) + size_match = re.search(r"size (\d+)", ptr_text) + size = int(size_match.group(1)) if size_match else 0 + + batch_url = f"{self._git_url()}/info/lfs/objects/batch" + batch_body = json.dumps({ + "operation": "download", + "transfers": ["basic"], + "objects": [{"oid": oid, "size": size}], + }).encode() + + # First request (may miss cache) + req1 = urllib.request.Request( + batch_url, data=batch_body, method="POST", + headers={"Content-Type": "application/vnd.git-lfs+json"}, + ) + with urllib.request.urlopen(req1, timeout=60) as resp: + resp1 = json.loads(resp.read().decode()) + href1 = resp1["objects"][0]["actions"]["download"]["href"] + with urllib.request.urlopen(href1, timeout=60) as resp: + data1 = resp.read() + + # Second request (should hit cache) + req2 = urllib.request.Request( + batch_url, data=batch_body, method="POST", + headers={"Content-Type": "application/vnd.git-lfs+json"}, + ) + with urllib.request.urlopen(req2, timeout=60) as resp: + resp2 = json.loads(resp.read().decode()) + href2 = resp2["objects"][0]["actions"]["download"]["href"] + with urllib.request.urlopen(href2, timeout=60) as resp: + data2 = resp.read() + + self.assertEqual(data1, data2, "cache hit should return identical content") + self.assertEqual(len(data1), size) + + # Verify the object exists in the local object store. + # The domain layer appends a schema version suffix (e.g. "-v3") to the + # configured object store root, so search the actual versioned directory. + versioned_root = self.object_root.parent / (self.object_root.name + "-v3") + search_root = versioned_root if versioned_root.is_dir() else self.object_root + found = False + for path in search_root.rglob("*"): + if path.is_file() and oid in path.name: + found = True + break + self.assertTrue(found, f"LFS object {oid} should exist in local object store") + + def test_lfs_pull_with_upstream_url_workaround(self) -> None: + """Clone with skip-smudge, set lfs.url to upstream, then git lfs pull.""" + clone_dir = self.tmp / "clone-lfs-pull-workaround" + result = _run( + [ + "git", "clone", "--depth", "1", + self._git_url(), str(clone_dir), + ], + env=self._clone_env(skip_smudge=True), + check=False, + ) + self.assertEqual(result.returncode, 0, f"clone failed:\n{result.stderr}") + + pointers_before = _find_lfs_pointer_files(clone_dir) + self.assertTrue( + len(pointers_before) > 0, + "expected at least one LFS pointer file before lfs pull", + ) + + # Point lfs.url at the real upstream LFS server + _run( + [ + "git", "config", "lfs.url", + f"https://github.com/{self.owner_repo}.git/info/lfs", + ], + cwd=clone_dir, + ) + + pull_result = _run( + ["git", "lfs", "pull"], + cwd=clone_dir, + check=False, + ) + self.assertEqual( + pull_result.returncode, 0, + f"git lfs pull failed:\n{pull_result.stderr}", + ) + + # Verify at least one former pointer is now real content + resolved = 0 + for ptr_path in pointers_before: + if not ptr_path.exists(): + continue + try: + text = ptr_path.read_text(errors="replace") + except OSError: + continue + if not _is_lfs_pointer(text): + resolved += 1 + + self.assertGreater( + resolved, 0, + "expected at least one LFS pointer to be resolved to real content after lfs pull", + ) + + def test_lfs_clone_through_cache_resolves_pointers(self) -> None: + """Batch+download a single LFS object through the cache and verify content.""" + clone_dir = self.tmp / "clone-lfs-through-cache" + _run( + [ + "git", "clone", "--depth", "1", + self._git_url(), str(clone_dir), + ], + env=self._clone_env(skip_smudge=True), + check=False, + ) + + pointers = _find_lfs_pointer_files(clone_dir) + self.assertTrue(len(pointers) > 0, "need at least one LFS pointer") + + # Pick a pointer, send a batch request through the cache, then download. + target = pointers[0] + ptr_text = target.read_text(errors="replace") + oid = _extract_oid_from_pointer(ptr_text) + self.assertIsNotNone(oid) + size_match = re.search(r"size (\d+)", ptr_text) + size = int(size_match.group(1)) if size_match else 0 + + batch_url = f"{self._git_url()}/info/lfs/objects/batch" + batch_body = json.dumps({ + "operation": "download", + "transfers": ["basic"], + "objects": [{"oid": oid, "size": size}], + }).encode() + + req = urllib.request.Request( + batch_url, data=batch_body, method="POST", + headers={"Content-Type": "application/vnd.git-lfs+json"}, + ) + with urllib.request.urlopen(req, timeout=60) as resp: + batch_resp = json.loads(resp.read().decode()) + + href = batch_resp["objects"][0]["actions"]["download"]["href"] + with urllib.request.urlopen(href, timeout=60) as dl_resp: + data = dl_resp.read() + + self.assertEqual(len(data), size, f"expected {size} bytes, got {len(data)}") + # Verify it's real content, not a pointer + try: + text = data.decode("utf-8") + self.assertFalse( + _is_lfs_pointer(text), + "downloaded content should be real data, not a pointer", + ) + except UnicodeDecodeError: + pass # binary content is definitely not a pointer + + +if __name__ == "__main__": + unittest.main(verbosity=2) diff --git a/python/aws/api_task_definition.py b/python/aws/api_task_definition.py index 2cb073c..9d93814 100644 --- a/python/aws/api_task_definition.py +++ b/python/aws/api_task_definition.py @@ -11,6 +11,7 @@ {"name": "GIT_CACHE_DISK_QUOTA_BYTES", "value": os.environ["GIT_CACHE_DISK_QUOTA_BYTES"]}, {"name": "GIT_CACHE_COMPACTION_CHAIN_DEPTH_THRESHOLD", "value": os.environ.get("GIT_CACHE_COMPACTION_CHAIN_DEPTH_THRESHOLD", "10")}, {"name": "GIT_CACHE_COMPACTION_INLINE", "value": os.environ.get("GIT_CACHE_COMPACTION_INLINE", "false")}, + {"name": "GIT_CACHE_PUBLIC_PATH_PREFIX", "value": os.environ.get("ECS_PUBLIC_PATH_PREFIX", os.environ.get("GIT_CACHE_PUBLIC_PATH_PREFIX", ""))}, {"name": "GIT_CACHE_GIT_TIMEOUT_SECONDS", "value": os.environ.get("GIT_CACHE_GIT_TIMEOUT_SECONDS", "3600")}, {"name": "GIT_CACHE_MAX_CONCURRENT_GIT_PROCESSES", "value": os.environ.get("GIT_CACHE_MAX_CONCURRENT_GIT_PROCESSES", "8")}, {"name": "GIT_CACHE_MAX_GIT_OUTPUT_BYTES", "value": os.environ.get("GIT_CACHE_MAX_GIT_OUTPUT_BYTES", "8589934592")},