refactor(git): log internal errors across graphql, smart, ssh, lfs and sync

Modified lib/git/src/graphql/error.rs +7 −0 View lib/git/src/graphql/error.rs

@@ -19,6 +19,12 @@

1919impl ErrorExtensions for GitError {

2020 fn extend(&self) -> async_graphql::Error {

2121 let code = code_for(self);

22 match self {

23 GitError::Cache(_) | GitError::Git(_) | GitError::Internal(_) => {

24 tracing::error!(error = %self, "graphql internal error");

25 }

26 _ => {}

27 }

2228 async_graphql::Error::new(self.to_string()).extend_with(|_, e| {

2329 e.set("code", code);

2430 })

@@ -34,6 +40,7 @@

3440/// gitbaby (low-level) errors carry no GitError context; surface them as

3541/// GIT_ERROR-tagged GraphQL errors.

3642pub fn map_git_baby<E: std::fmt::Display>(err: E) -> async_graphql::Error {

43 tracing::error!(error = %err, "graphql gitbaby error");

3744 async_graphql::Error::new(err.to_string()).extend_with(|_, e| {

3845 e.set("code", "GIT_ERROR");

3946 })

Modified lib/git/src/lfs/handle.rs +1 −0 View lib/git/src/lfs/handle.rs

@@ -376,6 +376,7 @@

376376 let message = match &err {

377377 GitError::Unauthorized => "authentication required".to_string(),

378378 GitError::Cache(_) | GitError::Git(_) | GitError::Internal(_) => {

379 tracing::error!(error = %err, "lfs internal error");

379380 "internal server error".to_string()

380381 }

381382 _ => err.to_string(),

Modified lib/git/src/smart/dispatch.rs +8 −3 View lib/git/src/smart/dispatch.rs

@@ -172,10 +172,15 @@

172172 .await

173173 {

174174 Ok(n) if n > 0 => {

175 tracing::debug!(count = n, "enqueued post-receive ref updates")

175 tracing::debug!(

176 count = n,

177 space_id = %space_id,

178 repo_id = %repo_id,

179 "enqueued post-receive ref updates"

180 )

176181 }

177 Ok(_) => tracing::debug!("no refs to enqueue"),

178 Err(e) => tracing::warn!(?e, "post-receive enqueue failed"),

182 Ok(_) => tracing::debug!(space_id = %space_id, repo_id = %repo_id, "no refs to enqueue"),

183 Err(e) => tracing::warn!(error = %e, space_id = %space_id, repo_id = %repo_id, "post-receive enqueue failed"),

179184 }

180185 }

181186 Ok::<_, axum::Error>(Bytes::new())

Modified lib/git/src/smart/push.rs +5 −1 View lib/git/src/smart/push.rs

@@ -239,7 +239,11 @@

239239 _ = stop_clone.notified() => break,

240240 _ = tokio::time::sleep(Duration::from_secs(interval)) => {

241241 match backend.renew(&key, token.as_bytes(), ttl).await {

242 Ok(false) | Err(_) => break,

242 Ok(false) => break,

243 Err(error) => {

244 tracing::warn!(%error, "lock renew failed, stopping renewer");

245 break;

246 }

243247 Ok(true) => {}

244248 }

245249 }

Modified lib/git/src/ssh/handler/channel.rs +21 −4 View lib/git/src/ssh/handler/channel.rs

@@ -210,10 +210,24 @@

210210 .await

211211 {

212212 Ok(n) if n > 0 => {

213 tracing::debug!(count = n, "enqueued post-receive ref updates")

213 tracing::debug!(

214 count = n,

215 space_id = %space_id,

216 repo_id = %repo_id,

217 "enqueued post-receive ref updates"

218 )

214219 }

215 Ok(_) => tracing::debug!("no refs to enqueue"),

216 Err(e) => tracing::warn!(?e, "post-receive enqueue failed"),

220 Ok(_) => tracing::debug!(

221 space_id = %space_id,

222 repo_id = %repo_id,

223 "no refs to enqueue"

224 ),

225 Err(e) => tracing::warn!(

226 error = %e,

227 space_id = %space_id,

228 repo_id = %repo_id,

229 "post-receive enqueue failed"

230 ),

217231 }

218232 }

219233 let code = if result.is_ok() { 0 } else { 1 };

@@ -381,7 +395,10 @@

381395 let repo = repos::Entity::find_by_id(repo_id).one(db.read()).await.ok()??;

382396 let cache = match CacheClient::new(&service.smart.git.state.config.cache).await {

383397 Ok(c) => Arc::new(c),

384 Err(_) => return None,

398 Err(error) => {

399 tracing::warn!(%error, "cache client init failed, skipping pre-receive check");

400 return None;

401 }

385402 };

386403 let git_cfg = &service.smart.git.state.config.git;

387404 let ctx = GitContext {

Modified lib/git/src/ssh/handler/command.rs +4 −1 View lib/git/src/ssh/handler/command.rs

@@ -75,7 +75,10 @@

7575 };

7676 let space_id = match self.space_id_for(&ctx.model.dir_id).await {

7777 Ok(id) => id,

78 Err(_) => ctx.model.dir_id,

78 Err(error) => {

79 tracing::warn!(%error, "space lookup failed, falling back to dir id");

80 ctx.model.dir_id

81 }

7982 };

8083

8184 let bare = ctx.repo_dir();

Modified lib/git/src/sync/branches.rs +14 −8 View lib/git/src/sync/branches.rs

@@ -131,15 +131,21 @@

131131 let mut am = repo.clone().into_active_model();

132132 am.default_branch = Set(desired.clone());

133133 am.update(txn).await.map_err(SyncError::Db)?;

134 }

134135

135 if let Err(e) = baby.set_default_branch(&desired).await {

136 tracing::warn!(

137 repo_id = %repo.id,

138 branch = %desired,

139 error = %e,

140 "failed to update on-disk HEAD symbolic-ref"

141 );

142 }

136 let head_ok = matches!(

137 baby.get_default_branch().await,

138 Ok(current) if current == desired

139 );

140 if !head_ok

141 && let Err(e) = baby.set_default_branch(&desired).await

142 {

143 tracing::warn!(

144 repo_id = %repo.id,

145 branch = %desired,

146 error = %e,

147 "failed to update on-disk HEAD symbolic-ref"

148 );

143149 }

144150

145151 Ok(())

Modified lib/git/src/sync/event.rs +19 −5 View lib/git/src/sync/event.rs

@@ -81,15 +81,23 @@

8181 task: &PostReceiveTask,

8282) -> Result<(), EnqueueError> {

8383 let payload = serde_json::to_vec(task)?;

84 queue

84 let queue_len = queue

8585 .publish_raw(

8686 &settings.queue_topic,

8787 Some(task.repo_id.as_bytes()),

8888 &payload,

8989 )

9090 .await

91 .map(|_| ())

92 .map_err(|e| EnqueueError::Queue(e.to_string()))

91 .map_err(|e| EnqueueError::Queue(e.to_string()))?;

92 tracing::info!(

93 task_id = %task.id,

94 repo_id = %task.repo_id,

95 refs = task.refs.len(),

96 topic = %settings.queue_topic,

97 queue_len,

98 "sync task enqueued"

99 );

100 Ok(())

93101}

