chore(git): trace smart-http, ssh session and sync pipelines

Modified lib/git/src/smart/dispatch.rs +31 −0 View lib/git/src/smart/dispatch.rs

@@ -44,6 +44,13 @@

4444 }

4545 };

4646 let bare = service.bare_path(&ctx);

47 tracing::debug!(

48 space,

49 repo,

50 service = service_kind.unwrap_or("dumb"),

51 bare = %bare.display(),

52 "info/refs authorized, generating advertisement"

53 );

4754 match service_kind {

4855 Some(name) => {

4956 let git_service = GitService::from_name(name)?;

@@ -69,12 +76,30 @@

6976 let actor_id = authorize(service, &ctx, auth, want_write).await?;

7077 let bare = service.bare_path(&ctx);

7178 let space_id = service.owning_space_id(&ctx).await?;

79 tracing::info!(

80 space,

81 repo,

82 service = git_service.name(),

83 actor_id = ?actor_id,

84 repo_id = %ctx.model.id,

85 "git rpc authorized, processing request"

86 );

7287

7388 let run = async {

7489 let input = decode_request_body(headers, body, MAX_RPC_BODY).await?;

7590 let is_receive = git_service == GitService::ReceivePack;

91 tracing::debug!(

92 service = git_service.name(),

93 body_bytes = input.len(),

94 "rpc request body decoded"

95 );

7696 if is_receive && actor_id.is_some() {

7797 let updates = parse_receive_pack_refs(&input)?;

98 tracing::debug!(

99 repo_id = %ctx.model.id,

100 ref_count = updates.len(),

101 "parsed receive-pack ref updates"

102 );

78103 if !updates.is_empty()

79104 && let Some(actor_uid) = actor_id

80105 {

@@ -88,6 +113,7 @@

88113 )

89114 .await?;

90115 verdict.accepted_only()?;

116 tracing::debug!(repo_id = %ctx.model.id, "pre-receive validation passed");

91117 }

92118 if let Err(e) = crate::sync::post_receive::prepare_post_receive(&bare) {

93119 tracing::warn!(?e, "post-receive hook setup failed");

@@ -213,6 +239,11 @@

213239 "decompressed body exceeds limit of {limit} bytes"

214240 )));

215241 }

242 tracing::debug!(

243 compressed = bytes.len(),

244 decompressed = out.len(),

245 "gzip request body decompressed"

246 );

216247 Ok(Bytes::from(out))

217248 } else {

218249 Ok(bytes)

Modified lib/git/src/smart/handler.rs +19 −1 View lib/git/src/smart/handler.rs

@@ -37,6 +37,14 @@

3737 let method = req.method().clone();

3838 let body = req.into_body();

3939

40 tracing::debug!(

41 method = %method,

42 space = %space,

43 repo = %repo,

44 path = %path,

45 "smart http request received"

46 );

47

4048 let repo_name = match repo.strip_suffix(".git") {

4149 Some(name) if !name.is_empty() => name.to_string(),

4250 _ => return SmartError::BadRequest("expected '<repo>.git' in URL".into()).into_response(),

@@ -124,7 +132,17 @@

124132 }

125133 _ => Ok(SmartError::BadRequest("method not allowed".into()).into_response()),

126134 };

127 result.unwrap_or_else(|e| e.into_response())

135 result.unwrap_or_else(|e| {

136 tracing::warn!(

137 method = %method,

138 space = %space,

139 repo = %repo_name,

140 path = %path,

141 error = %e,

142 "smart http request failed"

143 );

144 e.into_response()

145 })

128146}

129147

130148/// Whitelist of paths the GET fallback may serve. The dumb HTTP protocol

Modified lib/git/src/smart/info_refs.rs +6 −0 View lib/git/src/smart/info_refs.rs

@@ -14,6 +14,12 @@

1414 let output = rpc::run_rpc_bytes(bare, service, true, Bytes::new(), timeout_secs).await?;

1515 let mut body = pktline::encode_header(service.dashed_name())?;

1616 body.extend_from_slice(&output);

17 tracing::debug!(

18 svc = service.name(),

19 bare = %bare.display(),

20 body_bytes = body.len(),

21 "smart ref advertisement generated"

22 );

1723

