feat(gitsync): add gitsync crate (sync, branches, commits, tags, worker)
zhenyi committed
Added lib/gitsync/branches.rs lib/gitsync/branches.rs +165 −0 View lib/gitsync/branches.rs
@@ -0,0 +1,165 @@
1//! Branch reconciliation.
2
3use std::collections::HashMap;
4
5use db::git::git_branches;
6use db::repo::repos;
7use git::gitcli;
8use git::refs::impls::RefsOps;
9use sea_orm::{ActiveModelTrait, ColumnTrait, EntityTrait, IntoActiveModel, QueryFilter, Set};
10use time::OffsetDateTime;
11use uuid::Uuid;
12
13use crate::error::{SyncError, SyncResult};
14
15/// Upper bound on branches enumerated per sync.
16const MAX_BRANCHES: u64 = 10_000;
17
18/// Preferred default branch names, in priority order.
19const PREFERRED_DEFAULT_BRANCHES: &[&str] = &["main", "master", "trunk"];
20
21/// A branch tip collected from the repository.
22#[derive(Debug, Clone)]
23pub(crate) struct BranchTip {
24 pub name: String,
25 pub target_sha: String,
26}
27
28/// Synchronize branches for a repository.
29///
30/// Inserts new branches, updates moved tips, soft-deletes stale branches
31/// (unless protected), and reconciles the repository default branch.
32///
33/// # Errors
34/// Returns a sync error when enumeration or the transaction fails.
35pub async fn sync_branches(
36 git: &git::GitRepository,
37 txn: &sea_orm::DatabaseTransaction,
38 repo: &repos::Model,
39) -> SyncResult<usize> {
40 let collected = collect_branch_tips(git).await?;
41
42 let existing = git_branches::Entity::find()
43 .filter(git_branches::Column::RepoId.eq(repo.id))
44 .all(txn)
45 .await
46 .map_err(SyncError::Db)?;
47
48 let mut existing_by_name: HashMap<String, git_branches::Model> = existing
49 .into_iter()
50 .filter(|b| !b.is_deleted)
51 .map(|b| (b.name.clone(), b))
52 .collect();
53
54 let mut stale = existing_by_name
55 .keys()
56 .cloned()
57 .collect::<std::collections::HashSet<_>>();
58
59 for tip in &collected {
60 stale.remove(&tip.name);
61 match existing_by_name.get(&tip.name) {
62 Some(row) if row.commit_sha != tip.target_sha => {
63 let mut am = row.clone().into_active_model();
64 am.commit_sha = Set(tip.target_sha.clone());
65 am.updated_at = Set(OffsetDateTime::now_utc());
66 am.update(txn).await.map_err(SyncError::Db)?;
67 }
68 Some(_) => {}
69 None => {
70 let now = OffsetDateTime::now_utc();
71 git_branches::ActiveModel {
72 id: Set(Uuid::new_v4()),
73 repo_id: Set(repo.id),
74 name: Set(tip.name.clone()),
75 commit_sha: Set(tip.target_sha.clone()),
76 is_deleted: Set(false),
77 is_protected: Set(false),
78 deleted_at: Set(None),
79 created_at: Set(now),
80 updated_at: Set(now),
81 }
82 .insert(txn)
83 .await
84 .map_err(SyncError::Db)?;
85 }
86 }
87 }
88
89 for name in stale {
90 if existing_by_name.get(&name).is_some_and(|r| r.is_protected) {
91 continue;
92 }
93 let mut am = existing_by_name.remove(&name).unwrap().into_active_model();
94 am.is_deleted = Set(true);
95 am.deleted_at = Set(Some(OffsetDateTime::now_utc()));
96 am.updated_at = Set(OffsetDateTime::now_utc());
97 am.update(txn).await.map_err(SyncError::Db)?;
98 }
99
100 reconcile_default_branch(git, txn, repo, &collected).await?;
101
102 Ok(collected.len())
103}
104
105/// Collect current branch tips from the bare repository.
106///
107/// # Errors
108/// Returns a sync error when enumeration fails.
109pub(crate) async fn collect_branch_tips(git: &git::GitRepository) -> SyncResult<Vec<BranchTip>> {
110 let git = git.clone();
111 tokio::task::spawn_blocking(move || {
112 let branches = git.refs_branches(0, MAX_BRANCHES).map_err(SyncError::Git)?;
113 Ok(branches
114 .into_iter()
115 .map(|b| BranchTip {
116 name: b.name,
117 target_sha: b.commit_sha.to_string(),
118 })
119 .collect::<Vec<_>>())
120 })
121 .await
122 .map_err(|e| SyncError::Internal(e.to_string()))?
123}
124
125async fn reconcile_default_branch(
126 git: &git::GitRepository,
127 txn: &sea_orm::DatabaseTransaction,
128 repo: &repos::Model,
129 collected: &[BranchTip],
130) -> SyncResult<()> {
131 if collected.is_empty() {
132 return Ok(());
133 }
134
135 let desired = if collected.iter().any(|b| b.name == repo.default_branch) {
136 repo.default_branch.clone()
137 } else {
138 PREFERRED_DEFAULT_BRANCHES
139 .iter()
140 .copied()
141 .find(|name| collected.iter().any(|b| b.name == *name))
142 .or_else(|| collected.first().map(|b| b.name.as_str()))
143 .unwrap_or(&repo.default_branch)
144 .to_string()
145 };
146
147 if desired != repo.default_branch {
148 let mut am = repo.clone().into_active_model();
149 am.default_branch = Set(desired.clone());
150 am.update(txn).await.map_err(SyncError::Db)?;
151
152 let git = git.clone();
153 tokio::task::spawn_blocking(move || {
154 gitcli::run(
155 &git.path,
156 &["symbolic-ref", "HEAD", &format!("refs/heads/{desired}")],
157 )
158 })
159 .await
160 .map_err(|e| SyncError::Internal(e.to_string()))?
161 .map_err(SyncError::Git)?;
162 }
163
164 Ok(())
165}
Added lib/gitsync/commits.rs lib/gitsync/commits.rs +207 −0 View lib/gitsync/commits.rs
@@ -0,0 +1,207 @@
1//! Commit collection and persistence.
2
3use std::collections::{HashMap, HashSet, VecDeque};
4
5use db::conn::Db;
6use db::git::git_commits;
7use db::user::user_emails;
8use git::commit::impls::CommitOps;
9use git::commit::types::Commit;
10use git::refs::impls::RefsOps;
11use sea_orm::{ColumnTrait, EntityTrait, QueryFilter, QuerySelect, Set};
12use time::OffsetDateTime;
13
14use crate::error::{SyncError, SyncResult};
15
16/// Insert batch size.
17const BATCH_SIZE: usize = 100;
18
19/// A commit pending persistence.
20#[derive(Debug, Clone)]
21pub(crate) struct CommitData {
22 pub sha: String,
23 pub author_name: String,
24 pub author_email: String,
25 pub committer_name: String,
26 pub committer_email: String,
27 pub message: String,
28 pub authored_at: OffsetDateTime,
29 pub committed_at: OffsetDateTime,
30}
31
32/// Collect new commits for a repository (blocking, on the block pool).
33///
34/// Walks each branch tip's history (breadth-first) and returns commits whose
35/// shas are not already present in `existing_shas`.
36///
37/// # Errors
38/// Returns a sync error when git operations fail.
39pub(crate) fn collect_new_commits(
40 git: &git::GitRepository,
41 existing_shas: &HashSet<String>,
42 max_total: usize,
43) -> SyncResult<Vec<CommitData>> {
44 let branches = git.refs_branches(0, 10_000).map_err(SyncError::Git)?;
45
46 let mut seen: HashSet<String> = existing_shas.clone();
47 let mut out: Vec<CommitData> = Vec::new();
48
49 for branch in branches {
50 let tip = git
51 .commit_branch_commit(&branch.name)
52 .map_err(SyncError::Git)?;
53 let mut queue = VecDeque::new();
54 queue.push_back(tip.id);
55
56 while let Some(id) = queue.pop_front() {
57 if out.len() >= max_total {
58 break;
59 }
60 let sha = id.to_string();
61 if !seen.insert(sha.clone()) {
62 continue;
63 }
64 let commit = git.commit_by_id(&id).map_err(SyncError::Git)?;
65 out.push(commit_data(&commit));
66 for parent in commit.parents {
67 queue.push_back(parent);
68 }
69 }
70 if out.len() >= max_total {
71 break;
72 }
73 }
74
75 Ok(out)
76}
77
78fn commit_data(commit: &Commit) -> CommitData {
79 let message = if commit.message.raw.is_empty() {
80 commit.message.title.clone()
81 } else {
82 commit.message.raw.clone()
83 };
84 CommitData {
85 sha: commit.id.to_string(),
86 author_name: commit.author.name.clone(),
87 author_email: commit.author.email.clone(),
88 committer_name: commit.committer.name.clone(),
89 committer_email: commit.committer.email.clone(),
90 message,
91 authored_at: OffsetDateTime::from(commit.author.when),
92 committed_at: OffsetDateTime::from(commit.committer.when),
93 }
94}
95
96/// Load existing commit shas for a repository.
97///
98/// # Errors
99/// Returns a database error.
100pub(crate) async fn existing_commit_shas(
101 db: &Db,
102 repo_id: uuid::Uuid,
103) -> SyncResult<HashSet<String>> {
104 let rows = git_commits::Entity::find()
105 .filter(git_commits::Column::RepoId.eq(repo_id))
106 .select_only()
107 .column(git_commits::Column::Sha)
108 .into_tuple::<String>()
109 .all(&db.reader())
110 .await
111 .map_err(SyncError::Db)?;
112 Ok(rows.into_iter().collect())
113}
114
115/// Collect commits not yet present in the database.
116///
117/// # Errors
118/// Returns a sync or database error.
119pub(crate) async fn collect_new_commits_for_sync(
120 git: &git::GitRepository,
121 db: &Db,
122 repo_id: uuid::Uuid,
123 max_total: usize,
124) -> SyncResult<Vec<CommitData>> {
125 let existing = existing_commit_shas(db, repo_id).await?;
126 let git = git.clone();
127 tokio::task::spawn_blocking(move || collect_new_commits(&git, &existing, max_total))
128 .await
129 .map_err(|e| SyncError::Internal(e.to_string()))?
130}
131
132/// Persist new commits, resolving authors to internal user ids.
133///
134/// Returns the number of commits inserted.
135///
136/// # Errors
137/// Returns a database error.
138pub(crate) async fn sync_commits(
139 txn: &sea_orm::DatabaseTransaction,
140 repo_id: uuid::Uuid,
141 new_commits: &[CommitData],
142) -> SyncResult<usize> {
143 if new_commits.is_empty() {
144 return Ok(0);
145 }
146
147 let author_map = resolve_user_ids(txn, new_commits).await?;
148
149 for chunk in new_commits.chunks(BATCH_SIZE) {
150 let models = chunk
151 .iter()
152 .map(|c| {
153 let lower = c.author_email.to_lowercase();
154 git_commits::ActiveModel {
155 repo_id: Set(repo_id),
156 sha: Set(c.sha.clone()),
157 message: Set(c.message.clone()),
158 author_id: Set(author_map.get(&lower).copied()),
159 author_name: Set(c.author_name.clone()),
160 author_email: Set(c.author_email.clone()),
161 authored_at: Set(c.authored_at),
162 committer_name: Set(c.committer_name.clone()),
163 committer_email: Set(c.committer_email.clone()),
164 committed_at: Set(c.committed_at),
165 created_at: Set(OffsetDateTime::now_utc()),
166 }
167 })
168 .collect::<Vec<_>>();
169 git_commits::Entity::insert_many(models)
170 .exec(txn)
171 .await
172 .map_err(SyncError::Db)?;
173 }
174
175 Ok(new_commits.len())
176}
177
178/// Resolve lowercased author emails to user ids.
179///
180/// # Errors
181/// Returns a database error.
182async fn resolve_user_ids(
183 txn: &sea_orm::DatabaseTransaction,
184 commits: &[CommitData],
185) -> SyncResult<HashMap<String, uuid::Uuid>> {
186 let mut lower_emails: HashSet<String> = HashSet::new();
187 for c in commits {
188 lower_emails.insert(c.author_email.to_lowercase());
189 lower_emails.insert(c.committer_email.to_lowercase());
190 }
191 if lower_emails.is_empty() {
192 return Ok(HashMap::new());
193 }
194
195 let rows = user_emails::Entity::find()
196 .filter(
197 user_emails::Column::LowerEmail.is_in(lower_emails.iter().cloned().collect::<Vec<_>>()),
198 )
199 .all(txn)
200 .await
201 .map_err(SyncError::Db)?;
202
203 Ok(rows
204 .into_iter()
205 .map(|r| (r.lower_email, r.user_id))
206 .collect())
207}
Added lib/gitsync/error.rs lib/gitsync/error.rs +50 −0 View lib/gitsync/error.rs
@@ -0,0 +1,50 @@
1//! Error types for the gitsync crate.
2
3use std::io;
4
5use thiserror::Error;
6
7/// Error type for repository synchronization.
8#[derive(Debug, Error)]
9pub enum SyncError {
10 /// The per-repository sync lock is held by another worker.
11 #[error("repository sync lock is held by another worker")]
12 Locked,
13
14 #[error("database error: {0}")]
15 Db(#[from] sea_orm::DbErr),
16
17 #[error("redis error: {0}")]
18 Redis(#[from] cache::redis::RedisError),
19
20 #[error("queue error: {0}")]
21 Queue(#[from] queue::QueueError),
22
23 #[error("git error: {0}")]
24 Git(#[from] git::Error),
25
26 #[error("i/o error: {0}")]
27 Io(#[from] io::Error),
28
29 #[error("operation timed out: {0}")]
30 Timeout(&'static str),
31
32 #[error("repository not found: {0}")]
33 NotFound(String),
34
35 #[error("invalid sync task: {0}")]
36 InvalidTask(String),
37
38 #[error("internal error: {0}")]
39 Internal(String),
40}
41
42pub type SyncResult<T> = Result<T, SyncError>;
43
44impl SyncError {
45 /// Whether the sync failed because the repository is locked by another worker.
46 #[must_use]
47 pub fn is_locked(&self) -> bool {
48 matches!(self, SyncError::Locked)
49 }
50}
Added lib/gitsync/event.rs lib/gitsync/event.rs +118 −0 View lib/gitsync/event.rs
@@ -0,0 +1,118 @@
1//! Post-receive event model and queue helpers.
2
3use serde::{Deserialize, Serialize};
4use uuid::Uuid;
5
6/// A single ref update performed during a push.
7#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
8pub struct PostReceiveRefUpdate {
9 /// Full ref name, e.g. `refs/heads/main`.
10 pub name: String,
11 /// Object id before the push (`0000...` for a new ref).
12 pub old_oid: String,
13 /// Object id after the push (`0000...` for a deleted ref).
14 pub new_oid: String,
15}
16
17impl PostReceiveRefUpdate {
18 /// Create a new ref update.
19 #[must_use]
20 pub fn new(
21 name: impl Into<String>,
22 old_oid: impl Into<String>,
23 new_oid: impl Into<String>,
24 ) -> Self {
25 Self {
26 name: name.into(),
27 old_oid: old_oid.into(),
28 new_oid: new_oid.into(),
29 }
30 }
31
32 /// Whether this ref was deleted.
33 #[must_use]
34 pub fn is_delete(&self) -> bool {
35 self.is_zero(self.old_oid.as_str()) && !self.is_zero(self.new_oid.as_str())
36 }
37
38 /// Whether this ref was created.
39 #[must_use]
40 pub fn is_create(&self) -> bool {
41 self.is_zero(self.new_oid.as_str())
42 }
43
44 fn is_zero(&self, oid: &str) -> bool {
45 oid.is_empty() || oid.chars().all(|c| c == '0')
46 }
47}
48
49/// A post-receive task queued for synchronization.
50#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
51pub struct PostReceiveTask {
52 /// Unique task id (used for idempotent publishing).
53 pub id: Uuid,
54 /// Repository to synchronize.
55 pub repo_id: Uuid,
56 /// Channel the repository belongs to (needed to resolve the bare path).
57 pub channel_id: Uuid,
58 /// User who performed the push, when known.
59 pub actor_id: Option<Uuid>,
60 /// Ref updates performed by the push.
61 pub refs: Vec<PostReceiveRefUpdate>,
62}
63
64impl PostReceiveTask {
65 /// Create a new post-receive task.
66 #[must_use]
67 pub fn new(
68 repo_id: Uuid,
69 channel_id: Uuid,
70 actor_id: Option<Uuid>,
71 refs: Vec<PostReceiveRefUpdate>,
72 ) -> Self {
73 Self {
74 id: Uuid::new_v4(),
75 repo_id,
76 channel_id,
77 actor_id,
78 refs,
79 }
80 }
81}
82
83/// Publish a post-receive task to the sync queue.
84///
85/// The producer key is the repository id, so all events for one repository
86/// land on the same Kafka partition and are processed in order.
87///
88/// # Errors
89/// Returns [`queue::QueueError`] when the publish fails.
90pub async fn enqueue_post_receive(
91 queue: &queue::Queue,
92 settings: &crate::SyncSettings,
93 task: &PostReceiveTask,
94) -> Result<(), queue::QueueError> {
95 let payload = serde_json::to_vec(task)?;
96 queue
97 .publish_raw(
98 &settings.queue_topic,
99 Some(task.repo_id.as_bytes()),
100 &payload,
101 )
102 .await
103}
104
105/// Publish a task to the dead-letter topic.
106///
107/// # Errors
108/// Returns [`queue::QueueError`] when the publish fails.
109pub async fn enqueue_post_receive_dlq(
110 queue: &queue::Queue,
111 settings: &crate::SyncSettings,
112 task: &PostReceiveTask,
113) -> Result<(), queue::QueueError> {
114 let payload = serde_json::to_vec(task)?;
115 queue
116 .publish_raw(&settings.dlq_topic, Some(task.repo_id.as_bytes()), &payload)
117 .await
118}
Added lib/gitsync/fsck.rs lib/gitsync/fsck.rs +41 −0 View lib/gitsync/fsck.rs
@@ -0,0 +1,41 @@
1//! Repository integrity checks (fsck) and garbage collection.
2
3use git::gc::impls::GcOps;
4
5/// Run `git gc` on the blocking pool.
6pub async fn run_gc(git: &git::GitRepository) {
7 let git = git.clone();
8 if let Err(e) = tokio::task::spawn_blocking(move || git.gc_gc())
9 .await
10 .map_err(|e| crate::SyncError::Internal(e.to_string()))
11 .and_then(|r| r.map_err(crate::SyncError::Git))
12 {
13 tracing::warn!(error = %e, "git gc failed");
14 metrics::counter!("gitsync.gc_total", "result" => "failure").increment(1);
15 } else {
16 metrics::counter!("gitsync.gc_total", "result" => "success").increment(1);
17 }
18}
19
20/// Run `git fsck` and record integrity status. Non-fatal.
21pub async fn run_fsck(git: &git::GitRepository) {
22 let git = git.clone();
23 let result = tokio::task::spawn_blocking(move || git.gc_fsck())
24 .await
25 .map_err(|e| crate::SyncError::Internal(e.to_string()))
26 .and_then(|r| r.map_err(crate::SyncError::Git));
27
28 match result {
29 Ok(fsck) if fsck.ok => {
30 metrics::counter!("gitsync.fsck_total", "result" => "ok").increment(1);
31 }
32 Ok(fsck) => {
33 tracing::error!(errors = ?fsck.errors, "fsck reported integrity errors");
34 metrics::counter!("gitsync.fsck_total", "result" => "error").increment(1);
35 }
36 Err(e) => {
37 tracing::warn!(error = %e, "fsck failed");
38 metrics::counter!("gitsync.fsck_total", "result" => "failed").increment(1);
39 }
40 }
41}
Added lib/gitsync/languages.rs lib/gitsync/languages.rs +155 −0 View lib/gitsync/languages.rs
@@ -0,0 +1,155 @@
1//! Language statistics synchronization (diff-based upsert).
2
3use std::collections::{HashMap, HashSet};
4
5use db::git::git_languages;
6use db::repo::repos;
7use git::analysis::impls::AnalysisOps;
8use sea_orm::{ActiveModelTrait, ColumnTrait, EntityTrait, IntoActiveModel, QueryFilter, Set};
9use time::OffsetDateTime;
10use uuid::Uuid;
11
12use crate::error::{SyncError, SyncResult};
13use crate::settings::SyncSettings;
14
15/// Synchronize language statistics for a repository.
16///
17/// The detection runs on the blocking pool under a timeout. Only rows whose
18/// percentage or byte size changed are updated; new languages are inserted and
19/// stale languages are removed. On any failure the operation is logged and
20/// swallowed (non-fatal).
21///
22/// # Errors
23/// Returns a database error when the transaction fails.
24pub async fn sync_repo_languages(
25 git: &git::GitRepository,
26 txn: &sea_orm::DatabaseTransaction,
27 repo: &repos::Model,
28 settings: &SyncSettings,
29) -> SyncResult<()> {
30 let desired = match detect_languages(git, settings).await {
31 Ok(Some(langs)) => langs,
32 Ok(None) => return Ok(()),
33 Err(e) => {
34 tracing::warn!(repo_id = %repo.id, error = %e, "language detection failed; skipping");
35 return Ok(());
36 }
37 };
38
39 let existing = git_languages::Entity::find()
40 .filter(git_languages::Column::RepoId.eq(repo.id))
41 .all(txn)
42 .await
43 .map_err(SyncError::Db)?;
44
45 let mut existing_by_lang: HashMap<String, git_languages::Model> = existing
46 .into_iter()
47 .map(|r| (r.language.clone(), r))
48 .collect();
49
50 let mut stale = existing_by_lang.keys().cloned().collect::<HashSet<_>>();
51
52 for lang in &desired {
53 stale.remove(&lang.language);
54 match existing_by_lang.get(&lang.language) {
55 Some(row)
56 if (row.percentage - lang.percentage).abs() > f32::EPSILON
57 || row.size != lang.size =>
58 {
59 let mut am = row.clone().into_active_model();
60 am.percentage = Set(lang.percentage);
61 am.size = Set(lang.size);
62 am.update(txn).await.map_err(SyncError::Db)?;
63 }
64 Some(_) => {}
65 None => {
66 git_languages::ActiveModel {
67 id: Set(Uuid::new_v4()),
68 repo_id: Set(repo.id),
69 language: Set(lang.language.clone()),
70 percentage: Set(lang.percentage),
71 size: Set(lang.size),
72 created_at: Set(OffsetDateTime::now_utc()),
73 }
74 .insert(txn)
75 .await
76 .map_err(SyncError::Db)?;
77 }
78 }
79 }
80
81 for lang in stale {
82 let row = existing_by_lang.remove(&lang).unwrap();
83 git_languages::Entity::delete_by_id(row.id)
84 .exec(txn)
85 .await
86 .map_err(SyncError::Db)?;
87 }
88
89 if let Some(primary) = desired.first()
90 && repo.primary_language.as_deref() != Some(primary.language.as_str())
91 {
92 let mut am = repo.clone().into_active_model();
93 am.primary_language = Set(Some(primary.language.clone()));
94 am.update(txn).await.map_err(SyncError::Db)?;
95 }
96
97 Ok(())
98}
99
100/// A normalized language entry.
101#[derive(Debug, Clone)]
102struct LanguageEntry {
103 language: String,
104 percentage: f32,
105 size: i64,
106}
107
108/// Detect languages under a timeout, returning `None` on timeout.
109///
110/// # Errors
111/// Returns a sync error when detection fails.
112async fn detect_languages(
113 git: &git::GitRepository,
114 settings: &SyncSettings,
115) -> SyncResult<Option<Vec<LanguageEntry>>> {
116 let git = git.clone();
117 let stats = tokio::time::timeout(settings.language_timeout(), async move {
118 tokio::task::spawn_blocking(move || {
119 let stats = git.analysis_languages().map_err(SyncError::Git)?;
120 Ok::<_, SyncError>(stats)
121 })
122 .await
123 .map_err(|e| SyncError::Internal(e.to_string()))?
124 })
125 .await;
126
127 let stats = match stats {
128 Ok(Ok(s)) => s,
129 Ok(Err(e)) => return Err(e),
130 Err(_) => {
131 tracing::warn!("language detection timed out");
132 return Ok(None);
133 }
134 };
135
136 let mut entries: Vec<LanguageEntry> = stats
137 .languages
138 .into_iter()
139 .filter(|l| l.size > 0 && !l.language.is_empty())
140 .map(|l| LanguageEntry {
141 language: l.language,
142 percentage: l.percentage.clamp(0.0, 100.0),
143 size: l.size,
144 })
145 .collect();
146
147 entries.sort_by(|a, b| {
148 b.size
149 .cmp(&a.size)
150 .then_with(|| a.language.cmp(&b.language))
151 });
152 entries.truncate(settings.max_languages);
153
154 Ok(Some(entries))
155}
Added lib/gitsync/lib.rs lib/gitsync/lib.rs +28 −0 View lib/gitsync/lib.rs
@@ -0,0 +1,28 @@
1//! gitsync — post-receive repository synchronization.
2//!
3//! Consumes post-receive sync tasks from the Kafka queue ([`queue`](queue))
4//! and reconciles a repository's refs, commits, tags, languages and statistics
5//! into the database. A per-repository Redis distributed lock guarantees that
6//! only one worker performs a sync at a time.
7
8pub mod branches;
9pub mod commits;
10pub mod error;
11pub mod event;
12pub mod fsck;
13pub mod languages;
14pub mod lock;
15pub mod settings;
16pub mod stats;
17pub mod sync;
18pub mod tags;
19pub mod webhook;
20pub mod worker;
21
22pub use error::{SyncError, SyncResult};
23pub use event::{PostReceiveRefUpdate, PostReceiveTask, enqueue_post_receive};
24pub use lock::{SyncLock, acquire_lock};
25pub use settings::SyncSettings;
26pub use sync::RepoSync;
27pub use webhook::{NoopWebhookTx, WebhookTx};
28pub use worker::{RunningWorker, SyncWorker};
Added lib/gitsync/lock.rs lib/gitsync/lock.rs +92 −0 View lib/gitsync/lock.rs
@@ -0,0 +1,92 @@
1//! Per-repository distributed sync lock backed by Redis.
2
3use std::process;
4use std::sync::Arc;
5
6use cache::RedisCache;
7use uuid::Uuid;
8
9use crate::{SyncError, SyncResult};
10
11/// Redis key prefix for sync locks.
12const LOCK_KEY_PREFIX: &str = "git:sync:lock:";
13
14/// A held distributed lock. Release explicitly via [`SyncLock::release`].
15pub struct SyncLock {
16 kv: Arc<RedisCache>,
17 key: String,
18 value: String,
19}
20
21impl SyncLock {
22 /// The token identifying this lock holder.
23 #[must_use]
24 pub fn value(&self) -> &str {
25 &self.value
26 }
27
28 /// The Redis key under which this lock lives.
29 #[must_use]
30 pub fn key(&self) -> &str {
31 &self.key
32 }
33
34 /// Atomically extend the lock TTL while the token still matches.
35 ///
36 /// Returns `false` when the lock was lost (the value no longer matches).
37 ///
38 /// # Errors
39 /// Returns [`SyncError::Redis`] when the Redis command fails.
40 pub async fn refresh(&self, ttl_secs: u64) -> SyncResult<bool> {
41 self.kv
42 .refresh_lock(&self.key, self.value.as_bytes(), ttl_secs)
43 .await
44 .map_err(SyncError::Redis)
45 }
46
47 /// Release the lock. No-op when the lock was already lost.
48 ///
49 /// # Errors
50 /// Returns [`SyncError::Redis`] when the Redis command fails.
51 pub async fn release(&self) -> SyncResult<()> {
52 self.kv
53 .compare_and_delete(&self.key, self.value.as_bytes())
54 .await
55 .map_err(SyncError::Redis)?;
56 Ok(())
57 }
58}
59
60fn lock_key(repo_id: Uuid) -> String {
61 format!("{LOCK_KEY_PREFIX}{repo_id}")
62}
63
64/// Acquire the sync lock for a repository.
65///
66/// Returns `Ok(None)` when the lock is already held by another worker.
67///
68/// # Errors
69/// Returns [`SyncError::Redis`] when the Redis command fails.
70pub async fn acquire_lock(
71 kv: &Arc<RedisCache>,
72 repo_id: Uuid,
73 ttl_secs: u64,
74) -> SyncResult<Option<SyncLock>> {
75 let key = lock_key(repo_id);
76 let value = format!("{}:{}", Uuid::new_v4(), process::id());
77
78 let acquired = kv
79 .set_nx(&key, value.as_bytes().to_vec(), ttl_secs)
80 .await
81 .map_err(SyncError::Redis)?;
82
83 if !acquired {
84 return Ok(None);
85 }
86
87 Ok(Some(SyncLock {
88 kv: kv.clone(),
89 key,
90 value,
91 }))
92}
Added lib/gitsync/settings.rs lib/gitsync/settings.rs +56 −0 View lib/gitsync/settings.rs
@@ -0,0 +1,56 @@
1//! Configuration for the sync worker and per-repository sync.
2
3use std::time::Duration;
4
5/// Settings controlling synchronization behaviour.
6#[derive(Debug, Clone)]
7pub struct SyncSettings {
8 /// Kafka topic post-receive tasks are published to.
9 pub queue_topic: String,
10 /// Kafka topic for tasks that exceed the retry limit.
11 pub dlq_topic: String,
12 /// Consumer group id shared by all worker instances.
13 pub group_id: String,
14 /// Number of concurrent consumer loops inside one worker.
15 pub concurrency: usize,
16 /// TTL in seconds for the per-repository sync lock.
17 pub lock_ttl_secs: u64,
18 /// Upper bound of commits collected per sync.
19 pub max_commits_per_sync: usize,
20 /// Number of times a lock-contended task is re-published before going to the DLQ.
21 pub max_lock_retries: u32,
22 /// Timeout for language detection.
23 pub language_timeout_secs: u64,
24 /// Maximum number of languages persisted per repository.
25 pub max_languages: usize,
26}
27
28impl Default for SyncSettings {
29 fn default() -> Self {
30 Self {
31 queue_topic: "gta.git.sync".to_string(),
32 dlq_topic: "gta.git.sync.dlq".to_string(),
33 group_id: "git-sync-workers".to_string(),
34 concurrency: 4,
35 lock_ttl_secs: 300,
36 max_commits_per_sync: 10_000,
37 max_lock_retries: 5,
38 language_timeout_secs: 30,
39 max_languages: 50,
40 }
41 }
42}
43
44impl SyncSettings {
45 /// Lock TTL as a [`Duration`].
46 #[must_use]
47 pub fn lock_ttl(&self) -> Duration {
48 Duration::from_secs(self.lock_ttl_secs)
49 }
50
51 /// Language detection timeout as a [`Duration`].
52 #[must_use]
53 pub fn language_timeout(&self) -> Duration {
54 Duration::from_secs(self.language_timeout_secs)
55 }
56}
Added lib/gitsync/stats.rs lib/gitsync/stats.rs +34 −0 View lib/gitsync/stats.rs
@@ -0,0 +1,34 @@
1//! Repository statistics updates.
2
3use db::repo::repos;
4use git::gc::impls::GcOps;
5use sea_orm::{ActiveModelTrait, IntoActiveModel, Set};
6
7use crate::error::{SyncError, SyncResult};
8
9/// Update aggregate counters and git size on the repository row.
10///
11/// # Errors
12/// Returns a sync error when either the git stats or the transaction fails.
13pub async fn sync_repo_stats(
14 git: &git::GitRepository,
15 txn: &sea_orm::DatabaseTransaction,
16 repo: &repos::Model,
17 new_commits: usize,
18 branch_count: usize,
19 tag_count: usize,
20) -> SyncResult<()> {
21 let git = git.clone();
22 let stats = tokio::task::spawn_blocking(move || git.gc_stats().map_err(SyncError::Git))
23 .await
24 .map_err(|e| SyncError::Internal(e.to_string()))??;
25
26 let mut am = repo.clone().into_active_model();
27 am.git_size = Set(stats.git_size as i64);
28 am.nums_commits = Set(repo.nums_commits + new_commits as i32);
29 am.nums_branches = Set(branch_count as i32);
30 am.nums_tags = Set(tag_count as i32);
31 am.update(txn).await.map_err(SyncError::Db)?;
32
33 Ok(())
34}
Added lib/gitsync/sync.rs lib/gitsync/sync.rs +179 −0 View lib/gitsync/sync.rs
@@ -0,0 +1,179 @@
1//! Core per-repository synchronization.
2
3use std::sync::Arc;
4
5use cache::{Cache, RedisCache};
6use db::conn::Db;
7use db::repo::repos;
8use git::{GitRepository, gitcli};
9use repo::config::SharedRepoConfig;
10use repo::path::git_bare_path;
11use sea_orm::{DatabaseTransaction, TransactionTrait};
12use storage::StorageRepository;
13use tokio::sync::watch;
14
15use crate::commits::{self, CommitData};
16use crate::error::{SyncError, SyncResult};
17use crate::lock;
18use crate::settings::SyncSettings;
19use crate::{branches, languages, stats, tags};
20
21/// Runnable repository synchronizer.
22///
23/// Build via [`RepoSync::new`]. All git operations are blocking and therefore
24/// executed on the blocking thread pool.
25#[derive(Clone)]
26pub struct RepoSync {
27 db: Db,
28 repo: repos::Model,
29 git: GitRepository,
30 kv: Arc<RedisCache>,
31 settings: SyncSettings,
32}
33
34impl RepoSync {
35 /// Create a synchronizer for a single repository.
36 #[must_use]
37 pub fn new(
38 db: Db,
39 config: SharedRepoConfig,
40 repo: repos::Model,
41 kv: Arc<RedisCache>,
42 cache: Arc<Cache>,
43 storage: Arc<dyn StorageRepository>,
44 settings: SyncSettings,
45 ) -> Self {
46 let path = git_bare_path(&config, repo.channel_id, repo.id);
47 let object_cache = git::object::cache::ObjectCache {
48 prefix: format!("sync:{}", repo.id),
49 cache: cache.clone(),
50 expire_ttl: 300,
51 };
52 let git = GitRepository::new(path, object_cache, storage.clone());
53 Self {
54 db,
55 repo,
56 git,
57 kv,
58 settings,
59 }
60 }
61
62 /// The repository id being synchronized.
63 #[must_use]
64 pub fn repo_id(&self) -> uuid::Uuid {
65 self.repo.id
66 }
67
68 /// Run a full synchronization, holding the per-repository lock.
69 ///
70 /// Returns [`SyncError::Locked`] when the lock is held elsewhere or is
71 /// lost mid-run (in which case the lock is not released).
72 ///
73 /// # Errors
74 /// Returns any sync or database error encountered.
75 pub async fn sync(&self) -> SyncResult<()> {
76 let ttl = self.settings.lock_ttl_secs;
77 let Some(lock) = lock::acquire_lock(&self.kv, self.repo.id, ttl).await? else {
78 return Err(SyncError::Locked);
79 };
80 let lock = Arc::new(lock);
81
82 let (tx, mut rx) = watch::channel(());
83 let heartbeat_lock = lock.clone();
84 let heartbeat = tokio::spawn(async move {
85 let interval = (ttl / 3).max(1);
86 loop {
87 tokio::time::sleep(std::time::Duration::from_secs(interval)).await;
88 match heartbeat_lock.refresh(ttl).await {
89 Ok(true) => {}
90 Ok(false) | Err(_) => {
91 let _ = tx.send(());
92 break;
93 }
94 }
95 }
96 });
97
98 let result = tokio::select! {
99 r = self.sync_work() => r,
100 _ = rx.changed() => Err(SyncError::Locked),
101 };
102
103 heartbeat.abort();
104
105 let locked = result.as_ref().err().is_some_and(SyncError::is_locked);
106 if !locked {
107 lock.release().await?;
108 }
109
110 result
111 }
112
113 /// Execute the synchronization work without acquiring the lock.
114 ///
115 /// # Errors
116 /// Returns any sync or database error encountered.
117 pub async fn sync_work(&self) -> SyncResult<()> {
118 let new_commits = commits::collect_new_commits_for_sync(
119 &self.git,
120 &self.db,
121 self.repo.id,
122 self.settings.max_commits_per_sync,
123 )
124 .await?;
125
126 let txn = self.db.writer().begin().await.map_err(SyncError::Db)?;
127
128 let (branch_count, synced_commits, tag_count) =
129 self.sync_refs_and_data(&txn, &new_commits).await?;
130
131 metrics::counter!("gitsync.sync_detail", "kind" => "branches")
132 .increment(branch_count as u64);
133 metrics::counter!("gitsync.sync_detail", "kind" => "commits")
134 .increment(synced_commits as u64);
135 metrics::counter!("gitsync.sync_detail", "kind" => "tags").increment(tag_count as u64);
136
137 languages::sync_repo_languages(&self.git, &txn, &self.repo, &self.settings).await?;
138
139 txn.commit().await.map_err(SyncError::Db)?;
140
141 crate::fsck::run_fsck(&self.git).await;
142 crate::fsck::run_gc(&self.git).await;
143
144 Ok(())
145 }
146
147 async fn sync_refs_and_data(
148 &self,
149 txn: &DatabaseTransaction,
150 new_commits: &[CommitData],
151 ) -> SyncResult<(usize, usize, usize)> {
152 let branch_count = branches::sync_branches(&self.git, txn, &self.repo).await?;
153 let synced_commits = commits::sync_commits(txn, self.repo.id, new_commits).await?;
154 let tag_count = tags::sync_tags(&self.git, txn, &self.repo).await?;
155 stats::sync_repo_stats(
156 &self.git,
157 txn,
158 &self.repo,
159 new_commits.len(),
160 branch_count,
161 tag_count,
162 )
163 .await?;
164 Ok((branch_count, synced_commits, tag_count))
165 }
166
167 /// Set the repository's default branch on the bare repository (`git symbolic-ref HEAD`).
168 ///
169 /// # Errors
170 /// Returns a git error when the command fails.
171 pub fn set_default_branch(&self, name: &str) -> SyncResult<()> {
172 gitcli::run(
173 &self.git.path,
174 &["symbolic-ref", "HEAD", &format!("refs/heads/{name}")],
175 )
176 .map_err(SyncError::Git)?;
177 Ok(())
178 }
179}
Added lib/gitsync/tags.rs lib/gitsync/tags.rs +133 −0 View lib/gitsync/tags.rs
@@ -0,0 +1,133 @@
1//! Tag reconciliation.
2
3use std::collections::HashMap;
4
5use db::git::git_tags;
6use db::git::git_tags::TagType;
7use db::repo::repos;
8use git::tag::impls::TagOps;
9use sea_orm::{ActiveModelTrait, ColumnTrait, EntityTrait, IntoActiveModel, QueryFilter, Set};
10use time::OffsetDateTime;
11use uuid::Uuid;
12
13use crate::error::{SyncError, SyncResult};
14
15/// Upper bound on tags enumerated per sync.
16const MAX_TAGS: u64 = 10_000;
17
18/// Synchronize tags for a repository.
19///
20/// Inserts new tags, updates moved/changed tags, and physically deletes stale
21/// tags.
22///
23/// # Errors
24/// Returns a sync error when enumeration or the transaction fails.
25pub async fn sync_tags(
26 git: &git::GitRepository,
27 txn: &sea_orm::DatabaseTransaction,
28 repo: &repos::Model,
29) -> SyncResult<usize> {
30 let collected = collect_tag_tips(git).await?;
31
32 let existing = git_tags::Entity::find()
33 .filter(git_tags::Column::RepoId.eq(repo.id))
34 .all(txn)
35 .await
36 .map_err(SyncError::Db)?;
37
38 let mut existing_by_name: HashMap<String, git_tags::Model> =
39 existing.into_iter().map(|t| (t.name.clone(), t)).collect();
40
41 let mut stale = existing_by_name
42 .keys()
43 .cloned()
44 .collect::<std::collections::HashSet<_>>();
45
46 for tip in &collected {
47 stale.remove(&tip.name);
48 match existing_by_name.get(&tip.name) {
49 Some(row)
50 if row.commit_sha != tip.target_sha
51 || row.message != tip.message
52 || row.tag_type != tip.tag_type =>
53 {
54 let mut am = row.clone().into_active_model();
55 am.commit_sha = Set(tip.target_sha.clone());
56 am.tag_type = Set(tip.tag_type);
57 am.message = Set(tip.message.clone());
58 am.tagger_name = Set(tip.tagger_name.clone());
59 am.tagger_email = Set(tip.tagger_email.clone());
60 am.update(txn).await.map_err(SyncError::Db)?;
61 }
62 Some(_) => {}
63 None => {
64 git_tags::ActiveModel {
65 id: Set(Uuid::new_v4()),
66 repo_id: Set(repo.id),
67 name: Set(tip.name.clone()),
68 tag_type: Set(tip.tag_type),
69 commit_sha: Set(tip.target_sha.clone()),
70 message: Set(tip.message.clone()),
71 tagger_name: Set(tip.tagger_name.clone()),
72 tagger_email: Set(tip.tagger_email.clone()),
73 created_at: Set(OffsetDateTime::now_utc()),
74 }
75 .insert(txn)
76 .await
77 .map_err(SyncError::Db)?;
78 }
79 }
80 }
81
82 for name in stale {
83 let row = existing_by_name.remove(&name).unwrap();
84 git_tags::Entity::delete_by_id(row.id)
85 .exec(txn)
86 .await
87 .map_err(SyncError::Db)?;
88 }
89
90 Ok(collected.len())
91}
92
93/// A tag tip collected from the repository.
94#[derive(Debug, Clone)]
95pub(crate) struct TagTip {
96 pub name: String,
97 pub target_sha: String,
98 pub tag_type: TagType,
99 pub message: Option<String>,
100 pub tagger_name: Option<String>,
101 pub tagger_email: Option<String>,
102}
103
104/// Collect current tag tips from the bare repository.
105///
106/// # Errors
107/// Returns a sync error when enumeration fails.
108pub(crate) async fn collect_tag_tips(git: &git::GitRepository) -> SyncResult<Vec<TagTip>> {
109 let git = git.clone();
110 tokio::task::spawn_blocking(move || {
111 let tags = git.tag_list(1, MAX_TAGS).map_err(SyncError::Git)?;
112 Ok(tags
113 .tags
114 .into_iter()
115 .map(|t| {
116 let tag_type = match t.tag_type {
117 git::tag::types::TagType::Annotated => TagType::Annotated,
118 git::tag::types::TagType::Lightweight => TagType::Lightweight,
119 };
120 TagTip {
121 name: t.name,
122 target_sha: t.object.to_string(),
123 tag_type,
124 message: t.message,
125 tagger_name: t.tagger.as_ref().map(|s| s.name.clone()),
126 tagger_email: t.tagger.as_ref().map(|s| s.email.clone()),
127 }
128 })
129 .collect::<Vec<_>>())
130 })
131 .await
132 .map_err(|e| SyncError::Internal(e.to_string()))?
133}
Added lib/gitsync/webhook.rs lib/gitsync/webhook.rs +28 −0 View lib/gitsync/webhook.rs
@@ -0,0 +1,28 @@
1//! Webhook dispatch entry point.
2//!
3//! The concrete webhook implementation is deferred; the trait is invoked after
4//! every successful sync so a future dispatcher can be plugged in without
5//! changing the worker.
6
7use std::error::Error;
8
9use crate::event::PostReceiveTask;
10
11/// Dispatches webhook notifications for a completed sync.
12#[async_trait::async_trait]
13pub trait WebhookTx: Send + Sync {
14 /// Dispatch webhooks for the given task.
15 async fn dispatch(&self, task: &PostReceiveTask) -> Result<(), Box<dyn Error + Send + Sync>>;
16}
17
18/// Placeholder dispatcher that only logs.
19#[derive(Debug, Default, Clone)]
20pub struct NoopWebhookTx;
21
22#[async_trait::async_trait]
23impl WebhookTx for NoopWebhookTx {
24 async fn dispatch(&self, task: &PostReceiveTask) -> Result<(), Box<dyn Error + Send + Sync>> {
25 tracing::debug!(repo_id = %task.repo_id, refs = task.refs.len(), "webhook dispatch placeholder (not implemented)");
26 Ok(())
27 }
28}
Added lib/gitsync/worker.rs lib/gitsync/worker.rs +255 −0 View lib/gitsync/worker.rs
@@ -0,0 +1,255 @@
1//! Sync worker: consumes post-receive tasks from Kafka and syncs repos.
2
3use std::sync::Arc;
4
5use cache::{Cache, RedisCache};
6use db::conn::Db;
7use db::repo::repos;
8use queue::{AutoOffsetReset, Queue, QueueError, RawMessage, SubscriptionMode};
9use repo::config::SharedRepoConfig;
10use sea_orm::EntityTrait;
11use storage::StorageRepository;
12
13use crate::event::PostReceiveTask;
14use crate::settings::SyncSettings;
15use crate::sync::RepoSync;
16use crate::webhook::WebhookTx;
17use crate::{SyncError, SyncResult};
18
19/// Attempts-tracking key prefix for lock-contended tasks.
20const ATTEMPT_KEY_PREFIX: &str = "git:sync:attempts:";
21/// Retention for the attempts counter.
22const ATTEMPT_TTL_SECS: u64 = 3600;
23
24/// Worker that processes post-receive sync tasks.
25#[derive(Clone)]
26pub struct SyncWorker {
27 db: Db,
28 config: SharedRepoConfig,
29 kv: Arc<RedisCache>,
30 cache: Arc<Cache>,
31 storage: Arc<dyn StorageRepository>,
32 queue: Option<Arc<Queue>>,
33 webhook: Arc<dyn WebhookTx>,
34 settings: SyncSettings,
35 concurrency: usize,
36}
37
38/// Handle to a running worker; call [`shutdown`](RunningWorker::shutdown) to stop.
39pub struct RunningWorker {
40 handles: Vec<queue::SubscribeHandle>,
41}
42
43impl RunningWorker {
44 /// Gracefully shut down all consumer loops.
45 ///
46 /// # Errors
47 /// Returns a queue error when shutdown fails.
48 pub async fn shutdown(self) -> SyncResult<()> {
49 for handle in self.handles {
50 handle.shutdown().await.map_err(SyncError::Queue)?;
51 }
52 Ok(())
53 }
54}
55
56impl SyncWorker {
57 /// Create a new worker.
58 #[must_use]
59 #[allow(clippy::too_many_arguments)]
60 pub fn new(
61 db: Db,
62 config: SharedRepoConfig,
63 kv: Arc<RedisCache>,
64 cache: Arc<Cache>,
65 storage: Arc<dyn StorageRepository>,
66 queue: Option<Arc<Queue>>,
67 webhook: Arc<dyn WebhookTx>,
68 settings: SyncSettings,
69 ) -> Self {
70 let concurrency = settings.concurrency.max(1);
71 Self {
72 db,
73 config,
74 kv,
75 cache,
76 storage,
77 queue,
78 webhook,
79 settings,
80 concurrency,
81 }
82 }
83
84 /// Override the number of concurrent consumer loops.
85 #[must_use]
86 pub fn with_concurrency(mut self, concurrency: usize) -> Self {
87 self.concurrency = concurrency.max(1);
88 self
89 }
90
91 /// Start consuming the sync queue.
92 ///
93 /// Returns a [`RunningWorker`] that must be kept alive and shut down
94 /// explicitly. Spawns `concurrency` consumer loops sharing the same Kafka
95 /// consumer group, so partitions are rebalanced across them.
96 ///
97 /// # Errors
98 /// Returns a queue error when a subscription cannot be established.
99 pub async fn run(self) -> SyncResult<RunningWorker> {
100 let Some(queue) = self.queue else {
101 tracing::warn!("sync queue disabled; gitsync worker not started");
102 return Ok(RunningWorker {
103 handles: Vec::new(),
104 });
105 };
106
107 let mode = SubscriptionMode::Group {
108 group_id: self.settings.group_id.clone(),
109 auto_offset_reset: AutoOffsetReset::Earliest,
110 };
111 let topics = [self.settings.queue_topic.as_str()];
112
113 let mut handles = Vec::with_capacity(self.concurrency);
114 for _ in 0..self.concurrency {
115 let queue = queue.clone();
116 let db = self.db.clone();
117 let config = self.config.clone();
118 let kv = self.kv.clone();
119 let cache = self.cache.clone();
120 let storage = self.storage.clone();
121 let webhook = self.webhook.clone();
122 let settings = self.settings.clone();
123
124 let handle_queue = queue.clone();
125 let handle = queue
126 .subscribe_raw(mode.clone(), &topics, move |msg| {
127 let queue = handle_queue.clone();
128 let db = db.clone();
129 let config = config.clone();
130 let kv = kv.clone();
131 let cache = cache.clone();
132 let storage = storage.clone();
133 let webhook = webhook.clone();
134 let settings = settings.clone();
135 Box::pin(async move {
136 on_message(
137 &queue, &db, &config, &kv, &cache, storage, webhook, &settings, &msg,
138 )
139 .await
140 })
141 })
142 .map_err(SyncError::Queue)?;
143 handles.push(handle);
144 }
145
146 tracing::info!(concurrency = self.concurrency, topic = %self.settings.queue_topic, "gitsync worker started");
147 Ok(RunningWorker { handles })
148 }
149}
150
151#[allow(clippy::too_many_arguments)]
152async fn on_message(
153 queue: &Queue,
154 db: &Db,
155 config: &SharedRepoConfig,
156 kv: &Arc<RedisCache>,
157 cache: &Arc<Cache>,
158 storage: Arc<dyn StorageRepository>,
159 webhook: Arc<dyn WebhookTx>,
160 settings: &SyncSettings,
161 msg: &RawMessage,
162) -> Result<(), QueueError> {
163 let task: PostReceiveTask = match serde_json::from_slice(&msg.payload) {
164 Ok(t) => t,
165 Err(e) => {
166 tracing::error!(partition = msg.partition, offset = msg.offset, error = %e, "invalid sync task payload");
167 metrics::counter!("gitsync.tasks_total", "result" => "invalid").increment(1);
168 return Ok(());
169 }
170 };
171
172 let Some(repo) = repos::Entity::find_by_id(task.repo_id)
173 .one(&db.reader())
174 .await
175 .map_err(|e| QueueError::Handler(e.to_string()))?
176 else {
177 tracing::warn!(repo_id = %task.repo_id, "repository not found; skipping sync task");
178 return Ok(());
179 };
180
181 let rsync = RepoSync::new(
182 db.clone(),
183 config.clone(),
184 repo,
185 kv.clone(),
186 cache.clone(),
187 storage,
188 settings.clone(),
189 );
190
191 match rsync.sync().await {
192 Ok(()) => {
193 metrics::counter!("gitsync.tasks_total", "result" => "success").increment(1);
194 if let Err(e) = webhook.dispatch(&task).await {
195 tracing::warn!(repo_id = %task.repo_id, error = %e, "webhook dispatch failed");
196 }
197 Ok(())
198 }
199 Err(e) if e.is_locked() => {
200 metrics::counter!("gitsync.tasks_total", "result" => "locked").increment(1);
201 requeue_on_lock(queue, kv, settings, &task).await;
202 Ok(())
203 }
204 Err(e) => {
205 tracing::error!(repo_id = %task.repo_id, error = %e, "sync task failed");
206 metrics::counter!("gitsync.tasks_total", "result" => "failure").increment(1);
207 Err(queue_err_str(&e))
208 }
209 }
210}
211
212/// Re-publish a lock-contended task, routing to the DLQ after too many attempts.
213async fn requeue_on_lock(
214 queue: &Queue,
215 kv: &Arc<RedisCache>,
216 settings: &SyncSettings,
217 task: &PostReceiveTask,
218) {
219 let key = format!("{ATTEMPT_KEY_PREFIX}{}", task.id);
220 let attempts = match kv.incr_with_ttl(&key, ATTEMPT_TTL_SECS).await {
221 Ok(n) => n,
222 Err(e) => {
223 tracing::warn!(error = %e, "failed to record lock retry attempt");
224 1
225 }
226 };
227
228 if attempts > u64::from(settings.max_lock_retries) {
229 let _ = kv.delete(&key).await;
230 match crate::event::enqueue_post_receive_dlq(queue, settings, task).await {
231 Ok(()) => {
232 metrics::counter!("gitsync.tasks_total", "result" => "dlq").increment(1);
233 tracing::warn!(repo_id = %task.repo_id, attempts, "task exceeded lock retry limit; sent to DLQ");
234 }
235 Err(e) => {
236 tracing::error!(repo_id = %task.repo_id, error = %e, "failed to route task to DLQ")
237 }
238 }
239 return;
240 }
241
242 tokio::time::sleep(std::time::Duration::from_secs(5)).await;
243 match crate::event::enqueue_post_receive(queue, settings, task).await {
244 Ok(()) => {
245 tracing::debug!(repo_id = %task.repo_id, attempts, "re-queued lock-contended sync task")
246 }
247 Err(e) => {
248 tracing::error!(repo_id = %task.repo_id, error = %e, "failed to re-queue sync task")
249 }
250 }
251}
252
253fn queue_err_str(e: &SyncError) -> QueueError {
254 QueueError::Handler(e.to_string())
255}