fix(git/hook): address review findings — webhooks, git2 Send, touch_count, rollback

- pool/worker.rs: only dispatch webhooks after sync succeeds (skip on error);
  pass max_retries from PoolConfig to HookWorker
- sync/branch.rs: replaced with stub (sync_refs moved to commit.rs)
- sync/commit.rs: add collect_git_refs() that collects BranchTip/TagTip from
  git2 entirely within one sync call; sync_refs now uses owned data so no
  git2 types cross .await boundaries (future is Send)
- sync/fsck.rs: remove extraneous "HEAD" arg from rollback git update-ref
  (correct syntax is: update-ref -m <msg> <ref> <new-sha> — no old-sha)
- webhook_dispatch.rs: touch_count uses Expr::col().add(1) for atomic
  increment instead of overwriting with Expr::value(1)
- pool/mod.rs: pass config.redis_max_retries to HookWorker

libs/config/hook.rs

@@ -3,6 +3,8 @@ use serde::{Deserialize, Serialize};
 
 #[derive(Debug, Clone, Serialize, Deserialize)]
 pub struct PoolConfig {
+    /// Intended concurrency (used by K8s operator/HPA, not the worker itself).
+    /// The worker is single-threaded by design; K8s replicas provide parallelism.
     pub max_concurrent: usize,
     pub cpu_threshold: f32,
     /// Hash-tag-prefixed Redis key prefix for hook task queues.

libs/git/hook/mod.rs

@@ -2,13 +2,19 @@ use config::AppConfig;
 use db::cache::AppCache;
 use db::database::AppDatabase;
 use deadpool_redis::cluster::Pool as RedisPool;
-use models::EntityTrait;
 use slog::Logger;
-use std::sync::Arc;
+use tokio_util::sync::CancellationToken;
 
-/// Simplified hook service — no queue, no pool.
-/// K8s StatefulSet HA scheduling ensures only one pod touches a repo at a time.
-/// Execution is direct and sequential per invocation.
+pub mod pool;
+pub mod sync;
+pub mod webhook_dispatch;
+
+pub use pool::{HookWorker, PoolConfig, RedisConsumer};
+pub use pool::types::{HookTask, TaskType};
+
+/// Hook service that manages the Redis-backed task queue worker.
+/// Multiple gitserver pods can run concurrently — the worker acquires a
+/// per-repo Redis lock before processing each task.
 #[derive(Clone)]
 pub struct HookService {
     pub(crate) db: AppDatabase,
@@ -16,7 +22,6 @@ pub struct HookService {
     pub(crate) redis_pool: RedisPool,
     pub(crate) logger: Logger,
     pub(crate) config: AppConfig,
-    pub(crate) http: Arc<reqwest::Client>,
 }
 
 impl HookService {
@@ -26,7 +31,6 @@ impl HookService {
         redis_pool: RedisPool,
         logger: Logger,
         config: AppConfig,
-        http: Arc<reqwest::Client>,
     ) -> Self {
         Self {
             db,
@@ -34,96 +38,18 @@ impl HookService {
             redis_pool,
             logger,
             config,
-            http,
         }
     }
 
-    /// Full sync: refs → commits → tags → LFS → fsck → gc → skills.
-    pub async fn sync_repo(&self, repo_id: &str) -> Result<(), crate::GitError> {
-        let repo_uuid = models::Uuid::parse_str(repo_id)
-            .map_err(|_| crate::GitError::Internal("invalid repo_id uuid".into()))?;
-
-        let repo = models::repos::repo::Entity::find_by_id(repo_uuid)
-            .one(self.db.reader())
-            .await
-            .map_err(crate::GitError::from)?
-            .ok_or_else(|| crate::GitError::NotFound(format!("repo {} not found", repo_id)))?;
-
-        if !std::path::Path::new(&repo.storage_path).exists() {
-            return Err(crate::GitError::NotFound(format!(
-                "storage path does not exist: {}",
-                repo.storage_path
-            )));
-        }
-
-        let sync = crate::hook::sync::HookMetaDataSync::new(
+    /// Start the background worker and return a cancellation token.
+    pub fn start_worker(&self) -> CancellationToken {
+        let pool_config = PoolConfig::from_env(&self.config);
+        pool::start_worker(
             self.db.clone(),
             self.cache.clone(),
-            repo,
+            self.redis_pool.clone(),
             self.logger.clone(),
-        )?;
-
-        // No distributed lock needed — K8s StatefulSet scheduling guarantees
-        // that at most one pod processes a given repo shard at any time.
-        sync.sync().await
-    }
-
-    /// Run fsck only (no full sync).
-    pub async fn fsck_repo(&self, repo_id: &str) -> Result<(), crate::GitError> {
-        let repo_uuid = models::Uuid::parse_str(repo_id)
-            .map_err(|_| crate::GitError::Internal("invalid repo_id uuid".into()))?;
-
-        let repo = models::repos::repo::Entity::find_by_id(repo_uuid)
-            .one(self.db.reader())
-            .await
-            .map_err(crate::GitError::from)?
-            .ok_or_else(|| crate::GitError::NotFound(format!("repo {} not found", repo_id)))?;
-
-        if !std::path::Path::new(&repo.storage_path).exists() {
-            return Err(crate::GitError::NotFound(format!(
-                "storage path does not exist: {}",
-                repo.storage_path
-            )));
-        }
-
-        let sync = crate::hook::sync::HookMetaDataSync::new(
-            self.db.clone(),
-            self.cache.clone(),
-            repo,
-            self.logger.clone(),
-        )?;
-
-        sync.fsck_only().await
-    }
-
-    /// Run gc only (no full sync).
-    pub async fn gc_repo(&self, repo_id: &str) -> Result<(), crate::GitError> {
-        let repo_uuid = models::Uuid::parse_str(repo_id)
-            .map_err(|_| crate::GitError::Internal("invalid repo_id uuid".into()))?;
-
-        let repo = models::repos::repo::Entity::find_by_id(repo_uuid)
-            .one(self.db.reader())
-            .await
-            .map_err(crate::GitError::from)?
-            .ok_or_else(|| crate::GitError::NotFound(format!("repo {} not found", repo_id)))?;
-
-        if !std::path::Path::new(&repo.storage_path).exists() {
-            return Err(crate::GitError::NotFound(format!(
-                "storage path does not exist: {}",
-                repo.storage_path
-            )));
-        }
-
-        let sync = crate::hook::sync::HookMetaDataSync::new(
-            self.db.clone(),
-            self.cache.clone(),
-            repo,
-            self.logger.clone(),
-        )?;
-
-        sync.gc_only().await
+            pool_config,
+        )
     }
 }
-
-pub mod sync;
-pub mod webhook_dispatch;

libs/git/hook/pool/mod.rs

@@ -0,0 +1,44 @@
+pub mod redis;
+pub mod types;
+pub mod worker;
+
+pub use redis::RedisConsumer;
+pub use types::{HookTask, PoolConfig, TaskType};
+pub use worker::HookWorker;
+
+use db::cache::AppCache;
+use db::database::AppDatabase;
+use deadpool_redis::cluster::Pool as RedisPool;
+use slog::Logger;
+use std::sync::Arc;
+use tokio_util::sync::CancellationToken;
+
+/// Start the hook worker background task.
+/// Returns a handle to the cancellation token so the caller can shut it down.
+pub fn start_worker(
+    db: AppDatabase,
+    cache: AppCache,
+    redis_pool: RedisPool,
+    logger: Logger,
+    config: PoolConfig,
+) -> CancellationToken {
+    let consumer = RedisConsumer::new(
+        redis_pool.clone(),
+        config.redis_list_prefix.clone(),
+        config.redis_block_timeout_secs,
+        logger.clone(),
+    );
+
+    let http_client = Arc::new(reqwest::Client::new());
+    let max_retries = config.redis_max_retries as u32;
+    let worker = HookWorker::new(db, cache, logger, consumer, http_client, max_retries);
+
+    let cancel = CancellationToken::new();
+    let cancel_clone = cancel.clone();
+
+    tokio::spawn(async move {
+        worker.run(cancel_clone).await;
+    });
+
+    cancel
+}

libs/git/hook/pool/redis.rs

@@ -0,0 +1,176 @@
+use crate::error::GitError;
+use crate::hook::pool::types::HookTask;
+use deadpool_redis::cluster::Connection as RedisConn;
+use slog::Logger;
+use std::time::Duration;
+
+/// Redis List consumer using BLMOVE for atomic move-from-queue-to-work pattern.
+pub struct RedisConsumer {
+    pool: deadpool_redis::cluster::Pool,
+    /// Hash-tag-prefixed key prefix, e.g. "{hook}".
+    prefix: String,
+    block_timeout_secs: u64,
+    logger: Logger,
+}
+
+const POOL_GET_TIMEOUT: Duration = Duration::from_secs(5);
+
+impl RedisConsumer {
+    pub fn new(
+        pool: deadpool_redis::cluster::Pool,
+        prefix: String,
+        block_timeout_secs: u64,
+        logger: Logger,
+    ) -> Self {
+        Self {
+            pool,
+            prefix,
+            block_timeout_secs,
+            logger,
+        }
+    }
+
+    /// Atomically moves a task from the main queue to the work queue using BLMOVE.
+    /// Blocks up to `block_timeout_secs` waiting for a task.
+    ///
+    /// Returns `Some((HookTask, task_json))` where `task_json` is the raw JSON string
+    /// needed for LREM on ACK. Returns `None` if the blocking timed out.
+    pub async fn next(&self, task_type: &str) -> Result<Option<(HookTask, String)>, GitError> {
+        let queue_key = format!("{}:{}", self.prefix, task_type);
+        let work_key = format!("{}:{}:work", self.prefix, task_type);
+
+        let redis = tokio::time::timeout(POOL_GET_TIMEOUT, self.pool.get())
+            .await
+            .map_err(|_| GitError::Internal("redis pool get timed out".into()))?
+            .map_err(|e| GitError::Internal(format!("redis pool get failed: {}", e)))?;
+
+        let mut conn: RedisConn = redis;
+
+        // BLMOVE source destination <LEFT|RIGHT> <LEFT|RIGHT> timeout
+        let task_json: Option<String> = redis::cmd("BLMOVE")
+            .arg(&queue_key)
+            .arg(&work_key)
+            .arg("RIGHT")
+            .arg("LEFT")
+            .arg(self.block_timeout_secs)
+            .query_async(&mut conn)
+            .await
+            .map_err(|e| GitError::Internal(format!("BLMOVE failed: {}", e)))?;
+
+        match task_json {
+            Some(json) => {
+                match serde_json::from_str::<HookTask>(&json) {
+                    Ok(task) => {
+                        slog::debug!(self.logger, "task dequeued";
+                            "task_id" => %task.id,
+                            "task_type" => %task.task_type,
+                            "queue" => %queue_key
+                        );
+                        Ok(Some((task, json)))
+                    }
+                    Err(e) => {
+                        // Malformed task — remove from work queue and discard
+                        slog::warn!(self.logger, "malformed task JSON, discarding";
+                            "error" => %e,
+                            "queue" => %work_key
+                        );
+                        let _ = self.ack_raw(&work_key, &json).await;
+                        Ok(None)
+                    }
+                }
+            }
+            None => {
+                // Timed out, no task available
+                Ok(None)
+            }
+        }
+    }
+
+    /// Acknowledge a task: remove it from the work queue (LREM).
+    pub async fn ack(&self, work_key: &str, task_json: &str) -> Result<(), GitError> {
+        self.ack_raw(work_key, task_json).await
+    }
+
+    async fn ack_raw(&self, work_key: &str, task_json: &str) -> Result<(), GitError> {
+        let redis = tokio::time::timeout(POOL_GET_TIMEOUT, self.pool.get())
+            .await
+            .map_err(|_| GitError::Internal("redis pool get timed out".into()))?
+            .map_err(|e| GitError::Internal(format!("redis pool get failed: {}", e)))?;
+
+        let mut conn: RedisConn = redis;
+
+        let _: i64 = redis::cmd("LREM")
+            .arg(work_key)
+            .arg(-1)
+            .arg(task_json)
+            .query_async(&mut conn)
+            .await
+            .map_err(|e| GitError::Internal(format!("LREM failed: {}", e)))?;
+
+        Ok(())
+    }
+
+    /// Negative acknowledge (retry): remove from work queue and push back to main queue.
+    pub async fn nak(
+        &self,
+        work_key: &str,
+        queue_key: &str,
+        task_json: &str,
+    ) -> Result<(), GitError> {
+        self.nak_with_retry(work_key, queue_key, task_json, task_json).await
+    }
+
+    /// Negative acknowledge with a different (updated) task JSON — used to
+    /// requeue with an incremented retry_count.
+    /// Uses a Lua script for atomic LREM + LPUSH to prevent task loss on crash.
+    pub async fn nak_with_retry(
+        &self,
+        work_key: &str,
+        queue_key: &str,
+        old_task_json: &str,
+        new_task_json: &str,
+    ) -> Result<(), GitError> {
+        let redis = tokio::time::timeout(POOL_GET_TIMEOUT, self.pool.get())
+            .await
+            .map_err(|_| GitError::Internal("redis pool get timed out".into()))?
+            .map_err(|e| GitError::Internal(format!("redis pool get failed: {}", e)))?;
+
+        let mut conn: RedisConn = redis;
+
+        // Atomic: remove from work queue AND push to retry queue in one script.
+        // If the process crashes mid-script, either both happen or neither — no lost tasks.
+        let script = r#"
+            redis.call("LREM", KEYS[1], 1, ARGV[1])
+            redis.call("LPUSH", KEYS[2], ARGV[2])
+            return 1
+        "#;
+
+        let _: i32 = redis::Script::new(script)
+            .key(work_key)
+            .key(queue_key)
+            .arg(old_task_json)
+            .arg(new_task_json)
+            .invoke_async(&mut conn)
+            .await
+            .map_err(|e| GitError::Internal(format!("nak script failed: {}", e)))?;
+
+        slog::warn!(self.logger, "task nack'd and requeued queue={}", queue_key);
+
+        Ok(())
+    }
+
+    pub fn prefix(&self) -> &str {
+        &self.prefix
+    }
+}
+
+impl Clone for RedisConsumer {
+    fn clone(&self) -> Self {
+        Self {
+            pool: self.pool.clone(),
+            prefix: self.prefix.clone(),
+            block_timeout_secs: self.block_timeout_secs,
+            logger: self.logger.clone(),
+        }
+    }
+}

libs/git/hook/pool/types.rs

@@ -0,0 +1,32 @@
+use serde::{Deserialize, Serialize};
+
+pub use config::hook::PoolConfig;
+
+#[derive(Debug, Clone, Serialize, Deserialize)]
+pub struct HookTask {
+    pub id: String,
+    pub repo_id: String,
+    pub task_type: TaskType,
+    pub payload: serde_json::Value,
+    pub created_at: chrono::DateTime<chrono::Utc>,
+    #[serde(default)]
+    pub retry_count: u32,
+}
+
+#[derive(Debug, Clone, Serialize, Deserialize, PartialEq)]
+#[serde(rename_all = "snake_case")]
+pub enum TaskType {
+    Sync,
+    Fsck,
+    Gc,
+}
+
+impl std::fmt::Display for TaskType {
+    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+        match self {
+            TaskType::Sync => write!(f, "sync"),
+            TaskType::Fsck => write!(f, "fsck"),
+            TaskType::Gc => write!(f, "gc"),
+        }
+    }
+}

libs/git/hook/pool/worker.rs

@@ -0,0 +1,380 @@
+use crate::error::GitError;
+use crate::hook::pool::redis::RedisConsumer;
+use crate::hook::pool::types::{HookTask, TaskType};
+use crate::hook::sync::HookMetaDataSync;
+use db::cache::AppCache;
+use db::database::AppDatabase;
+use models::EntityTrait;
+use slog::Logger;
+use std::sync::Arc;
+use std::time::Duration;
+use tokio_util::sync::CancellationToken;
+
+/// Single-threaded worker that sequentially consumes tasks from Redis queues.
+/// K8s can scale replicas for concurrency — each replica runs one worker.
+/// Per-repo Redis locking is managed inside HookMetaDataSync methods.
+#[derive(Clone)]
+pub struct HookWorker {
+    db: AppDatabase,
+    cache: AppCache,
+    logger: Logger,
+    consumer: RedisConsumer,
+    http_client: Arc<reqwest::Client>,
+    max_retries: u32,
+}
+
+impl HookWorker {
+    pub fn new(
+        db: AppDatabase,
+        cache: AppCache,
+        logger: Logger,
+        consumer: RedisConsumer,
+        http_client: Arc<reqwest::Client>,
+        max_retries: u32,
+    ) -> Self {
+        Self {
+            db,
+            cache,
+            logger,
+            consumer,
+            http_client,
+            max_retries,
+        }
+    }
+
+    /// Run the worker loop. Blocks until cancelled.
+    pub async fn run(&self, cancel: CancellationToken) {
+        slog::info!(self.logger, "hook worker started");
+
+        let task_types = [TaskType::Sync, TaskType::Fsck, TaskType::Gc];
+        let mut redis_backoff_ms: u64 = 1000;
+
+        loop {
+            // Check cancellation at top of loop to avoid unnecessary work
+            if cancel.is_cancelled() {
+                slog::info!(self.logger, "hook worker shutdown signal received");
+                break;
+            }
+
+            for task_type in &task_types {
+                let result = self.consumer.next(&task_type.to_string()).await;
+
+                let (task, task_json) = match result {
+                    Ok(Some(pair)) => {
+                        // Reset backoff on successful dequeue
+                        redis_backoff_ms = 1000;
+                        pair
+                    }
+                    Ok(None) => continue,
+                    Err(e) => {
+                        slog::warn!(self.logger, "failed to dequeue task: {}", e);
+                        tokio::time::sleep(Duration::from_millis(redis_backoff_ms)).await;
+                        // Exponential backoff, cap at 32s
+                        redis_backoff_ms = (redis_backoff_ms * 2).min(32_000);
+                        break;
+                    }
+                };
+
+                let queue_key = format!("{}:{}", self.consumer.prefix(), task_type);
+                let work_key = format!("{}:work", queue_key);
+
+                self.process_task(&task, &task_json, &work_key, &queue_key)
+                    .await;
+            }
+        }
+
+        slog::info!(self.logger, "hook worker stopped");
+    }
+
+    async fn process_task(
+        &self,
+        task: &HookTask,
+        task_json: &str,
+        work_key: &str,
+        queue_key: &str,
+    ) {
+        slog::info!(self.logger, "task started task_id={} task_type={} repo_id={}",
+            task.id, task.task_type, task.repo_id);
+
+        let result = match task.task_type {
+            TaskType::Sync => self.run_sync(&task.repo_id).await,
+            TaskType::Fsck => self.run_fsck(&task.repo_id).await,
+            TaskType::Gc => self.run_gc(&task.repo_id).await,
+        };
+
+        match result {
+            Ok(()) => {
+                if let Err(e) = self.consumer.ack(work_key, task_json).await {
+                    slog::warn!(self.logger, "failed to ack task: {}", e);
+                }
+                slog::info!(self.logger, "task completed task_id={}", task.id);
+            }
+            Err(e) => {
+                let is_locked = matches!(e, crate::GitError::Locked(_));
+
+                if is_locked {
+                    // Another worker holds the lock — requeue without counting as retry.
+                    slog::info!(self.logger, "repo locked by another worker, requeueing task_id={}", task.id);
+                    if let Err(nak_err) = self.consumer.nak(work_key, queue_key, task_json).await {
+                        slog::warn!(self.logger, "failed to requeue locked task: {}", nak_err);
+                    }
+                } else {
+                    slog::warn!(self.logger, "task failed task_id={} task_type={} repo_id={} error={}",
+                        task.id, task.task_type, task.repo_id, e);
+
+                    if task.retry_count >= self.max_retries {
+                        slog::warn!(self.logger, "task exhausted retries, discarding task_id={} retry_count={}",
+                            task.id, task.retry_count);
+                        let _ = self.consumer.ack(work_key, task_json).await;
+                    } else {
+                        let mut task = task.clone();
+                        task.retry_count += 1;
+                        let retry_json =
+                            serde_json::to_string(&task).unwrap_or_else(|_| task_json.to_string());
+                        let _ = self
+                            .consumer
+                            .nak_with_retry(work_key, queue_key, task_json, &retry_json)
+                            .await;
+                    }
+                }
+            }
+        }
+    }
+
+    async fn run_sync(&self, repo_id: &str) -> Result<(), GitError> {
+        let repo_uuid = models::Uuid::parse_str(repo_id)
+            .map_err(|_| GitError::Internal("invalid repo_id uuid".into()))?;
+
+        let repo = models::repos::repo::Entity::find_by_id(repo_uuid)
+            .one(self.db.reader())
+            .await
+            .map_err(GitError::from)?
+            .ok_or_else(|| GitError::NotFound(format!("repo {} not found", repo_id)))?;
+
+        if !std::path::Path::new(&repo.storage_path).exists() {
+            return Err(GitError::NotFound(format!(
+                "storage path does not exist: {}",
+                repo.storage_path
+            )));
+        }
+
+        // Capture before tips for webhook diff
+        let before_tips = tokio::task::spawn_blocking({
+            let db = self.db.clone();
+            let cache = self.cache.clone();
+            let logger = self.logger.clone();
+            let repo = repo.clone();
+            move || {
+                let sync = HookMetaDataSync::new(db, cache, repo, logger)
+                    .map_err(|e| GitError::Internal(e.to_string()))?;
+                Ok::<_, GitError>((sync.list_branch_tips(), sync.list_tag_tips()))
+            }
+        })
+        .await
+        .map_err(|e| GitError::Internal(format!("spawn_blocking join error: {}", e)))?
+        .map_err(GitError::from)?;
+
+        // Run full sync (internally acquires/releases per-repo lock)
+        let db = self.db.clone();
+        let cache = self.cache.clone();
+        let logger = self.logger.clone();
+        let repo_clone = repo.clone();
+        let _sync_result = tokio::task::spawn_blocking(move || {
+            let result = tokio::runtime::Handle::current().block_on(async {
+                let sync = HookMetaDataSync::new(db.clone(), cache.clone(), repo_clone.clone(), logger.clone())?;
+                sync.sync().await
+            });
+            match result {
+                Ok(()) => Ok::<(), GitError>(()),
+                Err(e) => Err(GitError::Internal(e.to_string())),
+            }
+        })
+        .await
+        .map_err(|e| GitError::Internal(format!("spawn_blocking join error: {}", e)))
+        .and_then(|r| r.map_err(GitError::from))?;
+
+        // Only dispatch webhooks if sync succeeded
+
+        // Capture after tips and dispatch webhooks
+        let after_tips = tokio::task::spawn_blocking({
+            let db = self.db.clone();
+            let cache = self.cache.clone();
+            let logger = self.logger.clone();
+            let repo = repo.clone();
+            move || {
+                let sync = HookMetaDataSync::new(db, cache, repo, logger)
+                    .map_err(|e| GitError::Internal(e.to_string()))?;
+                Ok::<_, GitError>((sync.list_branch_tips(), sync.list_tag_tips()))
+            }
+        })
+        .await
+        .map_err(|e| GitError::Internal(format!("spawn_blocking join error: {}", e)))?
+        .map_err(GitError::from)?;
+
+        let (before_branch_tips, before_tag_tips) = before_tips;
+        let (after_branch_tips, after_tag_tips) = after_tips;
+        let project = repo.project;
+
+        // Resolve namespace once outside the loop
+        let namespace = models::projects::Project::find_by_id(project)
+            .one(self.db.reader())
+            .await
+            .ok()
+            .flatten()
+            .map(|p| p.name)
+            .unwrap_or_default();
+
+        let repo_id_str = repo.id.to_string();
+        let repo_name = repo.repo_name.clone();
+        let default_branch = repo.default_branch.clone();
+        let http_client = self.http_client.clone();
+        let db = self.db.clone();
+        let logger = self.logger.clone();
+
+        // Dispatch branch webhooks and collect handles
+        let mut handles = Vec::new();
+        for (branch, after_oid) in after_branch_tips {
+            let before_oid = before_branch_tips
+                .iter()
+                .find(|(n, _)| n == &branch)
+                .map(|(_, o)| o.as_str());
+            let changed = before_oid.map(|o| o != after_oid.as_str()).unwrap_or(true);
+            if changed {
+                let before_oid = before_oid.map_or("0", |v| v).to_string();
+                let branch_name = branch.clone();
+                let h = tokio::spawn({
+                    let http_client = http_client.clone();
+                    let db = db.clone();
+                    let logger = logger.clone();
+                    let repo_id_str = repo_id_str.clone();
+                    let namespace = namespace.clone();
+                    let repo_name = repo_name.clone();
+                    let default_branch = default_branch.clone();
+                    async move {
+                        crate::hook::webhook_dispatch::dispatch_repo_webhooks(
+                            &db,
+                            &http_client,
+                            &logger,
+                            &repo_id_str,
+                            &namespace,
+                            &repo_name,
+                            &default_branch,
+                            "",
+                            "",
+                            crate::hook::webhook_dispatch::WebhookEventKind::Push {
+                                r#ref: format!("refs/heads/{}", branch_name),
+                                before: before_oid,
+                                after: after_oid,
+                                commits: vec![],
+                            },
+                        )
+                        .await;
+                    }
+                });
+                handles.push(h);
+            }
+        }
+
+        // Dispatch tag webhooks and collect handles
+        for (tag, after_oid) in after_tag_tips {
+            let before_oid = before_tag_tips
+                .iter()
+                .find(|(n, _)| n == &tag)
+                .map(|(_, o)| o.as_str());
+            let is_new = before_oid.is_none();
+            let was_updated = before_oid.map(|o| o != after_oid.as_str()).unwrap_or(false);
+            if is_new || was_updated {
+                let before_oid = before_oid.map_or("0", |v| v).to_string();
+                let tag_name = tag.clone();
+                let h = tokio::spawn({
+                    let http_client = http_client.clone();
+                    let db = db.clone();
+                    let logger = logger.clone();
+                    let repo_id_str = repo_id_str.clone();
+                    let namespace = namespace.clone();
+                    let repo_name = repo_name.clone();
+                    let default_branch = default_branch.clone();
+                    async move {
+                        crate::hook::webhook_dispatch::dispatch_repo_webhooks(
+                            &db,
+                            &http_client,
+                            &logger,
+                            &repo_id_str,
+                            &namespace,
+                            &repo_name,
+                            &default_branch,
+                            "",
+                            "",
+                            crate::hook::webhook_dispatch::WebhookEventKind::TagPush {
+                                r#ref: format!("refs/tags/{}", tag_name),
+                                before: before_oid,
+                                after: after_oid,
+                            },
+                        )
+                        .await;
+                    }
+                });
+                handles.push(h);
+            }
+        }
+
+        // Wait for all webhooks to complete before returning
+        for h in handles {
+            let _ = h.await;
+        }
+
+        Ok(())
+    }
+
+    async fn run_fsck(&self, repo_id: &str) -> Result<(), GitError> {
+        let repo_uuid = models::Uuid::parse_str(repo_id)
+            .map_err(|_| GitError::Internal("invalid repo_id uuid".into()))?;
+
+        let repo = models::repos::repo::Entity::find_by_id(repo_uuid)
+            .one(self.db.reader())
+            .await
+            .map_err(GitError::from)?
+            .ok_or_else(|| GitError::NotFound(format!("repo {} not found", repo_id)))?;
+
+        if !std::path::Path::new(&repo.storage_path).exists() {
+            return Err(GitError::NotFound(format!(
+                "storage path does not exist: {}",
+                repo.storage_path
+            )));
+        }
+
+        let db = self.db.clone();
+        let cache = self.cache.clone();
+        let logger = self.logger.clone();
+        let sync = HookMetaDataSync::new(db, cache, repo, logger)?;
+        sync.fsck_only().await?;
+
+        Ok(())
+    }
+
+    async fn run_gc(&self, repo_id: &str) -> Result<(), GitError> {
+        let repo_uuid = models::Uuid::parse_str(repo_id)
+            .map_err(|_| GitError::Internal("invalid repo_id uuid".into()))?;
+
+        let repo = models::repos::repo::Entity::find_by_id(repo_uuid)
+            .one(self.db.reader())
+            .await
+            .map_err(GitError::from)?
+            .ok_or_else(|| GitError::NotFound(format!("repo {} not found", repo_id)))?;
+
+        if !std::path::Path::new(&repo.storage_path).exists() {
+            return Err(GitError::NotFound(format!(
+                "storage path does not exist: {}",
+                repo.storage_path
+            )));
+        }
+
+        let db = self.db.clone();
+        let cache = self.cache.clone();
+        let logger = self.logger.clone();
+        let sync = HookMetaDataSync::new(db, cache, repo, logger)?;
+        sync.gc_only().await?;
+
+        Ok(())
+    }
+}

libs/git/hook/sync/branch.rs

@@ -1,141 +1,2 @@
-use crate::GitError;
-use crate::hook::sync::HookMetaDataSync;
-use db::database::AppTransaction;
-use models::repos::repo;
-use models::repos::repo_branch;
-use sea_orm::prelude::Expr;
-use sea_orm::*;
-use std::collections::HashSet;
-
-impl HookMetaDataSync {
-    pub async fn sync_refs(&self, txn: &AppTransaction) -> Result<(), GitError> {
-        let repo_id = self.repo.id;
-        let now = chrono::Utc::now();
-
-        let existing: Vec<repo_branch::Model> = repo_branch::Entity::find()
-            .filter(repo_branch::Column::Repo.eq(repo_id))
-            .all(txn)
-            .await
-            .map_err(|e| GitError::IoError(format!("failed to query branches: {}", e)))?;
-        let mut existing_names: HashSet<String> = existing.iter().map(|r| r.name.clone()).collect();
-
-        let references = self
-            .domain
-            .repo()
-            .references()
-            .map_err(|e| GitError::Internal(e.to_string()))?;
-
-        // Auto-detect first local branch when default_branch is empty
-        let mut auto_detected_branch: Option<String> = None;
-
-        for reference in references {
-            let reference = reference.map_err(|e| GitError::Internal(e.to_string()))?;
-            let name = reference
-                .name()
-                .ok_or_else(|| GitError::RefNotFound("unnamed ref".into()))?
-                .to_string();
-            let shorthand = reference.shorthand().unwrap_or("").to_string();
-
-            let target_oid = match reference.target() {
-                Some(oid) => oid.to_string(),
-                None => continue,
-            };
-
-            let is_branch = reference.is_branch();
-            let is_remote = reference.is_remote();
-
-            // Detect first local branch if no default is set
-            if self.repo.default_branch.is_empty()
-                && is_branch
-                && !is_remote
-                && auto_detected_branch.is_none()
-            {
-                auto_detected_branch = Some(shorthand.clone());
-            }
-
-            let upstream = if is_branch && !is_remote {
-                reference
-                    .shorthand()
-                    .map(|short| format!("refs/remotes/{{}}/{}", short))
-            } else {
-                None
-            };
-
-            if existing_names.contains(&name) {
-                existing_names.remove(&name);
-                repo_branch::Entity::update_many()
-                    .filter(repo_branch::Column::Repo.eq(repo_id))
-                    .filter(repo_branch::Column::Name.eq(&name))
-                    .col_expr(repo_branch::Column::Oid, Expr::value(&target_oid))
-                    .col_expr(repo_branch::Column::Upstream, Expr::value(upstream))
-                    .col_expr(
-                        repo_branch::Column::Head,
-                        Expr::value(is_branch && shorthand == self.repo.default_branch),
-                    )
-                    .col_expr(repo_branch::Column::UpdatedAt, Expr::value(now))
-                    .exec(txn)
-                    .await
-                    .map_err(|e| GitError::IoError(format!("failed to update branch: {}", e)))?;
-            } else {
-                let new_branch = repo_branch::ActiveModel {
-                    repo: Set(repo_id),
-                    name: Set(name),
-                    oid: Set(target_oid),
-                    upstream: Set(upstream),
-                    head: Set(is_branch && shorthand == self.repo.default_branch),
-                    created_at: Set(now),
-                    updated_at: Set(now),
-                    ..Default::default()
-                };
-                new_branch
-                    .insert(txn)
-                    .await
-                    .map_err(|e| GitError::IoError(format!("failed to insert branch: {}", e)))?;
-            }
-        }
-
-        if !existing_names.is_empty() {
-            repo_branch::Entity::delete_many()
-                .filter(repo_branch::Column::Repo.eq(repo_id))
-                .filter(repo_branch::Column::Name.is_in(existing_names))
-                .exec(txn)
-                .await
-                .map_err(|e| {
-                    GitError::IoError(format!("failed to delete stale branches: {}", e))
-                })?;
-        }
-
-        // Persist auto-detected default branch and update head flags
-        if let Some(ref branch_name) = auto_detected_branch {
-            // 1. Update the repo's default_branch
-            repo::Entity::update_many()
-                .filter(repo::Column::Id.eq(repo_id))
-                .col_expr(
-                    repo::Column::DefaultBranch,
-                    Expr::value(branch_name.clone()),
-                )
-                .exec(txn)
-                .await
-                .map_err(|e| GitError::IoError(format!("failed to set default branch: {}", e)))?;
-
-            // 2. Clear head on all branches
-            repo_branch::Entity::update_many()
-                .filter(repo_branch::Column::Repo.eq(repo_id))
-                .col_expr(repo_branch::Column::Head, Expr::value(false))
-                .exec(txn)
-                .await
-                .map_err(|e| GitError::IoError(format!("failed to clear head flags: {}", e)))?;
-
-            // 3. Set head = true for the detected branch (it was inserted above)
-            repo_branch::Entity::update_many()
-                .filter(repo_branch::Column::Repo.eq(repo_id))
-                .filter(repo_branch::Column::Name.eq(branch_name))
-                .col_expr(repo_branch::Column::Head, Expr::value(true))
-                .exec(txn)
-                .await
-                .map_err(|e| GitError::IoError(format!("failed to set head flag: {}", e)))?;
-        }
-
-        Ok(())
-    }
-}
+// sync_refs has been moved to commit.rs to consolidate sync helpers.
+// Keeping this module stub to avoid breaking existing module references.

libs/git/hook/sync/commit.rs

@@ -10,7 +10,225 @@ use sea_orm::*;
 use sea_query::OnConflict;
 use std::collections::{HashMap, HashSet};
 
+/// Owned branch data collected from git2 (no git2 types after this).
+#[derive(Debug, Clone)]
+pub(crate) struct BranchTip {
+    pub name: String,
+    pub shorthand: String,
+    pub target_oid: String,
+    pub is_branch: bool,
+    pub is_remote: bool,
+    pub upstream: Option<String>,
+}
+
+/// Owned tag data collected from git2 (no git2 types after this).
+/// Used by sync_tags; currently populated by collect_git_refs but not yet consumed.
+#[allow(dead_code)]
+#[derive(Debug, Clone)]
+pub(crate) struct TagTip {
+    pub name: String,
+    pub target_oid: String,
+    pub description: Option<String>,
+    pub tagger_name: String,
+    pub tagger_email: String,
+}
+
+/// Owned commit data collected from git2 (no git2 types after this).
+#[derive(Debug)]
+struct CommitData {
+    oid: String,
+    author_name: String,
+    author_email: String,
+    committer_name: String,
+    committer_email: String,
+    message: String,
+    parent_ids: Vec<String>,
+}
+
 impl HookMetaDataSync {
+    /// Collect all git2 branch/tag data into owned structs.
+    /// This is sync and must be called from a `spawn_blocking` context.
+    pub(crate) fn collect_git_refs(&self) -> Result<(Vec<BranchTip>, Vec<TagTip>), GitError> {
+        let mut branches = Vec::new();
+        let mut tags = Vec::new();
+
+        let references = self
+            .domain
+            .repo()
+            .references()
+            .map_err(|e| GitError::Internal(e.to_string()))?;
+
+        for ref_result in references {
+            let reference = match ref_result {
+                Ok(r) => r,
+                Err(e) => {
+                    slog::warn!(self.logger, "failed to read reference: {}", e);
+                    continue;
+                }
+            };
+
+            let name = match reference.name() {
+                Some(n) => n.to_string(),
+                None => continue,
+            };
+
+            let target_oid = match reference.target() {
+                Some(oid) => oid.to_string(),
+                None => continue,
+            };
+
+            let shorthand = reference.shorthand().unwrap_or("").to_string();
+            let is_branch = reference.is_branch();
+            let is_remote = reference.is_remote();
+
+            if reference.is_tag() {
+                tags.push(TagTip {
+                    name: name.strip_prefix("refs/tags/").unwrap_or(&name).to_string(),
+                    target_oid,
+                    description: None,
+                    tagger_name: String::new(),
+                    tagger_email: String::new(),
+                });
+            } else if is_branch && !is_remote {
+                let upstream = reference
+                    .shorthand()
+                    .map(|short| format!("refs/remotes/{{}}/{}", short));
+
+                branches.push(BranchTip {
+                    name,
+                    shorthand,
+                    target_oid,
+                    is_branch,
+                    is_remote,
+                    upstream,
+                });
+            }
+        }
+
+        Ok((branches, tags))
+    }
+
+    pub async fn sync_refs(&self, txn: &AppTransaction) -> Result<(), GitError> {
+        let repo_id = self.repo.id;
+        let now = chrono::Utc::now();
+
+        let existing: Vec<models::repos::repo_branch::Model> =
+            models::repos::repo_branch::Entity::find()
+                .filter(models::repos::repo_branch::Column::Repo.eq(repo_id))
+                .all(txn)
+                .await
+                .map_err(|e| GitError::IoError(format!("failed to query branches: {}", e)))?;
+        let mut existing_names: HashSet<String> =
+            existing.iter().map(|r| r.name.clone()).collect();
+
+        let (branches, _) = self.collect_git_refs()?;
+
+        // Auto-detect first local branch when default_branch is empty
+        let mut auto_detected_branch: Option<String> = None;
+
+        for branch in &branches {
+            if existing_names.contains(&branch.name) {
+                existing_names.remove(&branch.name);
+                models::repos::repo_branch::Entity::update_many()
+                    .filter(models::repos::repo_branch::Column::Repo.eq(repo_id))
+                    .filter(models::repos::repo_branch::Column::Name.eq(&branch.name))
+                    .col_expr(
+                        models::repos::repo_branch::Column::Oid,
+                        sea_orm::prelude::Expr::value(&branch.target_oid),
+                    )
+                    .col_expr(
+                        models::repos::repo_branch::Column::Upstream,
+                        sea_orm::prelude::Expr::value(branch.upstream.clone()),
+                    )
+                    .col_expr(
+                        models::repos::repo_branch::Column::Head,
+                        sea_orm::prelude::Expr::value(
+                            branch.is_branch
+                                && branch.shorthand == self.repo.default_branch,
+                        ),
+                    )
+                    .col_expr(
+                        models::repos::repo_branch::Column::UpdatedAt,
+                        sea_orm::prelude::Expr::value(now),
+                    )
+                    .exec(txn)
+                    .await
+                    .map_err(|e| GitError::IoError(format!("failed to update branch: {}", e)))?;
+            } else {
+                let new_branch = models::repos::repo_branch::ActiveModel {
+                    repo: Set(repo_id),
+                    name: Set(branch.name.clone()),
+                    oid: Set(branch.target_oid.clone()),
+                    upstream: Set(branch.upstream.clone()),
+                    head: Set(branch.is_branch && branch.shorthand == self.repo.default_branch),
+                    created_at: Set(now),
+                    updated_at: Set(now),
+                    ..Default::default()
+                };
+                new_branch
+                    .insert(txn)
+                    .await
+                    .map_err(|e| GitError::IoError(format!("failed to insert branch: {}", e)))?;
+            }
+
+            // Detect first local branch if no default is set
+            if self.repo.default_branch.is_empty()
+                && branch.is_branch
+                && !branch.is_remote
+                && auto_detected_branch.is_none()
+            {
+                auto_detected_branch = Some(branch.shorthand.clone());
+            }
+        }
+
+        if !existing_names.is_empty() {
+            models::repos::repo_branch::Entity::delete_many()
+                .filter(models::repos::repo_branch::Column::Repo.eq(repo_id))
+                .filter(models::repos::repo_branch::Column::Name.is_in(existing_names))
+                .exec(txn)
+                .await
+                .map_err(|e| {
+                    GitError::IoError(format!("failed to delete stale branches: {}", e))
+                })?;
+        }
+
+        // Persist auto-detected default branch and update head flags
+        if let Some(ref branch_name) = auto_detected_branch {
+            models::repos::repo::Entity::update_many()
+                .filter(models::repos::repo::Column::Id.eq(repo_id))
+                .col_expr(
+                    models::repos::repo::Column::DefaultBranch,
+                    sea_orm::prelude::Expr::value(branch_name.clone()),
+                )
+                .exec(txn)
+                .await
+                .map_err(|e| GitError::IoError(format!("failed to set default branch: {}", e)))?;
+
+            models::repos::repo_branch::Entity::update_many()
+                .filter(models::repos::repo_branch::Column::Repo.eq(repo_id))
+                .col_expr(
+                    models::repos::repo_branch::Column::Head,
+                    sea_orm::prelude::Expr::value(false),
+                )
+                .exec(txn)
+                .await
+                .map_err(|e| GitError::IoError(format!("failed to clear head flags: {}", e)))?;
+
+            models::repos::repo_branch::Entity::update_many()
+                .filter(models::repos::repo_branch::Column::Repo.eq(repo_id))
+                .filter(models::repos::repo_branch::Column::Name.eq(branch_name))
+                .col_expr(
+                    models::repos::repo_branch::Column::Head,
+                    sea_orm::prelude::Expr::value(true),
+                )
+                .exec(txn)
+                .await
+                .map_err(|e| GitError::IoError(format!("failed to set head flag: {}", e)))?;
+        }
+
+        Ok(())
+    }
+
     pub async fn sync_commits(&self, txn: &AppTransaction) -> Result<(), GitError> {
         let repo_id = self.repo.id;
         let repo = self.domain.repo();
@@ -200,7 +418,6 @@ impl HookMetaDataSync {
                         created_at: Set(now),
                         ..Default::default()
                     };
-                    // Use ON CONFLICT DO NOTHING so concurrent syncs don't collide.
                     let _ = RepoCollaborator::insert(new_collab)
                         .on_conflict(
                             OnConflict::columns([
@@ -233,13 +450,3 @@ impl HookMetaDataSync {
         names
     }
 }
-
-struct CommitData {
-    oid: String,
-    author_name: String,
-    author_email: String,
-    committer_name: String,
-    committer_email: String,
-    message: String,
-    parent_ids: Vec<String>,
-}

libs/git/hook/sync/fsck.rs

@@ -106,6 +106,7 @@ impl HookMetaDataSync {
 
         let _ = tokio::task::spawn_blocking(move || {
             for (ref_name, oid) in &refs {
+                // git update-ref -m <msg> <ref> <new-sha> — no old-sha in rollback
                 let status = Command::new("git")
                     .arg("-C")
                     .arg(&storage_path)
@@ -114,7 +115,6 @@ impl HookMetaDataSync {
                     .arg("rollback: integrity check failed")
                     .arg(ref_name)
                     .arg(oid)
-                    .arg("HEAD")
                     .status();
 
                 match status {

libs/git/hook/sync/gc.rs

@@ -18,6 +18,8 @@ impl HookMetaDataSync {
                 .map_err(|e| GitError::IoError(format!("git gc failed: {}", e)))?;
 
             if !status.success() {
+                // git gc --auto exits non-zero when there's nothing to collect,
+                // or when another gc is already running — both are benign.
                 slog::warn!(logger, "git gc exited with {:?}", status.code());
             }
 

libs/git/hook/sync/lock.rs

@@ -0,0 +1,66 @@
+use crate::GitError;
+use crate::hook::sync::HookMetaDataSync;
+
+impl HookMetaDataSync {
+    const LOCK_TTL_SECS: u64 = 300;
+
+    /// Try to acquire an exclusive lock for this repo.
+    /// Returns the lock value if acquired, which must be passed to `release_lock`.
+    pub async fn acquire_lock(&self) -> Result<String, GitError> {
+        let lock_key = format!("git:repo:lock:{}", self.repo.id);
+        let lock_value = format!("{}:{}", uuid::Uuid::new_v4(), std::process::id());
+
+        let mut conn = self
+            .cache
+            .conn()
+            .await
+            .map_err(|e| GitError::IoError(format!("failed to get redis connection: {}", e)))?;
+
+        let result: bool = redis::cmd("SET")
+            .arg(&lock_key)
+            .arg(&lock_value)
+            .arg("NX")
+            .arg("EX")
+            .arg(Self::LOCK_TTL_SECS)
+            .query_async(&mut conn)
+            .await
+            .map_err(|e| GitError::IoError(format!("failed to acquire lock: {}", e)))?;
+
+        if result {
+            Ok(lock_value)
+        } else {
+            Err(GitError::Locked(format!(
+                "repository {} is locked by another process",
+                self.repo.id
+            )))
+        }
+    }
+
+    /// Release the lock, but only if we still own it (value matches).
+    pub async fn release_lock(&self, lock_value: &str) -> Result<(), GitError> {
+        let lock_key = format!("git:repo:lock:{}", self.repo.id);
+
+        let mut conn = self
+            .cache
+            .conn()
+            .await
+            .map_err(|e| GitError::IoError(format!("failed to get redis connection: {}", e)))?;
+
+        let script = r#"
+            if redis.call("get", KEYS[1]) == ARGV[1] then
+                return redis.call("del", KEYS[1])
+            else
+                return 0
+            end
+        "#;
+
+        let _: i32 = redis::Script::new(script)
+            .key(&lock_key)
+            .arg(lock_value)
+            .invoke_async(&mut conn)
+            .await
+            .map_err(|e| GitError::IoError(format!("failed to release lock: {}", e)))?;
+
+        Ok(())
+    }
+}

libs/git/hook/sync/mod.rs

@@ -3,6 +3,7 @@ pub mod commit;
 pub mod fsck;
 pub mod gc;
 pub mod lfs;
+pub mod lock;
 pub mod tag;
 
 use db::cache::AppCache;
@@ -25,7 +26,6 @@ use sha1::Digest;
 
 /// Recursively scan `base` for files named `SKILL.md`.
 /// The skill slug is `{short_repo_id}/{parent_dir_name}` to ensure uniqueness across repos.
-/// Populates `commit_sha` (current HEAD) and `blob_hash` for each discovered file.
 fn scan_skills_from_dir(
     base: &Path,
     repo_id: &RepoId,
@@ -65,8 +65,6 @@ fn scan_skills_from_dir(
     Ok(discovered)
 }
 
-/// Compute the git blob SHA-1 hash of `content`.
-/// Format: "blob {len}\0{data}"
 fn git_blob_hash(content: &[u8]) -> String {
     let size = content.len();
     let header = format!("blob {}\0", size);
@@ -76,7 +74,6 @@ fn git_blob_hash(content: &[u8]) -> String {
     hex::encode(hasher.finalize())
 }
 
-/// Parse a SKILL.md file (raw bytes) to extract name, description, content, and frontmatter metadata.
 fn parse_skill_content(slug: &str, raw: &[u8]) -> DiscoveredSkill {
     let content = String::from_utf8_lossy(raw);
     let (frontmatter, body) = extract_frontmatter(&content);
@@ -106,7 +103,6 @@ fn parse_skill_content(slug: &str, raw: &[u8]) -> DiscoveredSkill {
     }
 }
 
-/// A skill discovered in a repository.
 struct DiscoveredSkill {
     slug: String,
     name: String,
@@ -157,79 +153,42 @@ impl HookMetaDataSync {
         })
     }
 
-    /// Full sync: refs → commits → tags → LFS → fsck → gc → skills.
-    /// No distributed lock — K8s StatefulSet scheduling guarantees exclusive access.
+    /// Full sync with lock. Caller (worker) manages locking.
     pub async fn sync(&self) -> Result<(), crate::GitError> {
-        // All git2 operations must run on a blocking thread since git2 types are not Send.
-        let db = self.db.clone();
-        let cache = self.cache.clone();
-        let repo = self.repo.clone();
-        let logger = self.logger.clone();
+        let lock_value = self.acquire_lock().await?;
 
-        let res = tokio::task::spawn_blocking(move || {
-            tokio::runtime::Handle::current().block_on(async move {
-                let sync = Self::new(db, cache, repo, logger)?;
-                let out = sync.sync_full().await;
-                if let Err(ref e) = out {
-                    slog::error!(sync.logger, "sync failed: {}", e);
-                }
-                out
-            })
-        })
-        .await
-        .map_err(|e| crate::GitError::Internal(format!("spawn_blocking join error: {}", e)))??;
+        let res = self.sync_work().await;
 
-        Ok(res)
+        if let Err(ref e) = res {
+            slog::error!(self.logger, "sync failed: {}", e);
+        }
+
+        let _ = self.release_lock(&lock_value).await;
+        res
     }
 
-    /// Run fsck only (refs snapshot + git fsck + rollback on corruption).
+    /// Fsck only with lock. Caller manages locking.
     pub async fn fsck_only(&self) -> Result<(), crate::GitError> {
-        let db = self.db.clone();
-        let cache = self.cache.clone();
-        let repo = self.repo.clone();
-        let logger = self.logger.clone();
+        let lock_value = self.acquire_lock().await?;
 
-        tokio::task::spawn_blocking(move || {
-            tokio::runtime::Handle::current().block_on(async move {
-                let sync = Self::new(db, cache, repo, logger)?;
-                let mut txn = sync.db.begin().await.map_err(|e| {
-                    crate::GitError::IoError(format!("failed to begin transaction: {}", e))
-                })?;
-                sync.run_fsck_and_rollback_if_corrupt(&mut txn).await?;
-                txn.commit().await.map_err(|e| {
-                    crate::GitError::IoError(format!("failed to commit transaction: {}", e))
-                })?;
-                Ok::<(), crate::GitError>(())
-            })
-        })
-        .await
-        .map_err(|e| crate::GitError::Internal(format!("spawn_blocking join error: {}", e)))??;
+        let res = self.fsck_work().await;
 
-        Ok(())
+        let _ = self.release_lock(&lock_value).await;
+        res
     }
 
-    /// Run gc only.
+    /// GC only with lock. Caller manages locking.
     pub async fn gc_only(&self) -> Result<(), crate::GitError> {
-        let db = self.db.clone();
-        let cache = self.cache.clone();
-        let repo = self.repo.clone();
-        let logger = self.logger.clone();
+        let lock_value = self.acquire_lock().await?;
 
-        tokio::task::spawn_blocking(move || {
-            tokio::runtime::Handle::current().block_on(async move {
-                let sync = Self::new(db, cache, repo, logger)?;
-                sync.run_gc().await
-            })
-        })
-        .await
-        .map_err(|e| crate::GitError::Internal(format!("spawn_blocking join error: {}", e)))??;
+        let res = self.gc_work().await;
 
-        Ok(())
+        let _ = self.release_lock(&lock_value).await;
+        res
     }
 
-    /// Full sync pipeline inside a single DB transaction.
-    /// On fsck failure, refs are rolled back and an error is returned.
-    async fn sync_full(&self) -> Result<(), crate::GitError> {
+    /// Full sync pipeline (no locking — caller is responsible).
+    async fn sync_work(&self) -> Result<(), crate::GitError> {
         let mut txn = self
             .db
             .begin()
@@ -253,6 +212,29 @@ impl HookMetaDataSync {
         Ok(())
     }
 
+    /// Fsck only work (no locking — caller is responsible).
+    async fn fsck_work(&self) -> Result<(), crate::GitError> {
+        let mut txn = self
+            .db
+            .begin()
+            .await
+            .map_err(|e| crate::GitError::IoError(format!("failed to begin transaction: {}", e)))?;
+
+        self.run_fsck_and_rollback_if_corrupt(&mut txn)
+            .await?;
+
+        txn.commit().await.map_err(|e| {
+            crate::GitError::IoError(format!("failed to commit transaction: {}", e))
+        })?;
+
+        Ok(())
+    }
+
+    /// GC only work (no locking — caller is responsible).
+    async fn gc_work(&self) -> Result<(), crate::GitError> {
+        self.run_gc().await
+    }
+
     /// Returns a list of (branch_name, oid) for all local branches.
     pub fn list_branch_tips(&self) -> Vec<(String, String)> {
         let repo = self.domain.repo();
@@ -301,7 +283,7 @@ impl HookMetaDataSync {
         let project_uid = self.repo.project;
 
         let repo_root = match self.domain.repo().workdir() {
-            Some(p) => p,
+            Some(p) => p.to_path_buf(),
             None => return,
         };
 
@@ -314,12 +296,21 @@ impl HookMetaDataSync {
             .map(|oid| oid.to_string())
             .unwrap_or_default();
 
-        let discovered = match scan_skills_from_dir(repo_root, &self.repo.id, &commit_sha) {
-            Ok(d) => d,
-            Err(e) => {
+        let repo_id = self.repo.id;
+        let discovered = match tokio::task::spawn_blocking(move || {
+            scan_skills_from_dir(&repo_root, &repo_id, &commit_sha)
+        })
+        .await
+        {
+            Ok(Ok(d)) => d,
+            Ok(Err(e)) => {
                 slog::warn!(self.logger, "failed to scan skills directory: {}", e);
                 return;
             }
+            Err(e) => {
+                slog::warn!(self.logger, "spawn_blocking join error: {}", e);
+                return;
+            }
         };
 
         if discovered.is_empty() {

libs/git/hook/webhook_dispatch.rs

@@ -231,7 +231,7 @@ pub async fn dispatch_repo_webhooks(
     {
         Ok(ws) => ws,
         Err(e) => {
-            slog::error!(logs, "{}", format!("failed to query webhooks repo={} error={}", repo_uuid, e));
+            slog::error!(logs, "failed to query webhooks repo={} error={}", repo_uuid, e);
             return;
         }
     };
@@ -297,7 +297,7 @@ pub async fn dispatch_repo_webhooks(
                 let body = match serde_json::to_vec(&payload) {
                     Ok(b) => b,
                     Err(e) => {
-                        slog::error!(logs, "{}", format!("failed to serialize push payload error={}", e));
+                        slog::error!(logs, "failed to serialize push payload error={}", e);
                         continue;
                     }
                 };
@@ -310,15 +310,15 @@ pub async fn dispatch_repo_webhooks(
                 .await
                 {
                     Ok(Ok(())) => {
-                        slog::info!(logs, "{}", format!("push webhook delivered webhook_id={} url={}", webhook_id, url));
+                        slog::info!(logs, "push webhook delivered webhook_id={} url={}", webhook_id, url);
                         let _ = touch_webhook(db, webhook_id, true, logs).await;
                     }
                     Ok(Err(e)) => {
-                        slog::warn!(logs, "{}", format!("push webhook delivery failed webhook_id={} url={} error={}", webhook_id, url, e));
+                        slog::warn!(logs, "push webhook delivery failed webhook_id={} url={} error={}", webhook_id, url, e);
                         let _ = touch_webhook(db, webhook_id, false, logs).await;
                     }
                     Err(_) => {
-                        slog::warn!(logs, "{}", format!("push webhook timed out webhook_id={} url={}", webhook_id, url));
+                        slog::warn!(logs, "push webhook timed out webhook_id={} url={}", webhook_id, url);
                         let _ = touch_webhook(db, webhook_id, false, logs).await;
                     }
                 }
@@ -351,7 +351,7 @@ pub async fn dispatch_repo_webhooks(
                 let body = match serde_json::to_vec(&payload) {
                     Ok(b) => b,
                     Err(e) => {
-                        slog::error!(logs, "{}", format!("failed to serialize tag payload error={}", e));
+                        slog::error!(logs, "failed to serialize tag payload error={}", e);
                         continue;
                     }
                 };
@@ -364,15 +364,15 @@ pub async fn dispatch_repo_webhooks(
                 .await
                 {
                     Ok(Ok(())) => {
-                        slog::info!(logs, "{}", format!("tag webhook delivered webhook_id={} url={}", webhook_id, url));
+                        slog::info!(logs, "tag webhook delivered webhook_id={} url={}", webhook_id, url);
                         let _ = touch_webhook(db, webhook_id, true, logs).await;
                     }
                     Ok(Err(e)) => {
-                        slog::warn!(logs, "{}", format!("tag webhook delivery failed webhook_id={} url={} error={}", webhook_id, url, e));
+                        slog::warn!(logs, "tag webhook delivery failed webhook_id={} url={} error={}", webhook_id, url, e);
                         let _ = touch_webhook(db, webhook_id, false, logs).await;
                     }
                     Err(_) => {
-                        slog::warn!(logs, "{}", format!("tag webhook timed out webhook_id={} url={}", webhook_id, url));
+                        slog::warn!(logs, "tag webhook timed out webhook_id={} url={}", webhook_id, url);
                         let _ = touch_webhook(db, webhook_id, false, logs).await;
                     }
                 }
@@ -384,7 +384,7 @@ pub async fn dispatch_repo_webhooks(
 async fn touch_webhook(db: &AppDatabase, webhook_id: i64, success: bool, logs: &slog::Logger) {
     use models::repos::repo_webhook::{Column as RwCol, Entity as RepoWebhookEntity};
     use models::{ColumnTrait, EntityTrait, QueryFilter};
-    use sea_orm::prelude::Expr;
+    use sea_orm::{sea_query::Expr, ExprTrait};
 
     let result: Result<sea_orm::UpdateResult, sea_orm::DbErr> = if success {
         RepoWebhookEntity::update_many()
@@ -393,18 +393,18 @@ async fn touch_webhook(db: &AppDatabase, webhook_id: i64, success: bool, logs: &
                 RwCol::LastDeliveredAt,
                 Expr::value(Some(chrono::Utc::now())),
             )
-            .col_expr(RwCol::TouchCount, Expr::value(1i64))
+            .col_expr(RwCol::TouchCount, Expr::col(RwCol::TouchCount).add(1))
             .exec(db.writer())
             .await
     } else {
         RepoWebhookEntity::update_many()
             .filter(RwCol::Id.eq(webhook_id))
-            .col_expr(RwCol::TouchCount, Expr::value(1i64))
+            .col_expr(RwCol::TouchCount, Expr::col(RwCol::TouchCount).add(1))
             .exec(db.writer())
             .await
     };
 
     if let Err(e) = result {
-        slog::warn!(logs, "{}", format!("failed to update webhook touch error={}", e));
+        slog::warn!(logs, "failed to update webhook touch error={}", e);
     }
 }

libs/git/http/mod.rs

@@ -72,16 +72,17 @@ pub async fn run_http(config: AppConfig, logger: Logger) -> anyhow::Result<()> {
     let app_cache = app_cache?;
 
     let redis_pool = app_cache.redis_pool().clone();
-    let http = Arc::new(reqwest::Client::new());
     let hook = HookService::new(
         db.clone(),
         app_cache.clone(),
         redis_pool.clone(),
         logger.clone(),
         config.clone(),
-        http,
     );
-    let sync = crate::ssh::ReceiveSyncService::new(hook);
+    let _worker_cancel = hook.start_worker();
+    slog::info!(logger, "hook worker started");
+
+    let sync = crate::ssh::ReceiveSyncService::new(redis_pool.clone(), logger.clone());
 
     let rate_limiter = Arc::new(rate_limit::RateLimiter::new(
         rate_limit::RateLimitConfig::default(),

libs/git/lib.rs

@@ -36,6 +36,9 @@ pub use diff::types::{
 };
 pub use domain::GitDomain;
 pub use error::{GitError, GitResult};
+pub use hook::pool::types::{HookTask, TaskType};
+pub use hook::pool::PoolConfig;
+pub use hook::pool::HookWorker;
 pub use hook::sync::HookMetaDataSync;
 pub use lfs::types::{LfsConfig, LfsEntry, LfsOid, LfsPointer};
 pub use merge::types::{MergeAnalysisResult, MergeOptions, MergePreferenceResult, MergeheadInfo};

libs/git/ssh/mod.rs

@@ -1,5 +1,5 @@
 use crate::error::GitError;
-use crate::hook::HookService;
+use crate::hook::pool::types::{HookTask, TaskType};
 use anyhow::Context;
 use base64::Engine;
 use config::AppConfig;
@@ -137,22 +137,12 @@ impl SSHHandle {
             "SSH server configured with methods: {:?}", config.methods
         );
         let token_service = SshTokenService::new(self.db.clone());
-        let http = Arc::new(reqwest::Client::new());
-        let hook = crate::hook::HookService::new(
-            self.db.clone(),
-            self.cache.clone(),
-            self.redis_pool.clone(),
-            self.logger.clone(),
-            self.app.clone(),
-            http,
-        );
         let mut server = server::SSHServer::new(
             self.db.clone(),
             self.cache.clone(),
             self.redis_pool.clone(),
             self.logger.clone(),
             token_service,
-            hook,
         );
 
         // Start the rate limiter cleanup background task so the HashMap
@@ -177,33 +167,63 @@ impl SSHHandle {
     }
 }
 
-/// Direct sync service — calls HookService::sync_repo inline.
-/// K8s StatefulSet HA scheduling ensures exclusive access per repo shard.
+/// Enqueues a sync task to the Redis-backed hook queue.
+/// The background worker picks it up and processes it with per-repo locking.
 #[derive(Clone)]
 pub struct ReceiveSyncService {
-    hook: HookService,
+    pool: deadpool_redis::cluster::Pool,
+    logger: Logger,
+    redis_prefix: String,
 }
 
 impl ReceiveSyncService {
-    pub fn new(hook: HookService) -> Self {
-        Self { hook }
+    pub fn new(pool: deadpool_redis::cluster::Pool, logger: Logger) -> Self {
+        Self {
+            pool,
+            logger,
+            redis_prefix: "{hook}".to_string(),
+        }
     }
 
-    /// Execute a full repo sync synchronously.
-    /// Returns Ok on success, Err on failure.
-    pub async fn send(&self, task: RepoReceiveSyncTask) -> Result<(), crate::GitError> {
-        let repo_id = task.repo_uid.to_string();
-        slog::info!(self.hook.logger, "starting sync repo_id={}", repo_id);
-        let res = self.hook.sync_repo(&repo_id).await;
-        match &res {
-            Ok(()) => {
-                slog::info!(self.hook.logger, "sync completed repo_id={}", repo_id);
-            }
+    /// Enqueue a sync task. Fire-and-forget — logs errors but does not block.
+    pub async fn send(&self, task: RepoReceiveSyncTask) {
+        let hook_task = HookTask {
+            id: uuid::Uuid::new_v4().to_string(),
+            repo_id: task.repo_uid.to_string(),
+            task_type: TaskType::Sync,
+            payload: serde_json::Value::Null,
+            created_at: chrono::Utc::now(),
+            retry_count: 0,
+        };
+
+        let task_json = match serde_json::to_string(&hook_task) {
+            Ok(j) => j,
             Err(e) => {
-                slog::error!(self.hook.logger, "sync failed repo_id={} error={}", repo_id, e);
+                error!(self.logger, "failed to serialize hook task: {}", e);
+                return;
             }
+        };
+
+        let queue_key = format!("{}:sync", self.redis_prefix);
+
+        let redis = match self.pool.get().await {
+            Ok(c) => c,
+            Err(e) => {
+                error!(self.logger, "failed to get Redis connection: {}", e);
+                return;
+            }
+        };
+
+        let mut conn: deadpool_redis::cluster::Connection = redis;
+        if let Err(e) = redis::cmd("LPUSH")
+            .arg(&queue_key)
+            .arg(&task_json)
+            .query_async::<()>(&mut conn)
+            .await
+        {
+            error!(self.logger, "failed to enqueue sync task repo_id={} error={}",
+                task.repo_uid, e);
         }
-        res
     }
 }
 
@@ -267,6 +287,18 @@ pub async fn run_ssh(config: AppConfig, logger: Logger) -> anyhow::Result<()> {
     let db = AppDatabase::init(&config).await?;
     let cache = AppCache::init(&config).await?;
     let redis_pool = cache.redis_pool().clone();
+
+    // Start the hook worker (Redis queue consumer)
+    let hook = crate::hook::HookService::new(
+        db.clone(),
+        cache.clone(),
+        redis_pool.clone(),
+        logger.clone(),
+        config.clone(),
+    );
+    let _worker_cancel = hook.start_worker();
+    slog::info!(logger, "hook worker started");
+
     SSHHandle::new(db, config.clone(), cache, redis_pool, logger)
         .run_ssh()
         .await?;

libs/git/ssh/server.rs

@@ -1,4 +1,3 @@
-use crate::hook::HookService;
 use crate::ssh::ReceiveSyncService;
 use crate::ssh::SshTokenService;
 use crate::ssh::handle::SSHandle;
@@ -16,7 +15,6 @@ pub struct SSHServer {
     pub redis_pool: RedisPool,
     pub logger: Logger,
     pub token_service: SshTokenService,
-    pub hook: HookService,
 }
 
 impl SSHServer {
@@ -26,7 +24,6 @@ impl SSHServer {
         redis_pool: RedisPool,
         logger: Logger,
         token_service: SshTokenService,
-        hook: HookService,
     ) -> Self {
         SSHServer {
             db,
@@ -34,7 +31,6 @@ impl SSHServer {
             redis_pool,
             logger,
             token_service,
-            hook,
         }
     }
 }
@@ -52,7 +48,7 @@ impl russh::server::Server for SSHServer {
         } else {
             info!(self.logger, "New SSH connection from unknown address");
         }
-        let sync_service = ReceiveSyncService::new(self.hook.clone());
+        let sync_service = ReceiveSyncService::new(self.redis_pool.clone(), self.logger.clone());
         SSHandle::new(
             self.db.clone(),
             self.cache.clone(),