1824 Ok((

1925 StatusCode::OK,

Modified lib/git/src/smart/lfs.rs +14 −0 View lib/git/src/smart/lfs.rs

@@ -26,6 +26,7 @@

2626 body: Body,

2727 auth: &BasicAuth,

2828) -> SmartResult<Response> {

29 tracing::debug!(space, repo, path, method = %method, "lfs request received");

2930 match path {

3031 "info/lfs/objects/batch" if method == Method::POST => {

3132 let bytes = read_body(headers, body, MAX_LFS_BODY).await?;

@@ -82,7 +83,15 @@

8283 allow_repo_read(svc, auth, &ctx).await?;

8384 }

8485 }

86 tracing::debug!(

87 space,

88 repo,

89 operation = ?req.operation,

90 objects = req.objects.len(),

91 "lfs batch authorized"

92 );

8593 let resp = object::batch(svc, space.to_string(), repo.to_string(), req).await?;

94 tracing::debug!(space, repo, "lfs batch processed");

8695 Ok((

8796 StatusCode::OK,

8897 [(header::CONTENT_TYPE, "application/vnd.git-lfs+json")],

@@ -134,6 +143,7 @@

134143 .into_data_stream()

135144 .map(|chunk| chunk.map_err(|e| std::io::Error::other(e.to_string())));

136145 let reader = StreamReader::new(stream);

146 tracing::debug!(space, repo, oid = %canonical, size, "lfs object upload started");

137147 object::upload_stream(

138148 svc,

139149 space.to_string(),

@@ -145,6 +155,7 @@

145155 reader,

146156 )

147157 .await?;

158 tracing::debug!(space, repo, "lfs object upload completed");

148159 Ok(StatusCode::OK.into_response())

149160}

150161

@@ -160,6 +171,7 @@

160171 allow_repo_read(svc, auth, &ctx).await?;

161172 let canonical = crate::lfs::canonical_oid(oid)?;

162173 let key = format!("lfs/{}/{}", ctx.model.id.simple(), canonical);

174 tracing::debug!(space, repo, oid = %canonical, "lfs object download started");

163175 let stream = svc

164176 .state

165177 .storage

@@ -220,6 +232,7 @@

220232 None,

221233 )

222234 .await?;

235 tracing::debug!(space, repo, lock_id = %lock.id, "lfs lock created");

223236 Ok((

224237 StatusCode::CREATED,

225238 [(header::CONTENT_TYPE, "application/vnd.git-lfs+json")],

@@ -271,6 +284,7 @@

271284 let svc = &service.git;

272285 let ctx = service.resolve(space, repo).await?;

273286 let uid = require_writer(svc, auth, &ctx).await?;

287 tracing::debug!(space, repo, lock_id = %lock_id, "lfs unlock requested");

274288 crate::lfs::unlock_by_id(

275289 svc,

276290 space.to_string(),

Modified lib/git/src/smart/object.rs +11 −6 View lib/git/src/smart/object.rs

@@ -8,12 +8,15 @@

88pub async fn serve(bare: &Path, request_path: &str) -> SmartResult<Response> {

99 let file = resolve_within_root(bare, request_path)?;

1010 match tokio::fs::read(&file).await {

11 Ok(bytes) => Ok((

12 StatusCode::OK,

13 [(header::CONTENT_TYPE, guess_content_type(request_path))],

14 bytes,

15 )

16 .into_response()),

11 Ok(bytes) => {

12 tracing::debug!(path = %request_path, bytes = bytes.len(), "serving dumb http object");

13 Ok((

14 StatusCode::OK,

15 [(header::CONTENT_TYPE, guess_content_type(request_path))],

16 bytes,

17 )

18 .into_response())

19 }

1720 Err(e) if e.kind() == std::io::ErrorKind::NotFound => Err(SmartError::NotFound(format!(

1821 "file not found: {request_path}"

1922 ))),

@@ -28,6 +31,7 @@

2831 Component::Normal(part) => base.push(part),

2932 Component::CurDir => {}

3033 Component::ParentDir | Component::RootDir | Component::Prefix(_) => {

34 tracing::warn!(path = %request_path, "dumb http path traversal denied");

3135 return Err(SmartError::Forbidden("path traversal denied".into()));

3236 }

3337 }

@@ -41,6 +45,7 @@

4145 if canonical.starts_with(&root_canonical) {

4246 Ok(canonical)

4347 } else {

48 tracing::warn!(path = %request_path, "dumb http path escapes repository root");

4449 Err(SmartError::Forbidden("path escapes repository root".into()))

4550 }

4651}

Modified lib/git/src/smart/pre_receive.rs +21 −0 View lib/git/src/smart/pre_receive.rs

@@ -47,6 +47,12 @@

4747 actor_user_id: Uuid,

4848 refs: &[RefUpdate],

4949) -> SmartResult<PreReceiveVerdict> {

50 tracing::debug!(

51 %repo_id,

52 %actor_user_id,

53 ref_count = refs.len(),

54 "validating pre-receive ref updates"

55 );

5056 let repo = models::repo::repos::Entity::find_by_id(repo_id)

5157 .one(db)

5258 .await?

@@ -70,6 +76,13 @@

7076 actor_collab.map(|c| c.mode),

7177 Some(models::repo::repo_collaborations::CollaborationMode::Admin)

7278 );

79 tracing::debug!(

80 %repo_id,

81 branch_policies = branch_policies.len(),

82 tag_policies = tag_policies.len(),

83 actor_is_admin,

84 "loaded protection policies for pre-receive"

85 );

7386

7487 let branch_policy_ids: Vec<Uuid> = branch_policies.iter().map(|p| p.id).collect();

7588 let branch_whitelists = load_branch_whitelists(db, &branch_policy_ids)

@@ -97,6 +110,14 @@

97110 &zero_oid,

98111 )

99112 .await?;

113 match &verdict {

114 RefVerdict::Accept => {

115 tracing::debug!(refname = %r.refname, "pre-receive ref accepted");

116 }

117 RefVerdict::Reject(reason) => {

118 tracing::warn!(refname = %r.refname, %reason, "pre-receive ref rejected");

119 }

120 }

100121 out.push((r.clone(), verdict));

101122 }

102123 Ok(PreReceiveVerdict { per_ref: out })

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

@@ -165,9 +165,11 @@

165165 .acquire(key, token.as_bytes(), self.lock_ttl_secs)

166166 .await?

167167 {

168 tracing::debug!(key, ttl_secs = self.lock_ttl_secs, "push lock acquired");

168169 return self.run_locked(key, &token, f).await;

169170 }

170171

172 tracing::debug!(key, "push lock busy; queued, waiting for lock");

171173 self.backend.enqueue(key, token.as_bytes()).await?;

172174 loop {

173175 tokio::time::sleep(self.poll_interval).await;

@@ -177,6 +179,7 @@

177179 .await?

178180 {

179181 self.backend.dequeue(key, token.as_bytes()).await?;

182 tracing::debug!(key, "push lock acquired after waiting");

180183 return self.run_locked(key, &token, f).await;

181184 }

182185 }

@@ -201,6 +204,7 @@

201204 let result = f.await;

202205 heartbeat.stop().await;

203206 self.backend.release(key, token.as_bytes()).await?;

207 tracing::debug!(key, ok = result.is_ok(), "push lock released");

204208 result

205209 }

206210}

