refactor(gitsync): refresh crate across sync, branches, commits, tags, worker

Modified lib/gitsync/branches.rs +0 −16 View lib/gitsync/branches.rs

@@ -1,5 +1,3 @@

1//! Branch reconciliation.

2

31use std::collections::HashMap;

42

53use db::git::git_branches;

@@ -12,26 +10,16 @@

1210

1311use crate::error::{SyncError, SyncResult};

1412

15/// Upper bound on branches enumerated per sync.

1613const MAX_BRANCHES: u64 = 10_000;

1714

18/// Preferred default branch names, in priority order.

1915const PREFERRED_DEFAULT_BRANCHES: &[&str] = &["main", "master", "trunk"];

2016

21/// A branch tip collected from the repository.

2217#[derive(Debug, Clone)]

2318pub(crate) struct BranchTip {

2419 pub name: String,

2520 pub target_sha: String,

2621}

2722

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.

3523pub async fn sync_branches(

3624 git: &git::GitRepository,

3725 txn: &sea_orm::DatabaseTransaction,

@@ -102,10 +90,6 @@

10290 Ok(collected.len())

10391}

10492

105/// Collect current branch tips from the bare repository.

106///

107/// # Errors

108/// Returns a sync error when enumeration fails.

10993pub(crate) async fn collect_branch_tips(git: &git::GitRepository) -> SyncResult<Vec<BranchTip>> {

11094 let git = git.clone();

11195 tokio::task::spawn_blocking(move || {

Modified lib/gitsync/commits.rs +0 −29 View lib/gitsync/commits.rs

@@ -1,5 +1,3 @@

1//! Commit collection and persistence.

2

31use std::collections::{HashMap, HashSet, VecDeque};

42

53use db::conn::Db;

@@ -13,10 +11,8 @@

1311

1412use crate::error::{SyncError, SyncResult};

1513

16/// Insert batch size.

1714const BATCH_SIZE: usize = 100;

1815

19/// A commit pending persistence.

2016#[derive(Debug, Clone)]

2117pub(crate) struct CommitData {

2218 pub sha: String,

@@ -29,13 +25,6 @@

2925 pub committed_at: OffsetDateTime,

3026}

3127

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.

3928pub(crate) fn collect_new_commits(

4029 git: &git::GitRepository,

4130 existing_shas: &HashSet<String>,

@@ -93,10 +82,6 @@

9382 }

9483}

9584

96/// Load existing commit shas for a repository.

97///

98/// # Errors

99/// Returns a database error.

10085pub(crate) async fn existing_commit_shas(

10186 db: &Db,

10287 repo_id: uuid::Uuid,

@@ -112,10 +97,6 @@

11297 Ok(rows.into_iter().collect())

11398}

11499

115/// Collect commits not yet present in the database.

116///

117/// # Errors

118/// Returns a sync or database error.

119100pub(crate) async fn collect_new_commits_for_sync(

120101 git: &git::GitRepository,

121102 db: &Db,

@@ -129,12 +110,6 @@

129110 .map_err(|e| SyncError::Internal(e.to_string()))?

130111}

131112

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.

138113pub(crate) async fn sync_commits(

139114 txn: &sea_orm::DatabaseTransaction,

140115 repo_id: uuid::Uuid,

@@ -175,10 +150,6 @@

175150 Ok(new_commits.len())

176151}

177152

178/// Resolve lowercased author emails to user ids.

179///

180/// # Errors

181/// Returns a database error.

182153async fn resolve_user_ids(

183154 txn: &sea_orm::DatabaseTransaction,

184155 commits: &[CommitData],

Modified lib/gitsync/error.rs +2 −7 View lib/gitsync/error.rs

@@ -1,14 +1,10 @@

1//! Error types for the gitsync crate.

2

31use std::io;

42

53use thiserror::Error;

64

7/// Error type for repository synchronization.

85#[derive(Debug, Error)]

96pub enum SyncError {

10 /// The per-repository sync lock is held by another worker.

11 #[error("repository sync lock is held by another worker")]

7 #[error("repository sync lock is held by another worker")]

128 Locked,

139

1410 #[error("database error: {0}")]

@@ -42,8 +38,7 @@

4238pub type SyncResult<T> = Result<T, SyncError>;

4339

4440impl SyncError {

45 /// Whether the sync failed because the repository is locked by another worker.

46 #[must_use]

41 #[must_use]

4742 pub fn is_locked(&self) -> bool {

4843 matches!(self, SyncError::Locked)

4944 }

Modified lib/gitsync/event.rs +44 −41 View lib/gitsync/event.rs

@@ -1,22 +1,15 @@

1//! Post-receive event model and queue helpers.

2

31use serde::{Deserialize, Serialize};

42use uuid::Uuid;

53

6/// A single ref update performed during a push.

74#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]

85pub 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,

6 pub name: String,

7 pub old_oid: String,

8 pub new_oid: String,

159}

1610

1711impl PostReceiveRefUpdate {

18 /// Create a new ref update.

19 #[must_use]

12 #[must_use]

2013 pub fn new(

2114 name: impl Into<String>,

2215 old_oid: impl Into<String>,

@@ -29,16 +22,14 @@

2922 }

3023 }

3124

32 /// Whether this ref was deleted.

33 #[must_use]

25 #[must_use]

3426 pub fn is_delete(&self) -> bool {

35 self.is_zero(self.old_oid.as_str()) && !self.is_zero(self.new_oid.as_str())

27 self.is_zero(self.new_oid.as_str())

3628 }

3729

38 /// Whether this ref was created.

39 #[must_use]

30 #[must_use]

4031 pub fn is_create(&self) -> bool {

41 self.is_zero(self.new_oid.as_str())

32 self.is_zero(self.old_oid.as_str()) && !self.is_zero(self.new_oid.as_str())

4233 }

4334

4435 fn is_zero(&self, oid: &str) -> bool {

@@ -46,24 +37,17 @@

4637 }

4738}

4839

49/// A post-receive task queued for synchronization.

5040#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]

5141pub 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>,

42 pub id: Uuid,

43 pub repo_id: Uuid,

44 pub channel_id: Uuid,

45 pub actor_id: Option<Uuid>,

46 pub refs: Vec<PostReceiveRefUpdate>,

6247}

6348

6449impl PostReceiveTask {

65 /// Create a new post-receive task.

66 #[must_use]

50 #[must_use]

6751 pub fn new(

6852 repo_id: Uuid,

6953 channel_id: Uuid,

@@ -80,13 +64,6 @@

8064 }

8165}

8266

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.

9067pub async fn enqueue_post_receive(

9168 queue: &queue::Queue,

9269 settings: &crate::SyncSettings,

@@ -102,10 +79,6 @@

10279 .await

10380}

10481

105/// Publish a task to the dead-letter topic.

106///

107/// # Errors

108/// Returns [`queue::QueueError`] when the publish fails.

10982pub async fn enqueue_post_receive_dlq(

11083 queue: &queue::Queue,

11184 settings: &crate::SyncSettings,

@@ -116,3 +89,33 @@

11689 .publish_raw(&settings.dlq_topic, Some(task.repo_id.as_bytes()), &payload)

11790 .await

11891}

92

93#[cfg(test)]

94mod tests {

95 use super::*;

96

97 const ZERO: &str = "0000000000000000000000000000000000000000";

98 const AAA: &str = "aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa";

99 const BBB: &str = "bbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbb";

100

101 #[test]

102 fn create_when_old_is_zero() {

103 let u = PostReceiveRefUpdate::new("refs/heads/main", ZERO, AAA);

104 assert!(u.is_create());

105 assert!(!u.is_delete());

106 }

107

108 #[test]

109 fn delete_when_new_is_zero() {

110 let u = PostReceiveRefUpdate::new("refs/heads/main", AAA, ZERO);

111 assert!(u.is_delete());

112 assert!(!u.is_create());

113 }

114

115 #[test]

116 fn update_when_both_nonzero() {

117 let u = PostReceiveRefUpdate::new("refs/heads/main", AAA, BBB);

118 assert!(!u.is_create());

119 assert!(!u.is_delete());

120 }

121}

Modified lib/gitsync/fsck.rs +0 −4 View lib/gitsync/fsck.rs

@@ -1,8 +1,5 @@

1//! Repository integrity checks (fsck) and garbage collection.

2

31use git::gc::impls::GcOps;

42

5/// Run `git gc` on the blocking pool.

63pub async fn run_gc(git: &git::GitRepository) {

74 let git = git.clone();

85 if let Err(e) = tokio::task::spawn_blocking(move || git.gc_gc())

@@ -17,7 +14,6 @@

1714 }

1815}

1916

20/// Run `git fsck` and record integrity status. Non-fatal.

2117pub async fn run_fsck(git: &git::GitRepository) {

2218 let git = git.clone();

2319 let result = tokio::task::spawn_blocking(move || git.gc_fsck())

Modified lib/gitsync/languages.rs +0 −15 View lib/gitsync/languages.rs

@@ -1,5 +1,3 @@

1//! Language statistics synchronization (diff-based upsert).

2

31use std::collections::{HashMap, HashSet};

42

53use db::git::git_languages;

@@ -12,15 +10,7 @@

1210use crate::error::{SyncError, SyncResult};

1311use crate::settings::SyncSettings;

1412

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

2013/// swallowed (non-fatal).

21///

22/// # Errors

23/// Returns a database error when the transaction fails.

2414pub async fn sync_repo_languages(

2515 git: &git::GitRepository,

2616 txn: &sea_orm::DatabaseTransaction,

@@ -97,7 +87,6 @@

9787 Ok(())

9888}

9989

100/// A normalized language entry.

10190#[derive(Debug, Clone)]

10291struct LanguageEntry {

10392 language: String,

@@ -105,10 +94,6 @@

10594 size: i64,

10695}

10796

108/// Detect languages under a timeout, returning `None` on timeout.

109///

110/// # Errors

111/// Returns a sync error when detection fails.

11297async fn detect_languages(

11398 git: &git::GitRepository,

11499 settings: &SyncSettings,

Modified lib/gitsync/lib.rs +2 −7 View lib/gitsync/lib.rs

@@ -1,10 +1,3 @@

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

81pub mod branches;

92pub mod commits;

103pub mod error;

@@ -12,6 +5,7 @@

125pub mod fsck;

136pub mod languages;

147pub mod lock;

8pub mod post_receive;

159pub mod settings;

1610pub mod stats;

1711pub mod sync;

@@ -22,6 +16,7 @@

2216pub use error::{SyncError, SyncResult};

2317pub use event::{PostReceiveRefUpdate, PostReceiveTask, enqueue_post_receive};

2418pub use lock::{SyncLock, acquire_lock};

19pub use post_receive::{enqueue_post_receive_push, prepare_post_receive};

2520pub use settings::SyncSettings;

2621pub use sync::RepoSync;

2722pub use webhook::{NoopWebhookTx, WebhookTx};

Modified lib/gitsync/lock.rs +4 −26 View lib/gitsync/lock.rs

@@ -1,5 +1,3 @@

1//! Per-repository distributed sync lock backed by Redis.

2

31use std::process;

42use std::sync::Arc;

53

@@ -8,10 +6,8 @@

86

97use crate::{SyncError, SyncResult};

108

11/// Redis key prefix for sync locks.

129const LOCK_KEY_PREFIX: &str = "git:sync:lock:";

1310

14/// A held distributed lock. Release explicitly via [`SyncLock::release`].

1511pub struct SyncLock {

1612 kv: Arc<RedisCache>,

1713 key: String,

@@ -19,36 +15,24 @@

1915}

2016

2117impl SyncLock {

22 /// The token identifying this lock holder.

23 #[must_use]

18 #[must_use]

2419 pub fn value(&self) -> &str {

2520 &self.value

2621 }

2722

28 /// The Redis key under which this lock lives.

29 #[must_use]

23 #[must_use]

3024 pub fn key(&self) -> &str {

3125 &self.key

3226 }

3327

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> {

28 pub async fn refresh(&self, ttl_secs: u64) -> SyncResult<bool> {

4129 self.kv

4230 .refresh_lock(&self.key, self.value.as_bytes(), ttl_secs)

4331 .await

4432 .map_err(SyncError::Redis)

4533 }

4634

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<()> {

35 pub async fn release(&self) -> SyncResult<()> {

5236 self.kv

5337 .compare_and_delete(&self.key, self.value.as_bytes())

5438 .await

@@ -61,12 +45,6 @@

6145 format!("{LOCK_KEY_PREFIX}{repo_id}")

6246}

6347

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.

7048pub async fn acquire_lock(

7149 kv: &Arc<RedisCache>,

7250 repo_id: Uuid,

Added lib/gitsync/post_receive.rs +262 −0 View lib/gitsync/post_receive.rs

@@ -0,0 +1,262 @@

1use std::fs;

2use std::io;

3use std::os::unix::fs::PermissionsExt;

4use std::path::{Path, PathBuf};

5

6use uuid::Uuid;

7

8use crate::SyncSettings;

9use crate::event::{PostReceiveRefUpdate, PostReceiveTask, enqueue_post_receive};

10

11pub const POST_RECEIVE_HOOK: &str = "post-receive";

12pub const CAPTURE_FILE: &str = "_gta_post_receive";

13

14const HEX_OID_LEN: usize = 40;

15const HEX_OID_SHA256_LEN: usize = 64;

16

17pub fn hook_path(bare: &Path) -> PathBuf {

18 bare.join("hooks").join(POST_RECEIVE_HOOK)

19}

20

21pub fn capture_path(bare: &Path) -> PathBuf {

22 bare.join(CAPTURE_FILE)

23}

24

25pub fn ensure_post_receive_hook(bare: &Path) -> io::Result<()> {

26 let path = hook_path(bare);

27 if path.exists() {

28 return Ok(());

29 }

30 let hooks_dir = path.parent().expect("hook path always has a parent");

31 fs::create_dir_all(hooks_dir)?;

32 let script = format!(

33 "#!/bin/sh\n# gta post-receive capture hook\necho \"$$\" > /dev/null\nwhile read old_oid new_oid refname; do\n printf '%s %s %s\\n' \"$old_oid\" \"$new_oid\" \"$refname\" >> \"$GIT_DIR/{CAPTURE_FILE}\"\ndone\n"

34 );

35 fs::write(&path, script)?;

36 fs::set_permissions(&path, fs::Permissions::from_mode(0o755))?;

37 Ok(())

38}

39

40pub fn clear_post_receive_capture(bare: &Path) -> io::Result<()> {

41 match fs::remove_file(capture_path(bare)) {

42 Ok(()) => Ok(()),

43 Err(e) if e.kind() == io::ErrorKind::NotFound => Ok(()),

44 Err(e) => Err(e),

45 }

46}

47

48pub fn prepare_post_receive(bare: &Path) -> io::Result<()> {

49 ensure_post_receive_hook(bare)?;

50 clear_post_receive_capture(bare)

51}

52

53pub fn take_post_receive_capture(bare: &Path) -> Vec<PostReceiveRefUpdate> {

54 let path = capture_path(bare);

55 let content = match fs::read_to_string(&path) {

56 Ok(c) => c,

57 Err(_) => return Vec::new(),

58 };

59 let _ = fs::remove_file(&path);

60 content.lines().filter_map(parse_capture_line).collect()

61}

62

63fn parse_capture_line(line: &str) -> Option<PostReceiveRefUpdate> {

64 let mut parts = line.split_whitespace();

65 let old_oid = parts.next()?;

66 let new_oid = parts.next()?;

67 let name = parts.next()?;

68 if !is_valid_oid(old_oid) || !is_valid_oid(new_oid) || name.is_empty() {

69 return None;

70 }

71 Some(PostReceiveRefUpdate::new(name, old_oid, new_oid))

72}

73

74fn is_valid_oid(oid: &str) -> bool {

75 let len = oid.len();

76 if len != HEX_OID_LEN && len != HEX_OID_SHA256_LEN {

77 return false;

78 }

79 oid.bytes().all(|b| b.is_ascii_hexdigit())

80}

81

82pub async fn enqueue_post_receive_push(

83 queue: &queue::Queue,

84 settings: &SyncSettings,

85 channel_id: Uuid,

86 repo_id: Uuid,

87 actor_id: Option<Uuid>,

88 bare: &Path,

89) -> Result<usize, queue::QueueError> {

90 let refs = take_post_receive_capture(bare);

91 if refs.is_empty() {

92 return Ok(0);

93 }

94 let task = PostReceiveTask::new(repo_id, channel_id, actor_id, refs);

95 enqueue_post_receive(queue, settings, &task).await?;

96 metrics::counter!("gitsync.enqueue", "result" => "ok").increment(1);

97 Ok(task.refs.len())

98}

99

100#[cfg(test)]

101mod tests {

102 use super::*;

103

104 fn temp_bare() -> tempfile::TempDir {

105 tempfile::TempDir::new().expect("create temp dir")

106 }

107

108 #[test]

109 fn hook_install_is_idempotent() {

110 let dir = temp_bare();

111 assert!(ensure_post_receive_hook(dir.path()).is_ok());

112 let path = hook_path(dir.path());

113 assert!(path.exists());

114 let first = fs::read_to_string(&path).unwrap();

115 assert!(first.contains(CAPTURE_FILE));

116 assert_ne!(fs::metadata(&path).unwrap().permissions().mode() & 0o111, 0);

117

118 assert!(ensure_post_receive_hook(dir.path()).is_ok());

119 let second = fs::read_to_string(&path).unwrap();

120 assert_eq!(first, second);

121 }

122

123 #[test]

124 fn hook_never_overwrites_existing_hook() {

125 let dir = temp_bare();

126 let path = hook_path(dir.path());

127 fs::create_dir_all(path.parent().unwrap()).unwrap();

128 fs::write(&path, "#!/bin/sh\nexit 0\n").unwrap();

129 assert!(ensure_post_receive_hook(dir.path()).is_ok());

130 assert_eq!(

131 fs::read_to_string(&path).unwrap(),

132 "#!/bin/sh\nexit 0\n",

133 "existing hooks must be preserved"

134 );

135 }

136

137 #[test]

138 fn capture_roundtrip() {

139 let dir = temp_bare();

140 let bare = dir.path();

141 let capture = capture_path(bare);

142 fs::write(

143 &capture,

144 format!(

145 "{} {} refs/heads/main\n",

146 "0".repeat(HEX_OID_LEN),

147 "a".repeat(HEX_OID_LEN)

148 ),

149 )

150 .unwrap();

151 use std::io::Write as _;

152 fs::OpenOptions::new()

153 .append(true)

154 .open(&capture)

155 .unwrap()

156 .write_all(

157 format!(

158 "garbage line with !not-an-oid\n{} {} refs/tags/v1.0\n",

159 "b".repeat(HEX_OID_LEN),

160 "0".repeat(HEX_OID_LEN)

161 )

162 .as_bytes(),

163 )

164 .unwrap();

165

166 let updates = take_post_receive_capture(bare);

167 assert!(!capture.exists(), "capture file must be removed");

168 assert_eq!(updates.len(), 2, "malformed line must be skipped");

169 assert_eq!(updates[0].name, "refs/heads/main");

170 assert_eq!(updates[0].old_oid, "0".repeat(HEX_OID_LEN));

171 assert_eq!(updates[0].new_oid, "a".repeat(HEX_OID_LEN));

172 assert!(updates[1].is_delete(), "zero new_oid means delete");

173 }

174

175 #[test]

176 fn clear_is_idempotent() {

177 let dir = temp_bare();

178 fs::write(capture_path(dir.path()), "stale\n").unwrap();

179 assert!(clear_post_receive_capture(dir.path()).is_ok());

180 assert!(!capture_path(dir.path()).exists());

181 assert!(clear_post_receive_capture(dir.path()).is_ok());

182 }

183

184 #[test]

185 fn empty_capture_yields_nothing() {

186 let dir = temp_bare();

187 assert!(take_post_receive_capture(dir.path()).is_empty());

188 }

189

190 #[test]

191 fn real_git_push_captured_by_hook() {

192 let git = match which_git() {

193 Some(g) => g,

194 None => {

195 eprintln!("git not available; skipping real-push integration test");

196 return;

197 }

198 };

199 let dir = temp_bare();

200 let bare = dir.path();

201 git_run(&git, bare, &["init", "--bare", "."]).unwrap();

202 assert!(

203 prepare_post_receive(bare).is_ok(),

204 "hook install must succeed"

205 );

206 assert!(hook_path(bare).exists(), "hook file must exist");

207

208 let work = tempfile::tempdir().unwrap();

209 let work_path = work.path();

210 git_run(&git, work_path, &["init", "-q"]).unwrap();

211 git_run(&git, work_path, &["config", "user.email", "t@t"]).unwrap();

212 git_run(&git, work_path, &["config", "user.name", "t"]).unwrap();

213 fs::write(work_path.join("a.txt"), "hello\n").unwrap();

214 git_run(&git, work_path, &["add", "."]).unwrap();

215 git_run(&git, work_path, &["commit", "-q", "-m", "init"]).unwrap();

216 git_run(

217 &git,

218 work_path,

219 &["remote", "add", "origin", bare.to_str().unwrap()],

220 )

221 .unwrap();

222 git_run(

223 &git,

224 work_path,

225 &["push", "-q", "origin", "HEAD:refs/heads/main"],

226 )

227 .unwrap();

228

229 let updates = take_post_receive_capture(bare);

230 assert_eq!(updates.len(), 1, "one ref update expected");

231 let update = &updates[0];

232 assert_eq!(update.name, "refs/heads/main");

233 assert!(update.is_create(), "new branch push must be a create");

234 assert!(is_valid_oid(&update.new_oid), "new oid must be a real sha");

235 }

236

237 fn which_git() -> Option<std::path::PathBuf> {

238 let out = std::process::Command::new("git")

239 .arg("--version")

240 .output()

241 .ok()?;

242 out.status.success().then(|| "git".into())

243 }

244

245 fn git_run(git: &Path, cwd: &Path, args: &[&str]) -> io::Result<()> {

246 let out = std::process::Command::new(git)

247 .args(args)

248 .current_dir(cwd)

249 .stdout(std::process::Stdio::piped())

250 .stderr(std::process::Stdio::piped())

251 .output()?;

252 if out.status.success() {

253 Ok(())

254 } else {

255 Err(io::Error::other(format!(

256 "git {} failed: {}",

257 args.join(" "),

258 String::from_utf8_lossy(&out.stderr)

259 )))

260 }

261 }

262}

Modified lib/gitsync/settings.rs +11 −25 View lib/gitsync/settings.rs

@@ -1,28 +1,16 @@

1//! Configuration for the sync worker and per-repository sync.

2

31use std::time::Duration;

42

5/// Settings controlling synchronization behaviour.

63#[derive(Debug, Clone)]

74pub 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,

5 pub queue_topic: String,

6 pub dlq_topic: String,

7 pub group_id: String,

8 pub concurrency: usize,

9 pub lock_ttl_secs: u64,

10 pub max_commits_per_sync: usize,

11 pub max_lock_retries: u32,

12 pub language_timeout_secs: u64,

13 pub max_languages: usize,

2614}

2715

2816impl Default for SyncSettings {

@@ -42,14 +30,12 @@

4230}

4331

4432impl SyncSettings {

45 /// Lock TTL as a [`Duration`].

46 #[must_use]

33 #[must_use]

4734 pub fn lock_ttl(&self) -> Duration {

4835 Duration::from_secs(self.lock_ttl_secs)

4936 }

5037

51 /// Language detection timeout as a [`Duration`].

52 #[must_use]

38 #[must_use]

5339 pub fn language_timeout(&self) -> Duration {

5440 Duration::from_secs(self.language_timeout_secs)

5541 }

Modified lib/gitsync/stats.rs +0 −6 View lib/gitsync/stats.rs

@@ -1,15 +1,9 @@

1//! Repository statistics updates.

2

31use db::repo::repos;

42use git::gc::impls::GcOps;

53use sea_orm::{ActiveModelTrait, IntoActiveModel, Set};

64

75use crate::error::{SyncError, SyncResult};

86

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.

137pub async fn sync_repo_stats(

148 git: &git::GitRepository,

159 txn: &sea_orm::DatabaseTransaction,

Modified lib/gitsync/sync.rs +5 −28 View lib/gitsync/sync.rs

@@ -1,5 +1,3 @@

1//! Core per-repository synchronization.

2

31use std::sync::Arc;

42

53use cache::{Cache, RedisCache};

@@ -18,10 +16,6 @@

1816use crate::settings::SyncSettings;

1917use crate::{branches, languages, stats, tags};

2018

21/// Runnable repository synchronizer.

22///

23/// Build via [`RepoSync::new`]. All git operations are blocking and therefore

24/// executed on the blocking thread pool.

2519#[derive(Clone)]

2620pub struct RepoSync {

2721 db: Db,

@@ -32,8 +26,7 @@

3226}

3327

3428impl RepoSync {

35 /// Create a synchronizer for a single repository.

36 #[must_use]

29 #[must_use]

3730 pub fn new(

3831 db: Db,

3932 config: SharedRepoConfig,

@@ -59,20 +52,12 @@

5952 }

6053 }

6154

62 /// The repository id being synchronized.

63 #[must_use]

55 #[must_use]

6456 pub fn repo_id(&self) -> uuid::Uuid {

6557 self.repo.id

6658 }

6759

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<()> {

60 pub async fn sync(&self) -> SyncResult<()> {

7661 let ttl = self.settings.lock_ttl_secs;

7762 let Some(lock) = lock::acquire_lock(&self.kv, self.repo.id, ttl).await? else {

7863 return Err(SyncError::Locked);

@@ -110,11 +95,7 @@

11095 result

11196 }

11297

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<()> {

98 pub async fn sync_work(&self) -> SyncResult<()> {

11899 let new_commits = commits::collect_new_commits_for_sync(

119100 &self.git,

120101 &self.db,

@@ -164,11 +145,7 @@

164145 Ok((branch_count, synced_commits, tag_count))

165146 }

166147

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<()> {

148 pub fn set_default_branch(&self, name: &str) -> SyncResult<()> {

172149 gitcli::run(

173150 &self.git.path,

174151 &["symbolic-ref", "HEAD", &format!("refs/heads/{name}")],

Modified lib/gitsync/tags.rs +0 −15 View lib/gitsync/tags.rs

@@ -1,5 +1,3 @@

1//! Tag reconciliation.

2

31use std::collections::HashMap;

42

53use db::git::git_tags;

@@ -12,16 +10,8 @@

1210

1311use crate::error::{SyncError, SyncResult};

1412

15/// Upper bound on tags enumerated per sync.

1613const MAX_TAGS: u64 = 10_000;

1714

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.

2515pub async fn sync_tags(

2616 git: &git::GitRepository,

2717 txn: &sea_orm::DatabaseTransaction,

@@ -90,7 +80,6 @@

9080 Ok(collected.len())

9181}

9282

93/// A tag tip collected from the repository.

9483#[derive(Debug, Clone)]

9584pub(crate) struct TagTip {

9685 pub name: String,

@@ -101,10 +90,6 @@

10190 pub tagger_email: Option<String>,

10291}

10392

104/// Collect current tag tips from the bare repository.

105///

106/// # Errors

107/// Returns a sync error when enumeration fails.

10893pub(crate) async fn collect_tag_tips(git: &git::GitRepository) -> SyncResult<Vec<TagTip>> {

10994 let git = git.clone();

11095 tokio::task::spawn_blocking(move || {

Modified lib/gitsync/webhook.rs +1 −10 View lib/gitsync/webhook.rs

@@ -1,21 +1,12 @@

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

71use std::error::Error;

82

93use crate::event::PostReceiveTask;

104

11/// Dispatches webhook notifications for a completed sync.

125#[async_trait::async_trait]

136pub trait WebhookTx: Send + Sync {

14 /// Dispatch webhooks for the given task.

15 async fn dispatch(&self, task: &PostReceiveTask) -> Result<(), Box<dyn Error + Send + Sync>>;

7 async fn dispatch(&self, task: &PostReceiveTask) -> Result<(), Box<dyn Error + Send + Sync>>;

168}

179

18/// Placeholder dispatcher that only logs.

1910#[derive(Debug, Default, Clone)]

2011pub struct NoopWebhookTx;

2112

Modified lib/gitsync/worker.rs +4 −25 View lib/gitsync/worker.rs

@@ -1,5 +1,3 @@

1//! Sync worker: consumes post-receive tasks from Kafka and syncs repos.

2

31use std::sync::Arc;

42

53use cache::{Cache, RedisCache};

@@ -16,12 +14,9 @@

1614use crate::webhook::WebhookTx;

1715use crate::{SyncError, SyncResult};

1816

19/// Attempts-tracking key prefix for lock-contended tasks.

2017const ATTEMPT_KEY_PREFIX: &str = "git:sync:attempts:";

21/// Retention for the attempts counter.

2218const ATTEMPT_TTL_SECS: u64 = 3600;

2319

24/// Worker that processes post-receive sync tasks.

2520#[derive(Clone)]

2621pub struct SyncWorker {

2722 db: Db,

@@ -35,17 +30,12 @@

3530 concurrency: usize,

3631}

3732

38/// Handle to a running worker; call [`shutdown`](RunningWorker::shutdown) to stop.

3933pub struct RunningWorker {

4034 handles: Vec<queue::SubscribeHandle>,

4135}

4236

4337impl 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<()> {

38 pub async fn shutdown(self) -> SyncResult<()> {

4939 for handle in self.handles {

5040 handle.shutdown().await.map_err(SyncError::Queue)?;

5141 }

@@ -54,8 +44,7 @@

5444}

5545

5646impl SyncWorker {

57 /// Create a new worker.

58 #[must_use]

47 #[must_use]

5948 #[allow(clippy::too_many_arguments)]

6049 pub fn new(

6150 db: Db,

@@ -81,22 +70,13 @@

8170 }

8271 }

8372

84 /// Override the number of concurrent consumer loops.

85 #[must_use]

73 #[must_use]

8674 pub fn with_concurrency(mut self, concurrency: usize) -> Self {

8775 self.concurrency = concurrency.max(1);

8876 self

8977 }

9078

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> {

79 pub async fn run(self) -> SyncResult<RunningWorker> {

10080 let Some(queue) = self.queue else {

10181 tracing::warn!("sync queue disabled; gitsync worker not started");

10282 return Ok(RunningWorker {

@@ -209,7 +189,6 @@

209189 }

210190}

211191

212/// Re-publish a lock-contended task, routing to the DLQ after too many attempts.

213192async fn requeue_on_lock(

214193 queue: &Queue,

215194 kv: &Arc<RedisCache>,