refactor(gitsmart): refresh crate across handler, protocol, push, tests

Modified lib/gitsmart/handler/dumb.rs +0 −4 View lib/gitsmart/handler/dumb.rs

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

66use crate::error::{SmartError, SmartResult};

77use crate::service::{ResolvedRepo, SmartService};

88

9/// Dumb HTTP transport: serve bare-repository files over plain HTTP.

10/// Request paths are `info/refs`, `HEAD`, `objects/<sha>`, `info/packs`, ...

119pub async fn handle(

1210 state: &SmartService,

1311 space: &str,

@@ -55,8 +53,6 @@

5553 }

5654}

5755

58/// Resolve a request sub-path against the bare root, rejecting any traversal

59/// that escapes the repository directory.

6056fn resolve_within_root(root: &Path, request_path: &str) -> SmartResult<PathBuf> {

6157 let mut base = root.to_path_buf();

6258 for component in Path::new(request_path).components() {

Modified lib/gitsmart/handler/lfs.rs +0 −13 View lib/gitsmart/handler/lfs.rs

@@ -15,7 +15,6 @@

1515use crate::service::{ResolvedRepo, SmartService};

1616

1717const MAX_LFS_BODY: usize = 8 * 1024 * 1024 * 1024;

18/// Per-chunk buffer used when bridging axum `Body` to a `BoxStream`.

1918const STREAM_BUF_LEN: usize = 64 * 1024;

2019

2120pub async fn dispatch(

@@ -189,9 +188,6 @@

189188 }

190189}

191190

192/// Streaming PUT: pipe the request body directly into the LFS store's

193/// `put_object_stream` so peak memory is bounded to a single chunk (64 KiB)

194/// regardless of object size, instead of buffering the full 8 GiB body.

195191async fn upload(

196192 state: &SmartService,

197193 resolved: &ResolvedRepo,

@@ -202,15 +198,12 @@

202198 let repo = state.git_repository(resolved)?;

203199 let is_gzip = is_gzip(headers);

204200

205 // axum Body -> futures stream of Result<Bytes, LfsError>; map IO errors to LfsError::Io.

206201 let raw_stream: futures::stream::BoxStream<'static, Result<Bytes, LfsError>> = Box::pin(

207202 body.into_data_stream()

208203 .map(|item| item.map_err(|e| LfsError::Io(e.to_string()))),

209204 );

210205

211206 let stream: futures::stream::BoxStream<'static, Result<Bytes, LfsError>> = if is_gzip {

212 // StreamReader / GzipDecoder / ReaderStream operate on std::io::Error, so we

213 // bridge LfsError -> io::Error on the way in and back on the way out.

214207 let io_stream = raw_stream.map_err(|e| match e {

215208 LfsError::Storage(s) => std::io::Error::other(s.to_string()),

216209 other => std::io::Error::other(other.to_string()),

@@ -228,8 +221,6 @@

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

229222}

230223

231/// Streaming GET: build the response body directly from the store's

232/// `get_object_stream`. No buffering of the full object in memory.

233224async fn download(

234225 state: &SmartService,

235226 resolved: &ResolvedRepo,

@@ -254,10 +245,6 @@

254245 .unwrap_or(false)

255246}

256247

257/// Wrap a `BoxStream<Result<Bytes, LfsError>>` so that the cumulative byte

258/// count is tracked; once `limit` bytes have been seen, inject a

259/// `PayloadTooLarge` error into the stream to abort the store's multipart

260/// upload with a clear reason.

261248fn enforce_limit(

262249 inner: futures::stream::BoxStream<'static, Result<Bytes, LfsError>>,

263250 limit: usize,

Modified lib/gitsmart/handler/mod.rs +31 −7 View lib/gitsmart/handler/mod.rs

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

1616use std::io::Read;

1717

1818use crate::error::{SmartError, SmartResult};

19use crate::metrics;

1920use crate::service::{SmartService, SmartState};

2021

2122pub fn router(state: SmartState) -> Router {

@@ -43,21 +44,25 @@

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

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

4546 _ => {

47 metrics::record_http("other", false);

4648 return respond_error(SmartError::BadRequest(

4749 "expected '<repo>.git' in URL".into(),

4850 ));

4951 }

5052 };

5153

54 let mut endpoint: &'static str = "other";

5255 let result: SmartResult<Response> = async {

5356 let path_lower = path.to_lowercase();

5457 let is_get = method == Method::GET;

5558 match (path_lower.as_str(), is_get) {

5659 ("info/refs", true) => {

60 endpoint = "info_refs";

5761 let service = query.get("service").map(String::as_str);

5862 smart::info_refs(&state, &space, &repo_name, &headers, service).await

5963 }

6064 ("git-upload-pack", false) => {

65 endpoint = "upload_pack";

6166 smart::rpc(

6267 &state,

6368 &space,

@@ -69,6 +74,7 @@

6974 .await

7075 }

7176 ("git-receive-pack", false) => {

77 endpoint = "receive_pack";

7278 smart::rpc(

7379 &state,

7480 &space,

@@ -80,6 +86,7 @@

8086 .await

8187 }

8288 (p, _) if p.starts_with("info/lfs/") => {

89 endpoint = lfs_endpoint(&path_lower, &method);

8390 lfs::dispatch(

8491 &state,

8592 &space,

@@ -91,15 +98,37 @@

9198 )

9299 .await

93100 }

94 (_, true) => dumb::handle(&state, &space, &repo_name, &headers, &path_lower).await,

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

101 (_, true) => {

102 endpoint = "dumb";

103 dumb::handle(&state, &space, &repo_name, &headers, &path_lower).await

104 }

105 (_, _) => {

106 endpoint = "other";

107 Ok(SmartError::BadRequest("method not allowed".into()).into_response())

108 }

96109 }

97110 }

98111 .await;

99112

113 metrics::record_http(endpoint, result.is_ok());

100114 result.unwrap_or_else(respond_error)

101115}

102116

117fn lfs_endpoint(path: &str, method: &Method) -> &'static str {

118 match path {

119 "info/lfs/objects/batch" => "lfs_batch",

120 "info/lfs/verify" => "lfs_verify",

121 p if p.starts_with("info/lfs/objects/") => {

122 if *method == Method::PUT {

123 "lfs_upload"

124 } else {

125 "lfs_download"

126 }

127 }

128 _ => "lfs_other",

129 }

130}

131

103132fn respond_error(err: SmartError) -> Response {

104133 tracing::warn!(error = %err, "gitsmart request failed");

105134 if err.status() == StatusCode::INTERNAL_SERVER_ERROR {

@@ -118,7 +147,6 @@

118147}

119148

120149fn url_decode(s: &str) -> String {

121 // Minimal percent-decoding for the service query parameter.

122150 let mut out = String::with_capacity(s.len());

123151 let bytes = s.as_bytes();

124152 let mut i = 0;

@@ -136,8 +164,6 @@

136164 out

137165}

138166

139/// Authenticate the request. Returns the acting user (anonymous when no

140/// Authorization header is present).

141167pub(crate) async fn authenticate(

142168 state: &SmartService,

143169 headers: &HeaderMap,

@@ -156,7 +182,6 @@

156182 Ok(user.actor)

157183}

158184

159/// Authorize `required` access, returning the resolved repo.

160185pub(crate) async fn authorize(

161186 state: &SmartService,

162187 space: &str,

@@ -171,7 +196,6 @@

171196 .map_err(SmartError::from)

172197}

173198

174/// Decompress a request body when `Content-Encoding: gzip` is present.

175199pub(crate) async fn decode_request_body(

176200 headers: &HeaderMap,

177201 body: Body,

Modified lib/gitsmart/handler/smart.rs +25 −3 View lib/gitsmart/handler/smart.rs

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

66use crate::error::{SmartError, SmartResult};

77use crate::handler::dumb;

88use crate::handler::{authenticate, authorize, decode_request_body};

9use crate::metrics;

910use crate::protocol::{GitService, pktline, rpc};

1011use crate::service::{ResolvedRepo, SmartService};

1112

@@ -73,11 +74,35 @@

7374) -> SmartResult<Response> {

7475 let actor = authenticate(state, headers).await?;

7576 let resolved = authorize(state, space, repo, &actor, service.required_access()).await?;

77 let actor_id = actor.user_id;

7678

7779 let run = async {

80 let _timer = metrics::RpcTimer::start();

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

7982 let bare = state.bare_path(&resolved);

83 let is_receive = service == GitService::ReceivePack;

84 if is_receive && let Err(e) = gitsync::prepare_post_receive(&bare) {

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

86 }

8087 let output = rpc::run_rpc_bytes(&bare, service, false, input).await?;

88 if is_receive && let Some(queue) = &state.queue {

89 match gitsync::enqueue_post_receive_push(

90 queue,

91 &state.sync_settings,

92 resolved.channel.id,

93 resolved.repo.id,

94 actor_id,

95 &bare,

96 )

97 .await

98 {

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

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

101 }

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

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

104 }

105 }

81106 Ok::<_, SmartError>(

82107 (

83108 StatusCode::OK,

@@ -92,9 +117,6 @@

92117 };

93118

94119 match service {

95 // Pushes to the same repository are serialized through a Redis

96 // push lock + wait queue so concurrent receive-pack RPCs never

97 // interleave on one bare repository.

98120 GitService::ReceivePack => {

99121 let key = format!("push:{}:{}", resolved.channel.id, resolved.repo.id);

100122 state.push_coordinator().run(&key, run).await

Modified lib/gitsmart/protocol/rpc.rs +0 −7 View lib/gitsmart/protocol/rpc.rs

@@ -24,12 +24,6 @@

2424 done: bool,

2525}

2626

27/// Run `git <service> --stateless-rpc [--advertise-refs]` against a bare repo,

28/// streaming the request body to stdin and the response body from stdout.

29///

30/// The child is spawned with `GIT_DIR`, `GIT_PROTOCOL=version=2` and

31/// `GIT_TERMINAL_PROMPT=0`. If the child exits non-zero, the final stream item

32/// carries an error that includes its stderr.

3327pub async fn run_rpc(

3428 bare_path: &Path,

3529 service: GitService,

@@ -126,7 +120,6 @@

126120 Ok(Box::pin(stream))

127121}

128122

129/// Buffered convenience wrapper used by tests: sends the whole body, collects stdout.

130123pub async fn run_rpc_bytes(

131124 bare_path: &Path,

132125 service: GitService,

Modified lib/gitsmart/protocol/service.rs +1 −2 View lib/gitsmart/protocol/service.rs

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

2626 }

2727 }

2828

29 /// Dashed binary name used in the smart-HTTP `# service=` pkt-line header.

30 pub fn dashed_name(self) -> &'static str {

29 pub fn dashed_name(self) -> &'static str {

3130 match self {

3231 Self::UploadPack => "git-upload-pack",

3332 Self::ReceivePack => "git-receive-pack",

Modified lib/gitsmart/tests/smart.rs +0 −15 View lib/gitsmart/tests/smart.rs

@@ -517,10 +517,7 @@

517517 );

518518}

519519

520// === Streaming LFS tests ===

521520

522/// Generate a deterministic pseudo-random byte vector of `len` bytes,

523/// seeded by the supplied value so callers can produce distinct payloads.

524521fn pseudo_random_bytes(len: usize, seed: u64) -> Vec<u8> {

525522 let mut out = Vec::with_capacity(len);

526523 let mut state: u64 = seed;

@@ -533,7 +530,6 @@

533530 out

534531}

535532

536/// Build a payload that gzip can actually compress (low-entropy text-like data).

537533fn compressible_payload(len: usize) -> Vec<u8> {

538534 let pattern = b"the quick brown fox jumps over the lazy dog\n";

539535 let mut out = Vec::with_capacity(len);

@@ -544,7 +540,6 @@

544540 out

545541}

546542

547/// Compress `data` with gzip using flate2 (already in workspace deps).

548543fn gzip_compress(data: &[u8]) -> Vec<u8> {

549544 use flate2::Compression;

550545 use flate2::write::GzEncoder;

@@ -559,9 +554,6 @@

559554 let srv = start_server().await;

560555 let client = reqwest::Client::new();

561556

562 // 5 MiB — large enough to span multiple axum chunks (64 KiB each) and

563 // multiple storage multipart parts (5 MiB each). Peak memory on the

564 // server must remain bounded to one chunk + one part.

565557 let data = pseudo_random_bytes(5 * 1024 * 1024, 0xA1B2_C3D4_E5F6_0701);

566558 let oid = gitlfs::LfsOid::compute(&data);

567559 let url = format!("{}/info/lfs/objects/{}", srv.base, oid.as_str());

@@ -576,7 +568,6 @@

576568 .expect("lfs put");

577569 assert_eq!(put.status(), reqwest::StatusCode::OK, "lfs put failed");

578570

579 // Round-trip: GET and verify sha256 + byte-equal content.

580571 let got = client

581572 .get(&url)

582573 .header("Authorization", basic_auth("alice", "secret"))

@@ -630,7 +621,6 @@

630621 let srv = start_server().await;

631622 let client = reqwest::Client::new();

632623

633 // Compute OID for one payload, but PUT a different one.

634624 let real = pseudo_random_bytes(64 * 1024, 0x1111_1111_1111_1111);

635625 let fake = pseudo_random_bytes(64 * 1024, 0x2222_2222_2222_2222);

636626 let oid = gitlfs::LfsOid::compute(&real);

@@ -645,14 +635,12 @@

645635 .send()

646636 .await

647637 .expect("lfs put");

648 // Store auto-aborts multipart on hash mismatch; handler surfaces 4xx.

649638 assert!(

650639 put.status().is_client_error() || put.status().is_server_error(),

651640 "expected error status, got {}",

652641 put.status()

653642 );

654643

655 // Object must not have been stored.

656644 let head = client

657645 .get(&url)

658646 .header("Authorization", basic_auth("alice", "secret"))

@@ -671,7 +659,6 @@

671659 let oid = gitlfs::LfsOid::compute(&data);

672660 let url = format!("{}/info/lfs/objects/{}", srv.base, oid.as_str());

673661

674 // Seed the store.

675662 let put = client

676663 .put(&url)

677664 .header("Authorization", basic_auth("alice", "secret"))

@@ -681,8 +668,6 @@

681668 .expect("seed put");

682669 assert_eq!(put.status(), reqwest::StatusCode::OK);

683670

684 // Stream-download and verify, streaming the response body (no full

685 // buffering on client side either).

686671 let resp = client

687672 .get(&url)

688673 .header("Authorization", basic_auth("alice", "secret"))

Modified lib/gitsmart/lib.rs +1 −0 View lib/gitsmart/lib.rs

@@ -1,6 +1,7 @@

11pub mod config;

22pub mod error;

33pub mod handler;

4pub mod metrics;

45pub mod protocol;

56pub mod push;

67pub mod service;

Added lib/gitsmart/metrics.rs +63 −0 View lib/gitsmart/metrics.rs

@@ -0,0 +1,63 @@

1use std::sync::Once;

2use std::time::Instant;

3

4use metrics::{Unit, counter, describe_counter, describe_histogram, histogram};

5

6const OUTCOME_SUCCESS: &str = "success";

7const OUTCOME_ERROR: &str = "error";

8

9static DESCRIBE_ONCE: Once = Once::new();

10

11pub(crate) fn ensure_described() {

12 DESCRIBE_ONCE.call_once(describe_all);

13}

14

15fn describe_all() {

16 describe_counter!(

17 "gitsmart_http_requests_total",

18 "HTTP requests handled by gitsmart partitioned by endpoint and outcome (success, error)."

19 );

20 describe_counter!(

21 "gitsmart_push_total",

22 "Push operations partitioned by coordination mode (direct, waited) and outcome (success, error)."

23 );

24 describe_histogram!(

25 "gitsmart_rpc_duration_seconds",

26 Unit::Seconds,

27 "Duration of git smart-protocol RPC execution."

28 );

29}

30

31pub(crate) fn outcome(ok: bool) -> &'static str {

32 if ok { OUTCOME_SUCCESS } else { OUTCOME_ERROR }

33}

34

35pub(crate) fn record_http(endpoint: &'static str, ok: bool) {

36 ensure_described();

37 counter!("gitsmart_http_requests_total", "endpoint" => endpoint, "outcome" => outcome(ok))

38 .increment(1);

39}

40

41pub(crate) fn record_push(mode: &'static str, ok: bool) {

42 ensure_described();

43 counter!("gitsmart_push_total", "mode" => mode, "outcome" => outcome(ok)).increment(1);

44}

45

46pub(crate) struct RpcTimer {

47 start: Instant,

48}

49

50impl RpcTimer {

51 pub(crate) fn start() -> Self {

52 ensure_described();

53 Self {

54 start: Instant::now(),

55 }

56 }

57}

58

59impl Drop for RpcTimer {

60 fn drop(&mut self) {

61 histogram!("gitsmart_rpc_duration_seconds").record(self.start.elapsed());

62 }

63}

Modified lib/gitsmart/push.rs +25 −36 View lib/gitsmart/push.rs

@@ -8,28 +8,19 @@

88use crate::SmartError;

99use crate::error::SmartResult;

1010

11/// Backend primitives for the per-repository push lock + push queue.

1211#[async_trait]

1312pub trait PushOperations: Send + Sync {

14 /// Try to acquire the lock for `key`, holding it for `ttl_secs`.

15 /// Returns `true` when this caller now holds the lock.

16 async fn acquire(&self, key: &str, member: &[u8], ttl_secs: u64) -> Result<bool, SmartError>;

13 async fn acquire(&self, key: &str, member: &[u8], ttl_secs: u64) -> Result<bool, SmartError>;

1714

18 /// Renew the lock for the caller holding `token` (extend `ttl_secs`).

19 async fn renew(&self, key: &str, token: &[u8], ttl_secs: u64) -> Result<bool, SmartError>;

15 async fn renew(&self, key: &str, token: &[u8], ttl_secs: u64) -> Result<bool, SmartError>;

2016

21 /// Release the lock, but only if it is still owned by `token`.

22 async fn release(&self, key: &str, token: &[u8]) -> Result<(), SmartError>;

17 async fn release(&self, key: &str, token: &[u8]) -> Result<(), SmartError>;

2318

24 /// Append `member` to the wait queue for `key`.

25 async fn enqueue(&self, key: &str, member: &[u8]) -> Result<(), SmartError>;

19 async fn enqueue(&self, key: &str, member: &[u8]) -> Result<(), SmartError>;

2620

27 /// Remove `member` from the wait queue for `key`.

28 async fn dequeue(&self, key: &str, member: &[u8]) -> Result<(), SmartError>;

21 async fn dequeue(&self, key: &str, member: &[u8]) -> Result<(), SmartError>;

2922}

3023

31/// Redis-backed implementation. Lock = `SET NX EX`, renewal = atomic CAS,

32/// release = token-checked delete, queue = list `RPUSH`/`LREM`.

3324pub struct RedisPushBackend {

3425 pub kv: Arc<RedisCache>,

3526}

@@ -63,8 +54,6 @@

6354 }

6455}

6556

66/// No-op implementation used when no Redis is configured: every push proceeds

67/// without locking or queueing.

6857pub struct NoopPushBackend;

6958

7059#[async_trait]

@@ -95,9 +84,6 @@

9584 }

9685}

9786

98/// Serializes concurrent pushes to the same repository: the first caller holds

99/// the lock and runs immediately; later callers wait in the queue and run once

100/// the lock becomes available.

10187#[derive(Clone)]

10288pub struct PushCoordinator {

10389 backend: Arc<dyn PushOperations>,

@@ -120,34 +106,39 @@

120106 }

121107 }