@@ -237,7 +241,10 @@

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

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

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

240 Ok(false) => break,

244 Ok(false) => {

245 tracing::debug!(key = %key, "push lock no longer held; stopping renewer");

246 break;

247 }

241248 Err(error) => {

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

243250 break;

Modified lib/git/src/smart/rpc.rs +44 −6 View lib/git/src/smart/rpc.rs

@@ -95,6 +95,15 @@

9595 let mut child = cmd.spawn().map_err(|e| {

9696 SmartError::Internal(format!("failed to spawn git {}: {e}", service.name()))

9797 })?;

98 tracing::debug!(

99 svc = service.name(),

100 advertise,

101 bare = %bare_path.display(),

102 input_bytes = input.len(),

103 timeout_secs,

104 pid = ?child.id(),

105 "spawned git rpc process"

106 );

98107 let mut stdin = child

99108 .stdin

100109 .take()

@@ -114,6 +123,10 @@

114123 let child_arc = child_arc.clone();

115124 tokio::spawn(async move {

116125 if stdin.write_all(&input).await.is_err() {

126 tracing::warn!(

127 svc = service.name(),

128 "writing rpc input to git stdin failed; killing process"

129 );

117130 if let Some(mut ch) = child_arc.lock().await.take() {

118131 let _ = ch.start_kill();

119132 let _ = ch.wait().await;

@@ -137,6 +150,11 @@

137150 {

138151 Ok(Ok(s)) => Some(s),

139152 _ => {

153 tracing::warn!(

154 svc = service.name(),

155 timeout_secs,

156 "git rpc process timed out; killing"

157 );

140158 let _ = ch.start_kill();

141159 let _ = ch.wait().await;

142160 None

@@ -160,12 +178,27 @@

160178 }

161179 }

162180 let verdict = match status {

163 Some(s) if s.success() => Ok(()),

164 Some(s) => Err(SmartError::Internal(format!(

165 "git {} exited with {s}: {}",

166 service.name(),

167 String::from_utf8_lossy(text.as_slice()).trim()

168 ))),

181 Some(s) if s.success() => {

182 tracing::debug!(

183 svc = service.name(),

184 stderr_bytes = total,

185 "git rpc process exited successfully"

186 );

187 Ok(())

188 }

189 Some(s) => {

190 tracing::warn!(

191 svc = service.name(),

192 exit = %s,

193 stderr = %String::from_utf8_lossy(text.as_slice()).trim(),

194 "git rpc process exited with error"

195 );

196 Err(SmartError::Internal(format!(

197 "git {} exited with {s}: {}",

198 service.name(),

199 String::from_utf8_lossy(text.as_slice()).trim()

200 )))

201 }

169202 None => Err(SmartError::Internal(format!(

170203 "git {} timed out after {timeout_secs}s",

171204 service.name()

@@ -208,5 +241,10 @@

208241 if let Some(rx) = rx {

209242 let _ = rx.await;

210243 }

244 tracing::debug!(

245 svc = service.name(),

246 output_bytes = out.len(),

247 "git rpc output collected"

248 );

211249 Ok(Bytes::from(out))

212250}

Modified lib/git/src/ssh/handler/channel.rs +173 −17 View lib/git/src/ssh/handler/channel.rs

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

1212use crate::smart::pre_receive::parse_receive_pack_refs;

1313use crate::ssh::error::SshError;

1414use crate::ssh::git_service::apply_hardened_env;

15use crate::ssh::handler::GitSshHandler;

15use crate::ssh::handler::{ChannelState, GitSshHandler};

1616use crate::sync::post_receive::enqueue_post_receive_push;

1717

1818pub const MAX_SSH_CHANNEL_BUF: usize = 64 * 1024 * 1024;

1919pub const MAX_GIT_OUTPUT_BUF: usize = 256 * 1024 * 1024;

2020

21fn hex_preview(buf: &[u8]) -> String {

22 let n = buf.len().min(96);

23 let mut s = String::with_capacity(n * 2);

24 for b in &buf[..n] {

25 s.push_str(&format!("{b:02x}"));

26 }

27 s

28}

29

2130impl GitSshHandler {

2231 pub async fn data_impl(

2332 &mut self,

@@ -45,10 +54,62 @@

4554 return Ok(());

4655 }

4756 state.buffer.extend_from_slice(data);

57 if state.git_op.is_some() && state.buffer.len() == data.len() {

58 tracing::debug!(

59 ?channel,

60 first_hex = %hex_preview(&state.buffer),

61 "ssh: first data chunk on armed channel"

62 );

63 }

64 let is_receive = matches!(

65 state.git_op,

66 Some(crate::ssh::handler::git_op::GitOperation::ReceivePack { .. })

67 );

68 if !is_receive {

69 return Ok(());

70 }

71 if !state.nak_sent && Self::wants_section_complete(&state.buffer) {

72 // Over SSH the request and response share one stream: send-pack

73 // stops after the wants section and waits for NAK before it sends

74 // the pack. We only spawn receive-pack after channel EOF, so

75 // synthesize the negotiation reply now; the real process' own NAK

76 // is stripped from stdout later (see run_git).

77 tracing::debug!(?channel, "ssh push: synthesized NAK for wants section");

78 state.nak_sent = true;

79 let _ = session.data(channel, Bytes::from_static(b"0008NAK\n0000"));

80 let _ = session.flush();

81 }

4882 Ok(())

4983 }

5084

85 /// True when the buffer head is a complete `want` negotiation section

86 /// terminated by a flush packet.

87 fn wants_section_complete(buf: &[u8]) -> bool {

88 let mut i = 0usize;

89 let mut saw_want = false;

90 while i + 4 <= buf.len() {

91 let Ok(len) =

92 usize::from_str_radix(std::str::from_utf8(&buf[i..i + 4]).unwrap_or_default(), 16)

93 else {

94 return false;

95 };

96 if len == 0 {

97 return saw_want;

98 }

99 if len < 4 || i + len > buf.len() {

100 return false;

101 }

102 if !buf[i + 4..i + len].starts_with(b"want ") {

103 return false;

104 }

105 saw_want = true;

106 i += len;

107 }

108 false

109 }

110

51111 pub async fn channel_close_impl(&mut self, channel: ChannelId) {

112 tracing::debug!(?channel, "ssh channel closed");

52113 self.channels.remove(&channel);

53114 }

54115

@@ -57,23 +118,47 @@

57118 channel: ChannelId,

58119 session: &mut Session,

59120 ) -> Result<(), russh::Error> {

60 let Some(mut state) = self.channels.remove(&channel) else {

121 let Some(state) = self.channels.remove(&channel) else {

61122 let _ = session.close(channel);

62123 let _ = session.flush();

63124 return Ok(());

64125 };

126 self.dispatch_state(channel, state, session.handle(), "eof");

127 Ok(())

128 }

129

130 /// Detach buffered channel state and run its git operation in the

131 /// background. `reason` is only used for tracing.

132 fn dispatch_state(

133 &mut self,

134 channel: ChannelId,

135 state: ChannelState,

136 handle: russh::server::Handle,

137 reason: &str,

138 ) {

139 let mut state = state;

65140 let overflowed = state.buffer_overflow;

141 let nak_sent = state.nak_sent;

66142 let op = state.git_op.take();

67143 let buffer = std::mem::take(&mut state.buffer);

68144 let repo_id = state.repo_id;

69145 let space_id = state.space_id;

70146 let actor_id = state.actor_id;

147 tracing::debug!(

148 ?channel,

149 op = op.as_ref().map(|o| o.service_arg()),

150 buffered_bytes = buffer.len(),

151 repo_id = ?repo_id,

152 actor_id = ?actor_id,

153 overflowed,

154 %reason,

155 "ssh dispatching git operation"

156 );

71157 let push_coordinator = self.service.push_coordinator();

72158 let queue = self.service.smart.queue.clone();

73159 let sync_settings = self.service.sync_settings.clone();

74160 let ssh_service = self.service.clone();

75161

76 let handle = session.handle();

77162 let db = self.service.smart.git.state.database.clone();

78163 let rpc_timeout = self.service.smart.git.state.config.git.git_rpc_timeout_secs;

79164 tokio::spawn(async move {

@@ -158,6 +243,7 @@

158243 _ = handle.close(channel);

159244 return;

160245 }

246 tracing::debug!(repo_id = %repo_id, "ssh pre-receive validation passed");

161247 }

162248 Ok(_) => {}

163249 Err(e) => {

@@ -184,15 +270,28 @@

184270 tracing::warn!(?e, "post-receive hook setup failed");

185271 }

186272 let push_key = format!("git_op:{}", bare.display());

273 tracing::debug!(push_key = %push_key, repo_id = ?repo_id, "ssh push: acquiring push lock");

187274

188 let run =

189 async { run_git(op.as_ref(), &buffer, &handle, channel, rpc_timeout).await };

275 let run = async {

276 run_git(

277 op.as_ref(),

278 &buffer,

279 &handle,

280 channel,

281 rpc_timeout,

282 nak_sent,

283 )

284 .await

285 };

190286 let result = push_coordinator

191287 .run(&push_key, async {

192288 run.await.map_err(|e| SmartError::Internal(e.to_string()))

193289 })

194290 .await

195291 .map_err(|e: SmartError| SshError::from(e));

292 if let Err(e) = &result {

293 tracing::warn!(repo_id = ?repo_id, error = %e, "ssh push failed");

294 }

196295 if result.is_ok()

197296 && let (Some(repo_id), Some(space_id), Some(queue)) =

198297 (repo_id, space_id, &queue)

@@ -229,21 +328,37 @@

229328 }

230329 }

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

232 _ = handle.exit_status_request(channel, code);

233 tokio::time::sleep(std::time::Duration::from_millis(20)).await;

234 _ = handle.eof(channel);

235 _ = handle.close(channel);

331 tracing::debug!(

332 ?channel,

333 code,

334 nak_sent,

335 "ssh: sending exit-status for push"

336 );

337 let r1 = handle.exit_status_request(channel, code).await;

338 // Windows OpenSSH can drop buffered channel data if CLOSE

339 // races ahead of the client's read loop; give the client time

340 // to drain the report before tearing the channel down.

341 tokio::time::sleep(std::time::Duration::from_millis(250)).await;

342 let r2 = handle.eof(channel).await;

343 let r3 = handle.close(channel).await;

344 tracing::debug!(exit_status = ?r1, eof = ?r2, close = ?r3, "ssh: channel shutdown results");

236345 } else {

237 let result = run_git(op.as_ref(), &buffer, &handle, channel, rpc_timeout).await;

346 let result = run_git(

347 op.as_ref(),

348 &buffer,

349 &handle,

350 channel,

351 rpc_timeout,

352 nak_sent,

353 )

354 .await;

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

239356 _ = handle.exit_status_request(channel, code);

240 tokio::time::sleep(std::time::Duration::from_millis(20)).await;

357 tokio::time::sleep(std::time::Duration::from_millis(250)).await;

241358 _ = handle.eof(channel);

242359 _ = handle.close(channel);

243360 }

244361 });

245

246 Ok(())

247362 }

248363}

249364

@@ -253,6 +368,7 @@

253368 handle: &russh::server::Handle,

254369 channel: russh::ChannelId,

255370 timeout_secs: u64,

371 strip_nak: bool,

256372) -> Result<(), SshError> {

257373 let Some(op) = op else {

258374 return Ok(());

@@ -263,6 +379,9 @@

263379 if !op.is_upload_archive() {

264380 cmd.arg("--stateless-rpc");

265381 }

382 // Same as the advertise path: receive-pack requires an explicit

383 // <git-dir> argument, GIT_DIR alone is not enough.

384 cmd.arg(op.bare());

266385 cmd.env("GIT_DIR", op.bare())

267386 .env("GIT_TERMINAL_PROMPT", "0")

268387 .stdin(std::process::Stdio::piped())

@@ -271,6 +390,13 @@

271390 apply_hardened_env(&mut cmd);

272391

273392 let mut child = cmd.spawn().map_err(SshError::Io)?;

393 tracing::debug!(

394 service = op.service_arg(),

395 bare = %op.bare().display(),

396 input_bytes = input.len(),

397 timeout_secs,

398 "spawned git process for ssh operation"

399 );

274400 let mut stdin = child

275401 .stdin

276402 .take()

@@ -354,13 +480,43 @@

354480 )));

