Implement multi-worker conflict avoidance

crates/git-cache-api/src/lib.rs

@@ -14,7 +14,7 @@ use git_cache_domain::{
     MaterializerExecutor,
 };
 use git_cache_git::UploadPackProcess;
-use git_cache_worker::{InMemoryRepoLeaseManager, UpdateCoordinator, UpdateDisposition};
+use git_cache_worker::{ObjectStoreRepoLeaseManager, UpdateCoordinator, UpdateDisposition};
 use http::{header, Method, StatusCode, Uri};
 use serde::Serialize;
 use std::collections::HashMap;
@@ -28,6 +28,7 @@ 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;
 
@@ -92,7 +93,10 @@ 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::new(InMemoryRepoLeaseManager::new());
+        let leases = Arc::new(ObjectStoreRepoLeaseManager::new(
+            Arc::clone(&domain.store),
+            &domain.config.leases,
+        ));
         let coordinator = UpdateCoordinator::new(executor, leases);
         Materializer::new(Arc::clone(&domain)).enqueue_pending_generation_scan();
         Ok(Self {
@@ -174,10 +178,7 @@ async fn handle_materialize_request(
         .materialize_total
         .fetch_add(1, Ordering::Relaxed);
 
-    let use_coordinator = matches!(
-        request.selector,
-        Selector::Branch(_) | Selector::DefaultBranch
-    );
+    let use_coordinator = true;
 
     let verified_by_coordinator = if use_coordinator {
         let outcome = state
@@ -186,10 +187,18 @@ async fn handle_materialize_request(
             .await;
         match outcome {
             Ok(o) if o.disposition == UpdateDisposition::LeaseBusy => {
-                return Err(ApiError {
-                    status: StatusCode::SERVICE_UNAVAILABLE,
-                    message: "update in progress, retry later".into(),
-                });
+                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
@@ -372,6 +381,29 @@ async fn git_repo(
             .git_remote_upload_pack_total
             .fetch_add(1, Ordering::Relaxed);
 
+        // Acquire the repo-write lease before processing unknown-want
+        // fetches so that any publish_generation calls inside
+        // ensure_wants_available run under durable coordination.
+        let outcome = state
+            .coordinator
+            .read_through(repo.clone(), Selector::DefaultBranch)
+            .await;
+        match outcome {
+            Ok(o) if o.disposition == UpdateDisposition::LeaseBusy => {
+                let retry_after = state.domain.config.leases.busy_retry_after_seconds;
+                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) => {
+                warn!(%repo, ?error, "coordinator pre-fetch for upload-pack failed; proceeding without lease");
+            }
+            Ok(_) => {}
+        }
+
         match materializer.handle_upload_pack(&repo, &body).await {
             Ok(process) => stream_upload_pack_response(&state, process),
             Err(error) => ApiError::from(error).into_response(),
@@ -492,10 +524,16 @@ impl From<GitCacheError> for ApiError {
             GitCacheError::NotImplemented(_) => StatusCode::NOT_IMPLEMENTED,
             GitCacheError::Validation(_) => StatusCode::BAD_REQUEST,
             GitCacheError::Timeout(_) => StatusCode::GATEWAY_TIMEOUT,
-            GitCacheError::Conflict(_) => StatusCode::CONFLICT,
-            GitCacheError::Internal(_) | GitCacheError::Io(_) | GitCacheError::Json(_) => {
-                StatusCode::INTERNAL_SERVER_ERROR
-            }
+            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 +701,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();

crates/git-cache-api/tests/contention.rs

@@ -84,6 +84,7 @@ impl TestServer {
             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 router = app(config);

crates/git-cache-api/tests/contention_advanced.rs

@@ -74,6 +74,7 @@ impl TestServer {
             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(),
             allowed_upstream_hosts: vec!["github.com".into()],
             disk: git_cache_core::DiskConfig {
                 quota_bytes: 1024 * 1024 * 1024,
@@ -210,6 +211,7 @@ impl MultiRepoTestServer {
             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(),
             allowed_upstream_hosts: vec!["github.com".into()],
             disk: git_cache_core::DiskConfig {
                 quota_bytes: 1024 * 1024 * 1024,

crates/git-cache-api/tests/correctness.rs

@@ -68,6 +68,7 @@ impl TestServer {
             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(),
             allowed_upstream_hosts: vec!["github.com".into()],
             disk: git_cache_core::DiskConfig {
                 quota_bytes: 1024 * 1024 * 1024,

crates/git-cache-api/tests/git_client_advanced.rs

@@ -69,6 +69,7 @@ impl TestServer {
             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(),
             allowed_upstream_hosts: vec!["github.com".into()],
             disk: git_cache_core::DiskConfig {
                 quota_bytes: 1024 * 1024 * 1024,

crates/git-cache-api/tests/git_client_contract.rs

@@ -80,6 +80,7 @@ impl TestServer {
             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 router = app(config);

crates/git-cache-api/tests/git_protocol.rs

@@ -79,6 +79,7 @@ impl TestServer {
             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 router = app(config);

crates/git-cache-api/tests/git_remote_integration.rs

@@ -75,6 +75,7 @@ impl TestServer {
             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 router = app(config);

crates/git-cache-api/tests/git_session_protocol.rs

@@ -79,6 +79,7 @@ impl TestServer {
             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 listener = TcpListener::bind("127.0.0.1:0").await.unwrap();

crates/git-cache-api/tests/load.rs

@@ -45,6 +45,7 @@ impl TestServer {
             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(),
             allowed_upstream_hosts: vec!["github.com".into()],
             disk: git_cache_core::DiskConfig {
                 quota_bytes: 2 * 1024 * 1024 * 1024,

crates/git-cache-api/tests/performance.rs

@@ -80,6 +80,7 @@ impl TestServer {
             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 router = app(config);

crates/git-cache-api/tests/performance_advanced.rs

@@ -67,6 +67,7 @@ impl TestServer {
             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(),
             allowed_upstream_hosts: vec!["github.com".into()],
             disk: git_cache_core::DiskConfig {
                 quota_bytes: 1024 * 1024 * 1024,

crates/git-cache-api/tests/runtime_cache_recovery.rs

@@ -84,6 +84,7 @@ impl TestServer {
             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 router = app(config);

crates/git-cache-core/src/config.rs

@@ -36,6 +36,8 @@ pub struct AppConfig {
     pub session_cleanup_interval_secs: u64,
     #[serde(default = "default_max_concurrent_generation_verifications")]
     pub max_concurrent_generation_verifications: usize,
+    #[serde(default)]
+    pub leases: LeaseConfig,
 }
 
 impl AppConfig {
@@ -122,6 +124,22 @@ impl AppConfig {
                 "GIT_CACHE_MAX_CONCURRENT_GENERATION_VERIFICATIONS",
                 default_max_concurrent_generation_verifications(),
             )?,
+            leases: LeaseConfig {
+                worker_id: env::var("GIT_CACHE_WORKER_ID").ok(),
+                ttl_seconds: parse_env("GIT_CACHE_LEASE_TTL_SECONDS", default_lease_ttl_seconds())?,
+                renew_interval_seconds: parse_env(
+                    "GIT_CACHE_LEASE_RENEW_INTERVAL_SECONDS",
+                    default_lease_renew_interval_seconds(),
+                )?,
+                steal_skew_seconds: parse_env(
+                    "GIT_CACHE_LEASE_STEAL_SKEW_SECONDS",
+                    default_lease_steal_skew_seconds(),
+                )?,
+                busy_retry_after_seconds: parse_env(
+                    "GIT_CACHE_LEASE_BUSY_RETRY_AFTER_SECONDS",
+                    default_lease_busy_retry_after_seconds(),
+                )?,
+            },
         })
     }
 }
@@ -205,6 +223,48 @@ impl Default for CompactionConfig {
     }
 }
 
+#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
+pub struct LeaseConfig {
+    #[serde(default)]
+    pub worker_id: Option<String>,
+    #[serde(default = "default_lease_ttl_seconds")]
+    pub ttl_seconds: u64,
+    #[serde(default = "default_lease_renew_interval_seconds")]
+    pub renew_interval_seconds: u64,
+    #[serde(default = "default_lease_steal_skew_seconds")]
+    pub steal_skew_seconds: u64,
+    #[serde(default = "default_lease_busy_retry_after_seconds")]
+    pub busy_retry_after_seconds: u64,
+}
+
+impl Default for LeaseConfig {
+    fn default() -> Self {
+        Self {
+            worker_id: None,
+            ttl_seconds: default_lease_ttl_seconds(),
+            renew_interval_seconds: default_lease_renew_interval_seconds(),
+            steal_skew_seconds: default_lease_steal_skew_seconds(),
+            busy_retry_after_seconds: default_lease_busy_retry_after_seconds(),
+        }
+    }
+}
+
+pub fn default_lease_ttl_seconds() -> u64 {
+    300
+}
+
+pub fn default_lease_renew_interval_seconds() -> u64 {
+    60
+}
+
+pub fn default_lease_steal_skew_seconds() -> u64 {
+    30
+}
+
+pub fn default_lease_busy_retry_after_seconds() -> u64 {
+    5
+}
+
 fn default_compaction_threshold() -> u32 {
     10
 }
@@ -354,6 +414,11 @@ mod tests {
         "GIT_CACHE_MAX_CONCURRENT_GIT_PROCESSES",
         "GIT_CACHE_SESSION_CLEANUP_INTERVAL_SECS",
         "GIT_CACHE_MAX_CONCURRENT_GENERATION_VERIFICATIONS",
+        "GIT_CACHE_WORKER_ID",
+        "GIT_CACHE_LEASE_TTL_SECONDS",
+        "GIT_CACHE_LEASE_RENEW_INTERVAL_SECONDS",
+        "GIT_CACHE_LEASE_STEAL_SKEW_SECONDS",
+        "GIT_CACHE_LEASE_BUSY_RETRY_AFTER_SECONDS",
     ];
 
     struct EnvGuard {
@@ -452,6 +517,7 @@ min_free_bytes = 100000
         assert_eq!(config.max_git_output_bytes, 16 * 1024 * 1024);
         assert_eq!(config.compaction, CompactionConfig::default());
         assert_eq!(config.max_concurrent_generation_verifications, 1);
+        assert_eq!(config.leases, LeaseConfig::default());
     }
 
     #[test]
@@ -461,6 +527,16 @@ min_free_bytes = 100000
         assert!(!config.inline);
     }
 
+    #[test]
+    fn lease_config_default_values() {
+        let config = LeaseConfig::default();
+        assert_eq!(config.worker_id, None);
+        assert_eq!(config.ttl_seconds, 300);
+        assert_eq!(config.renew_interval_seconds, 60);
+        assert_eq!(config.steal_skew_seconds, 30);
+        assert_eq!(config.busy_retry_after_seconds, 5);
+    }
+
     #[test]
     fn git_remote_config_default_values() {
         let config = GitRemoteConfig::default();
@@ -497,6 +573,11 @@ min_free_bytes = 100000
             ("GIT_CACHE_COMPACTION_CHAIN_DEPTH_THRESHOLD", "4"),
             ("GIT_CACHE_COMPACTION_INLINE", "yes"),
             ("GIT_CACHE_MAX_CONCURRENT_GENERATION_VERIFICATIONS", "3"),
+            ("GIT_CACHE_WORKER_ID", "worker-a"),
+            ("GIT_CACHE_LEASE_TTL_SECONDS", "11"),
+            ("GIT_CACHE_LEASE_RENEW_INTERVAL_SECONDS", "3"),
+            ("GIT_CACHE_LEASE_STEAL_SKEW_SECONDS", "2"),
+            ("GIT_CACHE_LEASE_BUSY_RETRY_AFTER_SECONDS", "1"),
         ]);
 
         let config = AppConfig::from_env().unwrap();
@@ -512,6 +593,11 @@ min_free_bytes = 100000
         assert_eq!(config.compaction.chain_depth_threshold, 4);
         assert!(config.compaction.inline);
         assert_eq!(config.max_concurrent_generation_verifications, 3);
+        assert_eq!(config.leases.worker_id.as_deref(), Some("worker-a"));
+        assert_eq!(config.leases.ttl_seconds, 11);
+        assert_eq!(config.leases.renew_interval_seconds, 3);
+        assert_eq!(config.leases.steal_skew_seconds, 2);
+        assert_eq!(config.leases.busy_retry_after_seconds, 1);
 
         match config.object_store {
             ObjectStoreConfig::S3 {