122108

123 /// Run `f` under the push lock for `key`. If the lock is held by another

124 /// push, wait in the queue until it is released.

125 pub async fn run<T>(

109 pub async fn run<T>(

126110 &self,

127111 key: &str,

128112 f: impl std::future::Future<Output = SmartResult<T>>,

129113 ) -> SmartResult<T> {

130114 let token = Uuid::new_v4().to_string();

131 if self

132 .backend

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

134 .await?

135 {

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

137 }

138

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

140 loop {

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

115 let mut mode: &'static str = "waited";

116 let result: SmartResult<T> = async {

142117 if self

143118 .backend

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

145120 .await?

146121 {

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

122 mode = "direct";

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

149124 }

125

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

127 loop {

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

129 if self

130 .backend

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

132 .await?

133 {

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

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

136 }

137 }

150138 }

139 .await;

140 crate::metrics::record_push(mode, result.is_ok());

141 result

151142 }

152143

153144 async fn run_locked<T>(

@@ -170,7 +161,6 @@

170161 }

171162}

172163

173/// Periodically renews the push lock while a push is in flight.

174164struct Heartbeat {

175165 handle: tokio::task::JoinHandle<()>,

176166 stop: Arc<tokio::sync::Notify>,

@@ -315,7 +305,6 @@

315305 let backend = Arc::new(FakeBackend::default());

316306 let coordinator = PushCoordinator::new(backend.clone());

317307

318 // Pre-lock so the first run() must wait in the queue.

319308 backend.acquire("repo", b"stale-holder", 60).await.unwrap();

320309

321310 let coordinator_clone = coordinator.clone();

Modified lib/gitsmart/service.rs +16 −5 View lib/gitsmart/service.rs

@@ -17,6 +17,8 @@

1717 pub storage: Arc<dyn StorageRepository>,

1818 pub cache: Arc<cache::Cache>,

1919 pub kv: Option<Arc<RedisCache>>,

20 pub queue: Option<Arc<queue::Queue>>,

21 pub sync_settings: gitsync::SyncSettings,

2022}

2123

2224impl SmartService {

@@ -37,18 +39,27 @@

3739 storage,

3840 cache: Arc::new(cache),

3941 kv: None,

42 queue: None,

43 sync_settings: gitsync::SyncSettings::default(),

4044 })

4145 }

4246

43 /// Configure the Redis-backed push lock/queue backend.

44 /// When `None`, a no-op backend is used (pushes are never serialized).

45 pub fn with_kv(mut self, kv: Option<Arc<RedisCache>>) -> Self {

47 pub fn with_kv(mut self, kv: Option<Arc<RedisCache>>) -> Self {

4648 self.kv = kv;

4749 self

4850 }

4951

50 /// Build a push coordinator for per-repo lock + queue serialization.

51 pub fn push_coordinator(&self) -> PushCoordinator {

52 pub fn with_queue(mut self, queue: Option<Arc<queue::Queue>>) -> Self {

53 self.queue = queue;

54 self

55 }

56

57 pub fn with_sync_settings(mut self, settings: gitsync::SyncSettings) -> Self {

58 self.sync_settings = settings;

59 self

60 }

61

62 pub fn push_coordinator(&self) -> PushCoordinator {

5263 let backend: Arc<dyn PushOperations> = match &self.kv {

5364 Some(kv) => Arc::new(RedisPushBackend { kv: kv.clone() }),

5465 None => Arc::new(NoopPushBackend),