355481 }

356482 stdout_read.map_err(SshError::Io)?;

483 let mut stdout_buf = stdout_buf;

484 if strip_nak {

485 // We already answered the wants section with a synthetic NAK, so

486 // remove receive-pack's own NAK+flush to keep the client's

487 // negotiation read aligned with the report that follows.

488 const NAK: &[u8] = b"0008NAK\n";

489 if stdout_buf.starts_with(NAK) {

490 let mut off = NAK.len();

491 if stdout_buf.get(off..off + 4) == Some(b"0000") {

492 off += 4;

493 }

494 stdout_buf.drain(..off);

495 }

496 }

497 tracing::debug!(

498 service = op.service_arg(),

499 exit = ?status.code(),

500 stdout_bytes = stdout_buf.len(),

501 stderr_bytes = stderr_buf.len(),

502 "git process for ssh operation finished"

503 );

357504 if !stdout_buf.is_empty() {

358 let _ = handle.data(channel, Bytes::from(stdout_buf)).await;

505 if let Err(e) = handle.data(channel, Bytes::from(stdout_buf.clone())).await {

506 tracing::warn!(

507 ?e,

508 bytes = stdout_buf.len(),

509 "ssh: failed to send git stdout to client"

510 );

511 }

359512 }

360513 if !stderr_buf.is_empty() && !status.success() {

361 let _ = handle

362 .extended_data(channel, 1, Bytes::from(stderr_buf))

363 .await;

514 if let Err(e) = handle

515 .extended_data(channel, 1, Bytes::from(stderr_buf.clone()))

516 .await

517 {

518 tracing::warn!(?e, "ssh: failed to send git stderr to client");

519 }

364520 }

