Implement multi-worker conflict avoidance #47
There are no files selected for viewing
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
|
|
@@ -14,7 +14,10 @@ use git_cache_domain::{ | |
| MaterializerExecutor, | ||
| }; | ||
| use git_cache_git::UploadPackProcess; | ||
| use git_cache_worker::{ | ||
| LeaseAcquire, ObjectStoreRepoLeaseManager, RepoLeaseManager, UpdateCoordinator, | ||
| UpdateDisposition, | ||
| }; | ||
| use http::{header, Method, StatusCode, Uri}; | ||
| use serde::Serialize; | ||
| use std::collections::HashMap; | ||
|
|
@@ -28,8 +31,10 @@ use tokio::io::AsyncRead; | |
| use tokio::sync::OwnedSemaphorePermit; | ||
| use tokio::time::Sleep; | ||
| use tokio_util::io::ReaderStream; | ||
| use tracing::warn; | ||
|
|
||
| const GIT_UPLOAD_PACK_STREAM_BUFFER_BYTES: usize = 64 * 1024; | ||
| const LEASE_BUSY_RETRY_INTERVAL: Duration = Duration::from_millis(100); | ||
|
|
||
| pub fn app(config: AppConfig) -> Router { | ||
| app_result(config).expect("failed to initialize git-cache-api") | ||
|
|
@@ -73,6 +78,7 @@ fn router(git_remote_enabled: bool, state: Arc<ApiState>) -> CoreResult<Router> | |
| struct ApiState { | ||
| domain: Arc<AppState>, | ||
| coordinator: UpdateCoordinator, | ||
| leases: Arc<dyn RepoLeaseManager>, | ||
| metrics: Arc<Metrics>, | ||
| rate_limiter: Arc<RateLimiter>, | ||
| } | ||
|
|
@@ -92,18 +98,64 @@ impl ApiState { | |
|
|
||
| fn with_domain(rate_limiter: RateLimiter, domain: Arc<AppState>) -> CoreResult<Self> { | ||
| let executor = Arc::new(MaterializerExecutor::new(Arc::clone(&domain))); | ||
| let leases: Arc<dyn RepoLeaseManager> = Arc::new(ObjectStoreRepoLeaseManager::new( | ||
| Arc::clone(&domain.store), | ||
| &domain.config.leases, | ||
| )); | ||
| let coordinator = UpdateCoordinator::new(executor, Arc::clone(&leases)); | ||
| Materializer::new(Arc::clone(&domain)).enqueue_pending_generation_scan(); | ||
| Ok(Self { | ||
| domain, | ||
| coordinator, | ||
| leases, | ||
| metrics: Arc::new(Metrics::default()), | ||
| rate_limiter: Arc::new(rate_limiter), | ||
| }) | ||
| } | ||
| } | ||
|
|
||
| async fn read_through_with_busy_wait( | ||
| state: &Arc<ApiState>, | ||
| repo: git_cache_core::RepoKey, | ||
| selector: Selector, | ||
| ) -> CoreResult<git_cache_worker::UpdateOutcome> { | ||
| let max_wait = Duration::from_secs(state.domain.config.leases.busy_retry_after_seconds); | ||
| let started_at = Instant::now(); | ||
|
|
||
| loop { | ||
| let outcome = state | ||
| .coordinator | ||
| .read_through(repo.clone(), selector.clone()) | ||
| .await?; | ||
| if outcome.disposition != UpdateDisposition::LeaseBusy || started_at.elapsed() >= max_wait { | ||
| return Ok(outcome); | ||
| } | ||
| tokio::time::sleep( | ||
| LEASE_BUSY_RETRY_INTERVAL.min(max_wait.saturating_sub(started_at.elapsed())), | ||
| ) | ||
| .await; | ||
| } | ||
| } | ||
|
|
||
| async fn acquire_repo_write_lease_with_busy_wait( | ||
| state: &Arc<ApiState>, | ||
| repo: &git_cache_core::RepoKey, | ||
| ) -> CoreResult<LeaseAcquire> { | ||
| let max_wait = Duration::from_secs(state.domain.config.leases.busy_retry_after_seconds); | ||
| let started_at = Instant::now(); | ||
|
|
||
| loop { | ||
| let lease = state.leases.acquire(repo).await?; | ||
| if !matches!(lease, LeaseAcquire::Busy) || started_at.elapsed() >= max_wait { | ||
| return Ok(lease); | ||
| } | ||
| tokio::time::sleep( | ||
| LEASE_BUSY_RETRY_INTERVAL.min(max_wait.saturating_sub(started_at.elapsed())), | ||
| ) | ||
| .await; | ||
| } | ||
| } | ||
|
|
||
| async fn healthz() -> Json<HealthResponse> { | ||
| Json(HealthResponse { | ||
| ok: true, | ||
|
|
@@ -174,45 +226,37 @@ async fn handle_materialize_request( | |
| .materialize_total | ||
| .fetch_add(1, Ordering::Relaxed); | ||
|
|
||
| let outcome = | ||
| read_through_with_busy_wait(state, request.repo.clone(), request.selector.clone()).await; | ||
| match outcome { | ||
| Ok(o) if o.disposition == UpdateDisposition::LeaseBusy => { | ||
| let retry_after = state.domain.config.leases.busy_retry_after_seconds; | ||
| return Ok(Response::builder() | ||
| .status(StatusCode::SERVICE_UNAVAILABLE) | ||
| .header(header::RETRY_AFTER, retry_after.to_string()) | ||
| .header(header::CONTENT_TYPE, "application/json") | ||
| .body(Body::from( | ||
| serde_json::to_string(&serde_json::json!({ | ||
| "error": "update in progress, retry later" | ||
| })) | ||
| .expect("json serialization"), | ||
| )) | ||
| .expect("lease busy response")); | ||
| } | ||
| Err(error) => { | ||
| state | ||
| .metrics | ||
| .materialize_errors_total | ||
| .fetch_add(1, Ordering::Relaxed); | ||
| return Err(error.into()); | ||
| } | ||
| Ok(_) => {} | ||
| } | ||
|
|
||
| let materializer = Materializer::new(Arc::clone(&state.domain)); | ||
| let result = materializer | ||
| .materialize_after_upstream_validation(request) | ||
|
Owner
Author
There was a problem hiding this comment. [P1] This second materialization re-executes
Contributor
There was a problem hiding this comment. Fixed in |
||
| .await; | ||
|
|
||
| match result { | ||
| Ok(response) => Ok(Json(response).into_response()), | ||
|
|
@@ -372,9 +416,37 @@ async fn git_repo( | |
| .git_remote_upload_pack_total | ||
| .fetch_add(1, Ordering::Relaxed); | ||
|
|
||
| // Wait briefly for repo-write before falling back to 503, then hold the | ||
| // lease while processing wants. This keeps ensure_wants_available() and | ||
| // any publish_generation() it triggers under durable coordination, and | ||
| // passes the fencing token into Materializer so publish writes can verify | ||
| // ownership immediately before mutation. | ||
| let lease = match acquire_repo_write_lease_with_busy_wait(&state, &repo).await { | ||
|
Owner
Author
There was a problem hiding this comment. [P1] The direct upload-pack path can now fail ordinary Git protocol requests with repo-write contention.
Contributor
There was a problem hiding this comment. Fixed in |
||
| Ok(LeaseAcquire::Acquired(lease)) => lease, | ||
| Ok(LeaseAcquire::Busy) => { | ||
| let retry_after = state.domain.config.leases.busy_retry_after_seconds; | ||
|
Owner
Author
There was a problem hiding this comment. Why dont we retry ourselves first by holding the request for X seconds (X perhaps few seconds?)
Contributor
There was a problem hiding this comment. Fixed in 9afbdd4. The API now waits/retries before returning
Verification: |
||
| return Response::builder() | ||
| .status(StatusCode::SERVICE_UNAVAILABLE) | ||
| .header(header::RETRY_AFTER, retry_after.to_string()) | ||
| .header(header::CONTENT_TYPE, "application/json") | ||
| .body(Body::from(r#"{"error":"update in progress, retry later"}"#)) | ||
| .expect("lease busy response"); | ||
| } | ||
| Err(error) => return ApiError::from(error).into_response(), | ||
| }; | ||
| let token = lease.token().to_string(); | ||
| let leased_materializer = Materializer::with_lease_token(Arc::clone(&state.domain), token); | ||
| let process_result = leased_materializer.handle_upload_pack(&repo, &body).await; | ||
| let release_result = lease.release().await; | ||
|
|
||
| match (process_result, release_result) { | ||
| (Ok(process), Ok(())) => stream_upload_pack_response(&state, process), | ||
| (Err(error), Ok(())) => ApiError::from(error).into_response(), | ||
| (Ok(_), Err(error)) => ApiError::from(error).into_response(), | ||
| (Err(error), Err(release_error)) => { | ||
| warn!(%repo, %release_error, "failed to release repo lease after upload-pack error"); | ||
| ApiError::from(error).into_response() | ||
| } | ||
| } | ||
| } else { | ||
| ApiError::from(GitCacheError::Unsupported(format!( | ||
|
|
@@ -492,10 +564,16 @@ impl From<GitCacheError> for ApiError { | |
| GitCacheError::NotImplemented(_) => StatusCode::NOT_IMPLEMENTED, | ||
| GitCacheError::Validation(_) => StatusCode::BAD_REQUEST, | ||
| GitCacheError::Timeout(_) => StatusCode::GATEWAY_TIMEOUT, | ||
| GitCacheError::LeaseBusy(_) => StatusCode::SERVICE_UNAVAILABLE, | ||
| GitCacheError::Conflict(_) | ||
| | GitCacheError::LeaseLost(_) | ||
| | GitCacheError::LeaseStealConflict(_) | ||
| | GitCacheError::CasConflict(_) => StatusCode::CONFLICT, | ||
| GitCacheError::PendingGenerationInvalid(_) | ||
| | GitCacheError::ColdHydrationFailed(_) | ||
| | GitCacheError::Internal(_) | ||
| | GitCacheError::Io(_) | ||
| | GitCacheError::Json(_) => StatusCode::INTERNAL_SERVER_ERROR, | ||
| }; | ||
|
|
||
| Self { | ||
|
|
@@ -663,6 +741,7 @@ mod tests { | |
| max_concurrent_git_processes: git_cache_core::default_max_concurrent_git_processes(), | ||
| session_cleanup_interval_secs: 300, | ||
| max_concurrent_generation_verifications: 1, | ||
| leases: Default::default(), | ||
| }; | ||
| let api_state = ApiState::try_new(config).unwrap(); | ||
| let mut query = HashMap::new(); | ||
|
|
||
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Oops, something went wrong.
Oops, something went wrong.
Add this suggestion to a batch that can be applied as a single commit.
This suggestion is invalid because no changes were made to the code.
Suggestions cannot be applied while the pull request is closed.
Suggestions cannot be applied while viewing a subset of changes.
Only one suggestion per line can be applied in a batch.
Add this suggestion to a batch that can be applied as a single commit.
Applying suggestions on deleted lines is not supported.
You must change the existing code in this line in order to create a valid suggestion.
Outdated suggestions cannot be applied.
This suggestion has been applied or marked resolved.
Suggestions cannot be applied from pending reviews.
Suggestions cannot be applied on multi-line comments.
Suggestions cannot be applied while the pull request is queued to merge.
Suggestion cannot be applied right now. Please check back later.
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
[P1] This enters the coordinator before API-side host validation and before checking whether an exact commit is already complete. For Commit/ShortCommit, the worker calls
materialize_commit/materialize_short_commitdirectly, and those methods do not validate the repo host, so a disallowed host can perform upstream git work before being rejected; complete exact-cache hits also now wait/503 behind unrelated repo-write work. Validate the repo first, keep the complete-manifest fast path outside the lease, and coordinate only missing/incomplete read-through work.There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
Fixed in
23ca917:handle_materialize_requestnow validates the repo before coordinator entry, serves complete exact-commit manifests as an outside-lease fast path, and routes only missing/incomplete read-through work through coordination before local-only response materialization.