94102

95103pub async fn enqueue_post_receive_dlq(

@@ -101,6 +109,12 @@

101109 queue

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

103111 .await

104 .map(|_| ())

105 .map_err(|e| EnqueueError::Queue(e.to_string()))

112 .map_err(|e| EnqueueError::Queue(e.to_string()))?;

113 tracing::debug!(

114 task_id = %task.id,

115 repo_id = %task.repo_id,

116 topic = %settings.dlq_topic,

117 "task enqueued to dlq"

118 );

119 Ok(())

106120}

Modified lib/git/src/sync/languages.rs +45 −4 View lib/git/src/sync/languages.rs

@@ -22,7 +22,7 @@

2222 repo: &repos::Model,

2323 settings: &SyncSettings,

2424) -> SyncResult<()> {

25 let desired = match detect_languages(baby, settings).await {

25 let desired = match detect_languages(baby, repo, settings).await {

2626 Ok(Some(langs)) => langs,

2727 Ok(None) => return Ok(()),

2828 Err(e) => {

@@ -102,10 +102,19 @@

102102

103103async fn detect_languages(

104104 baby: &GitBaby,

105 repo: &repos::Model,

105106 settings: &SyncSettings,

106107) -> SyncResult<Option<Vec<LanguageEntry>>> {

108 let Some(revision) = resolve_language_revision(baby, repo).await else {

109 tracing::debug!(repo_id = %repo.id, "no resolvable revision; skipping language detection");

110 return Ok(None);

111 };

107112 let timeout = Duration::from_secs(settings.language_timeout_secs);

108 let stats = tokio::time::timeout(timeout, walk_lang_stats(baby.clone())).await;

113 let stats = tokio::time::timeout(

114 timeout,

115 walk_lang_stats(baby.clone(), revision),

116 )

117 .await;

109118

110119 let entries: Vec<(String, i64)> = match stats {

111120 Ok(Ok(map)) => map.into_iter().collect(),

@@ -140,10 +149,42 @@

140149 Ok(Some(out))

141150}

142151

143async fn walk_lang_stats(baby: GitBaby) -> SyncResult<std::collections::HashMap<String, i64>> {

152async fn resolve_language_revision(

153 baby: &GitBaby,

154 repo: &repos::Model,

155) -> Option<String> {

156 let mut candidates: Vec<String> = Vec::new();

157 candidates.push("HEAD".to_string());

158 let db = repo.default_branch.trim();

159 if !db.is_empty() {

160 let head = format!("refs/heads/{db}");

161 if !candidates.contains(&head) {

162 candidates.push(head);

163 }

164 }

165 for cand in &candidates {

166 match baby.rev_parse(cand).await {

167 Ok(_) => return Some(cand.clone()),

168 Err(gitbaby::error::BabyError::RevisionNotFound(_)) => continue,

169 Err(_) => continue,

170 }

171 }

172 match baby

173 .list_branches(gitbaby::branch::ListBranchesOptions::default())

174 .await

175 {

176 Ok(branches) => branches.first().map(|b| b.sha.to_string()),

177 Err(_) => None,

178 }

179}

180

181async fn walk_lang_stats(

182 baby: GitBaby,

183 revision: String,

184) -> SyncResult<std::collections::HashMap<String, i64>> {

144185 let entries = baby

145186 .walk_tree(

146 "HEAD",

187 &revision,

147188 std::path::Path::new(""),

148189 WalkTreeOptions {

149190 recurse: true,

Modified lib/git/src/sync/post_receive.rs +56 −10 View lib/git/src/sync/post_receive.rs

@@ -26,26 +26,42 @@

2626

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

2828 let path = hook_path(bare);

29 let script = capture_script();

2930 if path.exists() {

30 tracing::warn!(

31 path = %path.display(),

32 "existing post-receive hook will not be overwritten; post-receive sync may not capture pushes"

33 );

31 let existing = fs::read_to_string(&path)?;

32 if !existing.contains(CAPTURE_FILE) {

33 tracing::warn!(

34 path = %path.display(),

35 "existing post-receive hook will not be overwritten; post-receive sync may not capture pushes"

36 );

37 return Ok(());

38 }

39 if existing == script {

40 return Ok(());

41 }

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

43 #[cfg(unix)]

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

45 tracing::debug!(path = %path.display(), "refreshed post-receive capture hook");

3446 return Ok(());

3547 }

3648 let Some(hooks_dir) = path.parent() else {

3749 return Err(io::Error::other("hook path has no parent"));

3850 };

3951 fs::create_dir_all(hooks_dir)?;

40 let script = format!(

41 "#!/bin/sh\n# gitdata post-receive capture hook\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"

42 );

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

4453 #[cfg(unix)]

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

55 tracing::debug!(path = %path.display(), "installed post-receive capture hook");

4656 Ok(())

4757}

4858

59fn capture_script() -> String {

60 format!(

61 "#!/bin/sh\n# gitdata post-receive capture hook\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"

62 )

63}

64

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

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

5167 Ok(()) => Ok(()),

@@ -63,10 +79,40 @@

6379 let path = capture_path(bare);

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

6581 Ok(c) => c,

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

82 Err(e) if e.kind() == std::io::ErrorKind::NotFound => {

83 tracing::warn!(

84 bare = %bare.display(),

85 "post-receive capture file missing after push; capture hook may not have run"

86 );

87 return Vec::new();

88 }

89 Err(e) => {

90 tracing::warn!(error = %e, "failed to read post-receive capture file");

91 return Vec::new();

92 }

6793 };

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

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

94 if let Err(e) = fs::remove_file(&path) {

95 tracing::warn!(error = %e, "failed to remove post-receive capture file");

96 }

97 let mut dropped = 0usize;

98 let updates: Vec<_> = content

99 .lines()

100 .filter_map(|line| {

101 let parsed = parse_capture_line(line);

102 if parsed.is_none() {

103 dropped += 1;

104 }

105 parsed

106 })

107 .collect();

108 if dropped > 0 {

109 tracing::warn!(

110 bare = %bare.display(),

111 dropped,

112 "discarded unparseable post-receive capture lines"

113 );

114 }

115 updates

70116}

71117

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

Modified lib/git/src/sync/sync.rs +16 −0 View lib/git/src/sync/sync.rs

@@ -129,13 +129,25 @@

129129 self.settings.max_commits_per_sync,

130130 )

131131 .await?;

132 tracing::debug!(

133 repo_id = %self.repo.id,

134 new_commits = new_commits.len(),

135 "collected new commits for sync"

136 );

132137

133138 let txn: DatabaseTransaction = self.db.write().begin().await.map_err(SyncError::Db)?;

134139

135140 let (branch_count, _synced_commits, tag_count) =

136141 self.sync_refs_and_data(&txn, &new_commits).await?;

142 tracing::debug!(

143 repo_id = %self.repo.id,

144 branches = branch_count,

145 tags = tag_count,

146 "refs synced"

147 );

137148

138149 languages::sync_repo_languages(&baby, &txn, &self.repo, &self.settings).await?;

150 tracing::debug!(repo_id = %self.repo.id, "languages synced");

139151

140152stats::sync_repo_stats(

141153 &txn,

@@ -147,12 +159,16 @@

147159 self.settings.git_command_timeout_secs,

148160 )

149161 .await?;

162 tracing::debug!(repo_id = %self.repo.id, "stats synced");

150163

151164 license::sync_repo_license(&txn, &self.repo_dir(), &self.repo).await?;

165 tracing::debug!(repo_id = %self.repo.id, "license synced");

152166

153167 txn.commit().await.map_err(SyncError::Db)?;

168 tracing::debug!(repo_id = %self.repo.id, "sync transaction committed");

154169

155170 let gc_timeout = self.settings.git_command_timeout_secs.max(1800);

171 tracing::debug!(repo_id = %self.repo.id, "starting fsck/gc");

156172 fsck::run_gc(&self.repo_dir(), gc_timeout).await;

157173 fsck::run_fsck(&self.repo_dir(), gc_timeout).await;

158174

Modified lib/git/src/sync/webhook.rs +1 −0 View lib/git/src/sync/webhook.rs

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

1414impl WebhookTx for NoopWebhookTx {

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

1616 tracing::warn!(

17 task_id = %task.id,

1718 repo_id = %task.repo_id,

1819 refs = task.refs.len(),

1920 "webhook dispatch not configured; message dropped (NoopWebhookTx)"

Modified lib/git/src/sync/worker.rs +80 −12 View lib/git/src/sync/worker.rs

@@ -9,7 +9,7 @@

99use sea_orm::EntityTrait;

1010use tokio::task::JoinHandle;

1111use tokio_util::sync::CancellationToken;

12use tracing::{debug, error, warn};

12use tracing::{debug, error, info, warn};

1313

1414use crate::sync::error::{SyncError, SyncResult};

1515use crate::sync::event::PostReceiveTask;

@@ -62,7 +62,7 @@

6262 cancel: CancellationToken,

6363 ) -> Self {

6464 let concurrency = settings.concurrency.max(1);

65 debug!(

65 info!(

6666 concurrency,

6767 topic = %settings.queue_topic,

6868 "starting sync worker"

@@ -84,13 +84,18 @@

8484 let topic = self.settings.queue_topic.clone();

8585 let brpop = self.settings.brpop_timeout_secs;

8686

87 let mut queue = self.queue;

88 if queue.consumer_conns().is_empty() {

89 queue = SyncQueue::with_consumer_conns(queue.redis().clone(), concurrency).await?;

90 }

91

8792 let mut handles = Vec::with_capacity(concurrency);

8893 for i in 0..concurrency {

8994 let db = self.db.clone();

9095 let redis = self.redis.clone();

9196 let cache = self.cache.clone();

9297 let repo_root = self.repo_root.clone();

93 let queue = self.queue.clone();

98 let queue = queue.clone();

9499 let webhook = self.webhook.clone();

95100 let settings = self.settings.clone();

96101 let cancel = self.cancel.clone();

@@ -130,7 +135,7 @@

130135 let payload = tokio::select! {

131136 biased;

132137 _ = cancel.cancelled() => break,

133 res = queue.consume(topic, brpop_timeout) => match res {

138 res = queue.consume(topic, brpop_timeout, id) => match res {

134139 Ok(Some(p)) => p,

135140 Ok(None) => continue,

136141 Err(e) => {

@@ -144,19 +149,44 @@

144149 let task: PostReceiveTask = match serde_json::from_slice(&payload) {

145150 Ok(t) => t,

146151 Err(e) => {

147 error!(consumer = id, error = %e, "invalid sync task payload; dropping");

152 error!(

153 consumer = id,

154 error = %e,

155 payload_len = payload.len(),

156 "invalid sync task payload; dropping"

157 );

148158 continue;

149159 }

150160 };

151161

162 let started = std::time::Instant::now();

163 debug!(

164 consumer = id,

165 task_id = %task.id,

166 repo_id = %task.repo_id,

167 refs = task.refs.len(),

168 "sync task received"

169 );

170

152171 let repo = match repos::Entity::find_by_id(task.repo_id).one(db.read()).await {

153172 Ok(Some(r)) => r,

154173 Ok(None) => {

155 warn!(consumer = id, repo_id = %task.repo_id, "repo not found; skipping");

174 warn!(

175 consumer = id,

176 task_id = %task.id,

177 repo_id = %task.repo_id,

178 "repo not found; skipping"

179 );

156180 continue;

157181 }

158182 Err(e) => {

159 error!(consumer = id, repo_id = %task.repo_id, error = %e, "repo lookup failed; skipping");

183 error!(

184 consumer = id,

185 task_id = %task.id,

186 repo_id = %task.repo_id,

187 error = %e,

188 "repo lookup failed; skipping"

189 );

160190 continue;

161191 }

162192 };

@@ -172,20 +202,56 @@

172202

173203 match rsync.sync().await {

174204 Ok(()) => {

205 info!(

206 consumer = id,

207 task_id = %task.id,

208 repo_id = %task.repo_id,

209 elapsed_ms = started.elapsed().as_millis() as u64,

210 "sync task completed"

211 );

175212 if let Err(e) = webhook.dispatch(&task).await {

176 warn!(consumer = id, repo_id = %task.repo_id, error = %e, "webhook dispatch failed");

213 warn!(

214 consumer = id,

215 task_id = %task.id,

216 repo_id = %task.repo_id,

217 error = %e,

218 "webhook dispatch failed"

219 );

177220 }

178221 }

179222 Err(e) if e.is_locked() => {

180 warn!(consumer = id, repo_id = %task.repo_id, "sync locked; requeueing");

223 warn!(

224 consumer = id,

225 task_id = %task.id,

226 repo_id = %task.repo_id,

227 "sync locked; requeueing"

228 );

181229 if let Err(e2) = requeue_or_dlq(queue, settings, &task, &payload).await {

182 error!(consumer = id, repo_id = %task.repo_id, error = %e2, "requeue failed");

230 error!(

231 consumer = id,

232 task_id = %task.id,

233 repo_id = %task.repo_id,

234 error = %e2,

235 "requeue failed"

236 );

183237 }

184238 }

185239 Err(e) => {

186 error!(consumer = id, repo_id = %task.repo_id, error = %e, "sync task failed");

240 error!(

241 consumer = id,

242 task_id = %task.id,

243 repo_id = %task.repo_id,

244 error = %e,

245 "sync task failed"

246 );

187247 if let Err(e2) = requeue_or_dlq(queue, settings, &task, &payload).await {

188 error!(consumer = id, repo_id = %task.repo_id, error = %e2, "requeue failed");

248 error!(

249 consumer = id,

250 task_id = %task.id,

251 repo_id = %task.repo_id,

252 error = %e2,

253 "requeue failed"

254 );

189255 }

190256 }

191257 }

@@ -208,6 +274,7 @@

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

209275 .await?;

210276 warn!(

277 task_id = %task.id,

211278 repo_id = %task.repo_id,

212279 attempts,

213280 "task exceeded retry limit; sent to DLQ"

@@ -224,6 +291,7 @@

224291 )

225292 .await?;

226293 debug!(

294 task_id = %task.id,

227295 repo_id = %task.repo_id,

228296 attempts,

229297 "re-queued retry task"