365521 if status.success() {

366522 Ok(())

Modified lib/git/src/ssh/handler/command.rs +29 −0 View lib/git/src/ssh/handler/command.rs

@@ -40,9 +40,16 @@

4040 return Ok(());

4141 };

4242 let Some((space_name, repo_name)) = parse_repo_path(raw_path) else {

43 tracing::debug!(path = %sanitize_for_log(raw_path), "ssh exec: invalid repository path");

4344 invalid_command("invalid repository path", session, channel).await;

4445 return Ok(());

4546 };

47 tracing::debug!(

48 service = service.as_subcommand(),

49 space = space_name,

50 repo = repo_name,

51 "ssh exec: parsed git command"

52 );

4653

4754 let actor_id = self.operator.as_ref().map(|u| u.uid);

4855 let required = if service.is_write() { "write" } else { "read" };

@@ -59,11 +66,18 @@

5966 let ctx = match self.service.smart.resolve(space_name, repo_name).await {

6067 Ok(c) => c,

6168 Err(e) => {

69 tracing::warn!(space = space_name, repo = repo_name, error = %e, "ssh exec: repo resolve failed");

6270 let msg = format!("fatal: {e}");

6371 invalid_command(&msg, session, channel).await;

6472 return Ok(());

6573 }

6674 };

75 tracing::debug!(

76 space = space_name,

77 repo = repo_name,

78 repo_id = %ctx.model.id,

79 "ssh exec: repo access authorized and resolved"

80 );

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

6882 Ok(id) => id,

6983 Err(error) => {

@@ -74,6 +88,7 @@

7488

7589 let bare = ctx.repo_dir();

7690 if !bare.is_dir() {

91 tracing::warn!(bare = %bare.display(), "ssh exec: bare repository directory missing");

7792 let msg = format!("fatal: repository '{space_name}/{repo_name}' does not exist");

7893 invalid_command(&msg, session, channel).await;

7994 return Ok(());

@@ -88,16 +103,29 @@

88103 {

89104 Ok(adv) => adv,

90105 Err(e) => {

106 tracing::warn!(

107 service = service.as_subcommand(),

108 bare = %bare.display(),

109 error = %e,

110 "ssh exec: ref advertisement failed"

111 );

91112 let msg = format!("fatal: unable to read repository advertisement: {e}");

92113 invalid_command(&msg, session, channel).await;

93114 return Ok(());

94115 }

95116 };

117 let adv_len = adv.len();

96118 if !adv.is_empty() {

97119 let _ = session.data(channel, Bytes::from(adv));

98120 }

99121 let _ = session.channel_success(channel);

100122 let _ = session.flush();

123 tracing::debug!(

124 service = service.as_subcommand(),

125 repo_id = %ctx.model.id,

126 adv_bytes = adv_len,

127 "ssh exec: advertisement sent, channel armed"

128 );

101129

102130 let op = match service {

103131 GitService::UploadPack => GitOperation::UploadPack { bare: bare.clone() },

@@ -148,6 +176,7 @@

148176 channel: ChannelId,

149177 session: &mut Session,

150178 ) -> Result<(), russh::Error> {

179 tracing::debug!(client = ?self.client_addr, "interactive shell request denied (git-only server)");

151180 let _ = session.data(

152181 channel,

153182 Bytes::from_static(

Modified lib/git/src/ssh/handler/mod.rs +7 −3 View lib/git/src/ssh/handler/mod.rs

@@ -26,6 +26,7 @@

2626 pub space_id: Option<Uuid>,

2727 pub actor_id: Option<Uuid>,

2828 pub buffer_overflow: bool,

29 pub nak_sent: bool,

2930}

3031

3132pub struct GitSshHandler {

@@ -62,11 +63,13 @@

6263impl Handler for GitSshHandler {

6364 type Error = russh::Error;

6465

65 async fn auth_none(&mut self, _user: &str) -> Result<Auth, Self::Error> {

66 async fn auth_none(&mut self, user: &str) -> Result<Auth, Self::Error> {

67 tracing::debug!(user, client = ?self.client_addr, "ssh auth_none rejected");

6668 Ok(Auth::reject())

6769 }

6870

69 async fn auth_password(&mut self, _user: &str, _password: &str) -> Result<Auth, Self::Error> {

71 async fn auth_password(&mut self, user: &str, _password: &str) -> Result<Auth, Self::Error> {

72 tracing::debug!(user, client = ?self.client_addr, "ssh password auth rejected (publickey only)");

7073 Ok(Auth::reject())

7174 }

7275

@@ -88,9 +91,10 @@

8891

8992 async fn auth_openssh_certificate(

9093 &mut self,

91 _user: &str,

94 user: &str,

9295 _certificate: &russh::keys::Certificate,

9396 ) -> Result<Auth, Self::Error> {

97 tracing::debug!(user, "ssh certificate auth rejected");

9498 Ok(Auth::reject())

9599 }

96100

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

@@ -40,7 +40,15 @@

4040 .map(|b| (b.name.clone(), b))

4141 .collect();

4242

43 tracing::debug!(

44 repo_id = %repo.id,

45 tips = collected.len(),

46 db_branches = existing_by_name.len(),

47 "syncing branches"

48 );

49

4350 let mut stale: std::collections::HashSet<String> = existing_by_name.keys().cloned().collect();

51 let (mut updated, mut inserted) = (0usize, 0usize);

4452

4553 for tip in &collected {

4654 stale.remove(&tip.name);

@@ -50,6 +58,7 @@

5058 am.commit_sha = Set(tip.target_sha.clone());

5159 am.updated_at = Set(OffsetDateTime::now_utc());

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

61 updated += 1;

5362 }

5463 Some(_) => {}

5564 None => {

@@ -68,12 +77,15 @@

6877 .insert(txn)

6978 .await

7079 .map_err(SyncError::Db)?;

80 inserted += 1;

7181 }

7282 }

7383 }

7484

85 let mut deleted = 0usize;

7586 for name in stale {

7687 if existing_by_name.get(&name).is_some_and(|r| r.is_protected) {

88 tracing::debug!(branch = %name, "stale branch is protected; keeping row");

7789 continue;

7890 }

7991 let Some(row) = existing_by_name.remove(&name) else {

@@ -84,7 +96,15 @@

8496 am.deleted_at = Set(Some(OffsetDateTime::now_utc()));

8597 am.updated_at = Set(OffsetDateTime::now_utc());

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

99 deleted += 1;

87100 }

101 tracing::debug!(

102 repo_id = %repo.id,

103 updated,

104 inserted,

105 deleted,

106 "branch sync wrote changes"

107 );

88108

89109 reconcile_default_branch(baby, txn, repo, &collected).await?;

90110

@@ -128,6 +148,12 @@

128148 };

129149

130150 if desired != repo.default_branch {

151 tracing::info!(

152 repo_id = %repo.id,

153 old = %repo.default_branch,

154 new = %desired,

155 "default branch changed during sync"

156 );

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

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

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

Modified lib/git/src/sync/commits.rs +13 −2 View lib/git/src/sync/commits.rs

@@ -40,6 +40,11 @@

4040

4141 for branch in branches {

4242 if out.len() >= max_total {

43 tracing::debug!(

44 new_commits = out.len(),

45 max_total,

46 "commit collection hit max_commits_per_sync cap"

47 );

4348 break;

4449 }

4550 let remaining = max_total - out.len();

@@ -169,8 +174,14 @@

169174 .await

170175 .map_err(SyncError::Db)?;

171176

172 Ok(rows

177 let map: HashMap<String, Uuid> = rows

173178 .into_iter()

174179 .map(|r| (r.lower_email, r.user_id))

175 .collect())

180 .collect();

181 tracing::debug!(

182 emails = lower_emails.len(),

183 matched = map.len(),

184 "resolved commit author emails to users"

185 );

186 Ok(map)

176187}

Modified lib/git/src/sync/languages.rs +20 −7 View lib/git/src/sync/languages.rs

@@ -85,6 +85,12 @@

8585 if let Some(primary) = desired.first()

8686 && repo.primary_language.as_deref() != Some(primary.language.as_str())

8787 {

88 tracing::debug!(

89 repo_id = %repo.id,

90 old = ?repo.primary_language,

91 new = %primary.language,

92 "primary language changed during sync"

93 );

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

8995 am.primary_language = Set(Some(primary.language.clone()));

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

@@ -110,15 +116,16 @@

110116 return Ok(None);

111117 };

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

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

119 let Ok(stats) =

120 tokio::time::timeout(timeout, walk_lang_stats(baby.clone(), revision.clone())).await

121 else {

122 tracing::warn!(revision = %revision, "language detection timed out");

123 return Ok(None);

124 };

114125

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

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

117 Ok(Err(e)) => return Err(e),

118 Err(_) => {

119 tracing::warn!("language detection timed out");

120 return Ok(None);

121 }

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

128 Err(e) => return Err(e),

122129 };

123130

124131 let mut entries = entries;

@@ -142,6 +149,12 @@

142149 })

143150 .collect();

144151

152 tracing::debug!(

153 repo_id = %repo.id,

154 revision = %revision,

155 languages = out.len(),

156 "language detection finished"

157 );

145158 Ok(Some(out))

146159}

147160

Modified lib/git/src/sync/license.rs +10 −1 View lib/git/src/sync/license.rs

@@ -23,6 +23,12 @@

2323 return Ok(());

2424 };

2525 let body = read_license_body(repo_dir).await?;

26 tracing::debug!(

27 repo_id = %repo.id,

28 license = %name,

29 body_found = body.is_some(),

30 "upserting license record"

31 );

2632 let display_name = infer_display_name(&name, body.as_deref());

2733 let description = infer_description(&name, body.as_deref());

2834 let url = infer_url(&name);

@@ -81,7 +87,10 @@

8187 }

8288 }

8389 Err(error) if error.kind() == std::io::ErrorKind::NotFound => continue,

84 Err(error) => return Err(SyncError::Io(error)),

90 Err(error) => {

91 tracing::warn!(path = %path.display(), error = %error, "failed to read license file");

92 return Err(SyncError::Io(error));

93 }

8594 }

8695 }

8796 Ok(None)

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

@@ -50,6 +50,9 @@

5050 .query_async(c)

5151 .await

5252 })?;

53 if n != 1 {

54 tracing::warn!(key = %self.key, "sync lock heartbeat refresh lost ownership");

55 }

5356 Ok(n == 1)

5457 }

5558

@@ -64,6 +67,7 @@

6467 .query_async(c)

6568 .await

6669 })?;

70 tracing::debug!(key = %self.key, "sync lock released");

6771 Ok(())

6872 }

6973}

@@ -93,9 +97,11 @@

9397 })?;

9498

9599 if !acquired {

100 tracing::debug!(key = %key, repo_id = %repo_id, "sync lock busy; another sync in progress");

96101 return Ok(None);

97102 }

98103

104 tracing::debug!(key = %key, repo_id = %repo_id, ttl_secs, "sync lock acquired");

99105 Ok(Some(SyncLock {

100106 redis: redis.clone(),

101107 key,

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

@@ -36,6 +36,7 @@

3636 let conn = open_multiplexed(&redis).await?;

3737 consumer_conns.push(conn);

3838 }

39 tracing::debug!(count, "sync queue consumer connections opened");

3940 Ok(Self {

4041 redis,

4142 consumer_conns,

@@ -97,6 +98,12 @@

9798 .query_async(c)

9899 .await

99100 })?;

101 tracing::debug!(

102 topic = key,

103 queue_len = n,

104 payload_bytes = payload.len(),

105 "sync queue message pushed"

106 );

100107 Ok(n)

101108 }

102109

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

@@ -39,5 +39,14 @@

3939 am.updated_at = Set(now);

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

4141

42 tracing::debug!(

43 repo_id = %repo.id,

44 git_size,

45 total_commits,

46 branches = branch_count,

47 tags = tag_count,

48 new_commits,

49 "repo stats updated"

50 );

4251 Ok(())

4352}

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

@@ -42,7 +42,15 @@

4242 let mut existing_by_name: HashMap<String, git_tags::Model> =

4343 existing.into_iter().map(|t| (t.name.clone(), t)).collect();

4444

45 tracing::debug!(

46 repo_id = %repo.id,

47 tips = collected.len(),

48 db_tags = existing_by_name.len(),

49 "syncing tags"

50 );

51

4552 let mut stale: std::collections::HashSet<String> = existing_by_name.keys().cloned().collect();

53 let (mut updated, mut inserted) = (0usize, 0usize);

4654

4755 for tip in &collected {

4856 stale.remove(&tip.name);

@@ -59,6 +67,7 @@

5967 am.tagger_name = Set(tip.tagger_name.clone());

6068 am.tagger_email = Set(tip.tagger_email.clone());

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

70 updated += 1;

6271 }

6372 Some(_) => {}

6473 None => {

@@ -76,10 +85,12 @@

7685 .insert(txn)

7786 .await

7887 .map_err(SyncError::Db)?;

88 inserted += 1;

7989 }

8090 }

8191 }

8292

93 let mut deleted = 0usize;

8394 for name in stale {

8495 let Some(row) = existing_by_name.remove(&name) else {

8596 continue;

@@ -88,7 +99,15 @@

8899 .exec(txn)

89100 .await

90101 .map_err(SyncError::Db)?;

102 deleted += 1;

91103 }

104 tracing::debug!(

105 repo_id = %repo.id,

106 updated,

107 inserted,

108 deleted,

109 "tag sync wrote changes"

110 );

92111

93112 Ok(collected.len())

94113}