Skip to content
zhenyi

gitdata-core

0
Clone

HTTPS

SSH

SSH needs a key in your settings.

Download ZIP

feat(gitsmart): add gitsmart crate (protocol, handler, push, tests)

zhenyi committed
parent 6baf0a8 commit b861633 Browse files

13 files changed

+2330 −0

File tree 13
Added lib/gitsmart/handler/dumb.rs View lib/gitsmart/handler/dumb.rs
@@ -0,0 +1,131 @@
1use std::path::{Component, Path, PathBuf};
2
3use axum::http::{StatusCode, header};
4use axum::response::{IntoResponse, Response};
5
6use crate::error::{SmartError, SmartResult};
7use crate::service::{ResolvedRepo, SmartService};
8
9/// Dumb HTTP transport: serve bare-repository files over plain HTTP.
10/// Request paths are `info/refs`, `HEAD`, `objects/<sha>`, `info/packs`, ...
11pub async fn handle(
12 state: &SmartService,
13 space: &str,
14 repo: &str,
15 _headers: &axum::http::HeaderMap,
16 path: &str,
17) -> SmartResult<Response> {
18 let actor = crate::handler::authenticate(state, _headers).await?;
19 let resolved =
20 crate::handler::authorize(state, space, repo, &actor, repo::actor::RepoAccess::Read)
21 .await?;
22 let bare = state.bare_path(&resolved);
23 let file = resolve_within_root(&bare, path)?;
24 match tokio::fs::read(&file).await {
25 Ok(bytes) => Ok((
26 StatusCode::OK,
27 [(header::CONTENT_TYPE, guess_content_type(path))],
28 bytes,
29 )
30 .into_response()),
31 Err(e) if e.kind() == std::io::ErrorKind::NotFound => {
32 Err(SmartError::NotFound(format!("file not found: {path}")))
33 }
34 Err(e) => Err(SmartError::Io(e)),
35 }
36}
37
38pub async fn info_refs_text(
39 state: &SmartService,
40 resolved: &ResolvedRepo,
41) -> SmartResult<Response> {
42 let bare = state.bare_path(resolved);
43 let file = bare.join("info").join("refs");
44 match tokio::fs::read(&file).await {
45 Ok(bytes) => Ok((
46 StatusCode::OK,
47 [(header::CONTENT_TYPE, "text/plain; charset=utf-8")],
48 bytes,
49 )
50 .into_response()),
51 Err(e) if e.kind() == std::io::ErrorKind::NotFound => {
52 Err(SmartError::NotFound("info/refs not found".into()))
53 }
54 Err(e) => Err(SmartError::Io(e)),
55 }
56}
57
58/// Resolve a request sub-path against the bare root, rejecting any traversal
59/// that escapes the repository directory.
60fn resolve_within_root(root: &Path, request_path: &str) -> SmartResult<PathBuf> {
61 let mut base = root.to_path_buf();
62 for component in Path::new(request_path).components() {
63 match component {
64 Component::Normal(part) => base.push(part),
65 Component::CurDir => {}
66 Component::ParentDir | Component::RootDir | Component::Prefix(_) => {
67 return Err(SmartError::Forbidden("path traversal denied".into()));
68 }
69 }
70 }
71 let canonical = base
72 .canonicalize()
73 .map_err(|_| SmartError::NotFound("file not found".into()))?;
74 let root_canonical = root
75 .canonicalize()
76 .map_err(|_| SmartError::NotFound("repository not found".into()))?;
77 if canonical.starts_with(&root_canonical) {
78 Ok(canonical)
79 } else {
80 Err(SmartError::Forbidden("path escapes repository root".into()))
81 }
82}
83
84fn guess_content_type(path: &str) -> &'static str {
85 if path.ends_with("/info/refs") || path.ends_with("info/refs") {
86 "text/plain; charset=utf-8"
87 } else if path.ends_with(".pack") {
88 "application/octet-stream"
89 } else if path.ends_with(".idx") {
90 "application/octet-stream"
91 } else {
92 "application/octet-stream"
93 }
94}
95
96#[cfg(test)]
97mod tests {
98 use super::*;
99 use std::fs::{self, File};
100
101 #[test]
102 fn rejects_parent_traversal() {
103 let root = tempfile::tempdir().expect("tmpdir");
104 let err = resolve_within_root(root.path(), "../secret").expect_err("traversal");
105 assert!(matches!(err, SmartError::Forbidden(_)));
106 }
107
108 #[test]
109 fn resolves_normal_file() {
110 let root = tempfile::tempdir().expect("tmpdir");
111 fs::create_dir_all(root.path().join("objects/ab")).expect("mkdir");
112 File::create(root.path().join("objects/ab/abcdef")).expect("file");
113 let out = resolve_within_root(root.path(), "objects/ab/abcdef").expect("resolve");
114 assert_eq!(
115 out,
116 root.path()
117 .join("objects")
118 .join("ab")
119 .join("abcdef")
120 .canonicalize()
121 .expect("canon")
122 );
123 }
124
125 #[test]
126 fn rejects_absolute_path() {
127 let root = tempfile::tempdir().expect("tmpdir");
128 let err = resolve_within_root(root.path(), "/etc/passwd").expect_err("abs");
129 assert!(matches!(err, SmartError::Forbidden(_)));
130 }
131}
Added lib/gitsmart/handler/lfs.rs View lib/gitsmart/handler/lfs.rs
@@ -0,0 +1,310 @@
1use std::sync::atomic::{AtomicBool, AtomicUsize, Ordering};
2
3use axum::body::Body;
4use axum::http::{HeaderMap, Method, StatusCode, header};
5use axum::response::{IntoResponse, Response};
6use bytes::Bytes;
7use futures::{StreamExt, TryStreamExt};
8use tokio_util::io::StreamReader;
9
10use git::lfs::types::{LfsBatchObject, LfsBatchRequest, LfsBatchResponse, LfsObject, LfsOperation};
11use gitlfs::{LfsError, LfsOid};
12
13use crate::error::{SmartError, SmartResult};
14use crate::handler::{authenticate, authorize, decode_request_body};
15use crate::service::{ResolvedRepo, SmartService};
16
17const MAX_LFS_BODY: usize = 8 * 1024 * 1024 * 1024;
18/// Per-chunk buffer used when bridging axum `Body` to a `BoxStream`.
19const STREAM_BUF_LEN: usize = 64 * 1024;
20
21pub async fn dispatch(
22 state: &SmartService,
23 space: &str,
24 repo: &str,
25 headers: &HeaderMap,
26 path: &str,
27 method: Method,
28 body: Body,
29) -> SmartResult<Response> {
30 let actor = authenticate(state, headers).await?;
31 match path {
32 "info/lfs/objects/batch" if method == Method::POST => {
33 let resolved =
34 authorize(state, space, repo, &actor, repo::actor::RepoAccess::Write).await?;
35 batch(state, &resolved, headers, body).await
36 }
37 "info/lfs/verify" if method == Method::POST => {
38 let resolved =
39 authorize(state, space, repo, &actor, repo::actor::RepoAccess::Read).await?;
40 verify(state, &resolved, headers, body).await
41 }
42 p if p.starts_with("info/lfs/objects/") => {
43 let oid = &p["info/lfs/objects/".len()..];
44 let oid = LfsOid::new(oid).map_err(|_| SmartError::BadRequest("invalid oid".into()))?;
45 let resolved = authorize(
46 state,
47 space,
48 repo,
49 &actor,
50 if method == Method::PUT {
51 repo::actor::RepoAccess::Write
52 } else {
53 repo::actor::RepoAccess::Read
54 },
55 )
56 .await?;
57 match method {
58 Method::PUT => upload(state, &resolved, oid, headers, body).await,
59 Method::GET => download(state, &resolved, oid).await,
60 _ => Ok(SmartError::BadRequest("method not allowed".into()).into_response()),
61 }
62 }
63 _ => Ok(SmartError::NotFound("lfs endpoint not found".into()).into_response()),
64 }
65}
66
67async fn batch(
68 state: &SmartService,
69 resolved: &ResolvedRepo,
70 headers: &HeaderMap,
71 body: Body,
72) -> SmartResult<Response> {
73 let bytes = decode_request_body(headers, body, MAX_LFS_BODY).await?;
74 let req: LfsBatchRequest = serde_json::from_slice(&bytes)
75 .map_err(|e| SmartError::BadRequest(format!("invalid batch request: {e}")))?;
76 let repo = state.git_repository(resolved)?;
77 let lfs = repo.lfs;
78 let base = format!(
79 "/{}/{}.git/info/lfs/objects",
80 resolved.space.lower_name, resolved.repo.name
81 );
82
83 let mut objects = Vec::with_capacity(req.objects.len());
84 for obj in req.objects {
85 let oid_str = obj.oid.clone();
86 let item = match req.operation {
87 LfsOperation::Upload => {
88 let present = lfs
89 .has_object(&LfsOid::new(&oid_str).map_err(SmartError::from)?)
90 .await?;
91 if present {
92 LfsBatchObject {
93 oid: obj.oid,
94 size: obj.size,
95 actions: None,
96 error: None,
97 }
98 } else {
99 LfsBatchObject {
100 oid: obj.oid,
101 size: obj.size,
102 actions: Some(maplit("upload", format!("{base}/{}", oid_str))),
103 error: None,
104 }
105 }
106 }
107 LfsOperation::Download => {
108 let present = lfs
109 .has_object(&LfsOid::new(&oid_str).map_err(SmartError::from)?)
110 .await?;
111 if present {
112 LfsBatchObject {
113 oid: obj.oid,
114 size: obj.size,
115 actions: Some(maplit("download", format!("{base}/{}", oid_str))),
116 error: None,
117 }
118 } else {
119 LfsBatchObject {
120 oid: obj.oid,
121 size: obj.size,
122 actions: None,
123 error: Some(git::lfs::types::LfsError {
124 code: 404,
125 message: "object not found".into(),
126 }),
127 }
128 }
129 }
130 LfsOperation::Verify => {
131 let present = lfs
132 .has_object(&LfsOid::new(&oid_str).map_err(SmartError::from)?)
133 .await?;
134 if present {
135 LfsBatchObject {
136 oid: obj.oid,
137 size: obj.size,
138 actions: None,
139 error: None,
140 }
141 } else {
142 LfsBatchObject {
143 oid: obj.oid,
144 size: obj.size,
145 actions: None,
146 error: Some(git::lfs::types::LfsError {
147 code: 404,
148 message: "object not found".into(),
149 }),
150 }
151 }
152 }
153 };
154 objects.push(item);
155 }
156
157 let resp = LfsBatchResponse {
158 transfer: "basic".to_string(),
159 objects,
160 };
161 Ok((
162 StatusCode::OK,
163 [(header::CONTENT_TYPE, "application/vnd.git-lfs+json")],
164 serde_json::to_vec(&resp).map_err(serde_err)?,
165 )
166 .into_response())
167}
168
169async fn verify(
170 state: &SmartService,
171 resolved: &ResolvedRepo,
172 headers: &HeaderMap,
173 body: Body,
174) -> SmartResult<Response> {
175 let bytes = decode_request_body(headers, body, MAX_LFS_BODY).await?;
176 let req: LfsObject = serde_json::from_slice(&bytes)
177 .map_err(|e| SmartError::BadRequest(format!("invalid verify request: {e}")))?;
178 let repo = state.git_repository(resolved)?;
179 let oid = LfsOid::new(&req.oid).map_err(SmartError::from)?;
180 if repo.lfs.has_object(&oid).await? {
181 Ok(StatusCode::OK.into_response())
182 } else {
183 Ok((
184 StatusCode::NOT_FOUND,
185 [(header::CONTENT_TYPE, "application/json")],
186 r#"{"message":"object not found"}"#,
187 )
188 .into_response())
189 }
190}
191
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.
195async fn upload(
196 state: &SmartService,
197 resolved: &ResolvedRepo,
198 oid: LfsOid,
199 headers: &HeaderMap,
200 body: Body,
201) -> SmartResult<Response> {
202 let repo = state.git_repository(resolved)?;
203 let is_gzip = is_gzip(headers);
204
205 // axum Body -> futures stream of Result<Bytes, LfsError>; map IO errors to LfsError::Io.
206 let raw_stream: futures::stream::BoxStream<'static, Result<Bytes, LfsError>> = Box::pin(
207 body.into_data_stream()
208 .map(|item| item.map_err(|e| LfsError::Io(e.to_string()))),
209 );
210
211 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.
214 let io_stream = raw_stream.map_err(|e| match e {
215 LfsError::Storage(s) => std::io::Error::other(s.to_string()),
216 other => std::io::Error::other(other.to_string()),
217 });
218 let reader = StreamReader::new(io_stream);
219 let decoder = async_compression::tokio::bufread::GzipDecoder::new(reader);
220 let decoded = tokio_util::io::ReaderStream::with_capacity(decoder, STREAM_BUF_LEN);
221 Box::pin(decoded.map_err(|e| LfsError::Io(format!("gzip decode: {e}"))))
222 } else {
223 raw_stream
224 };
225
226 let limited = enforce_limit(stream, MAX_LFS_BODY);
227 repo.lfs.put_object_stream(&oid, limited).await?;
228 Ok(StatusCode::OK.into_response())
229}
230
231/// Streaming GET: build the response body directly from the store's
232/// `get_object_stream`. No buffering of the full object in memory.
233async fn download(
234 state: &SmartService,
235 resolved: &ResolvedRepo,
236 oid: LfsOid,
237) -> SmartResult<Response> {
238 let repo = state.git_repository(resolved)?;
239 let stream = repo.lfs.get_object_stream(&oid).await?;
240 let body = axum::body::Body::from_stream(stream.map_err(axum::Error::new));
241 Ok((
242 StatusCode::OK,
243 [(header::CONTENT_TYPE, "application/octet-stream")],
244 body,
245 )
246 .into_response())
247}
248
249fn is_gzip(headers: &HeaderMap) -> bool {
250 headers
251 .get(header::CONTENT_ENCODING)
252 .and_then(|v| v.to_str().ok())
253 .map(|v| v.eq_ignore_ascii_case("gzip"))
254 .unwrap_or(false)
255}
256
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.
261fn enforce_limit(
262 inner: futures::stream::BoxStream<'static, Result<Bytes, LfsError>>,
263 limit: usize,
264) -> futures::stream::BoxStream<'static, Result<Bytes, LfsError>> {
265 let total = AtomicUsize::new(0);
266 let exceeded = AtomicBool::new(false);
267 let limit_u64: u64 = limit as u64;
268 let limit_for_closure: u64 = limit_u64;
269 Box::pin(futures::stream::try_unfold(
270 (inner, total, exceeded, limit_for_closure),
271 |(mut inner, total, exceeded, limit_u64)| async move {
272 if exceeded.load(Ordering::Relaxed) {
273 return Err(LfsError::PayloadTooLarge { limit: limit_u64 });
274 }
275 match inner.next().await {
276 Some(Ok(chunk)) => {
277 let prev = total.fetch_add(chunk.len(), Ordering::Relaxed);
278 let after = prev + chunk.len();
279 if after > limit_u64 as usize {
280 exceeded.store(true, Ordering::Relaxed);
281 return Err(LfsError::PayloadTooLarge { limit: limit_u64 });
282 }
283 Ok(Some((chunk, (inner, total, exceeded, limit_u64))))
284 }
285 Some(Err(e)) => Err(e),
286 None => Ok(None),
287 }
288 },
289 ))
290}
291
292fn maplit(
293 action: &str,
294 href: String,
295) -> std::collections::HashMap<String, git::lfs::types::LfsAction> {
296 let mut map = std::collections::HashMap::new();
297 map.insert(
298 action.to_string(),
299 git::lfs::types::LfsAction {
300 href,
301 header: None,
302 expires_at: None,
303 },
304 );
305 map
306}
307
308fn serde_err(e: serde_json::Error) -> SmartError {
309 SmartError::BadRequest(format!("serialization error: {e}"))
310}
Added lib/gitsmart/handler/mod.rs View lib/gitsmart/handler/mod.rs
@@ -0,0 +1,197 @@
1pub mod dumb;
2pub mod lfs;
3pub mod smart;
4
5use std::collections::HashMap;
6
7use axum::Router;
8use axum::body::Body;
9use axum::extract::{Path, Request, State};
10use axum::http::{HeaderMap, Method, StatusCode, header};
11use axum::response::{IntoResponse, Response};
12use bytes::Bytes;
13use flate2::read::GzDecoder;
14use gitauth::BasicCredentials;
15use repo::actor::{ActorContext, RepoAccess};
16use std::io::Read;
17
18use crate::error::{SmartError, SmartResult};
19use crate::service::{SmartService, SmartState};
20
21pub fn router(state: SmartState) -> Router {
22 Router::new()
23 .route(
24 "/{space}/{repo}/{*path}",
25 axum::routing::get(dispatch)
26 .post(dispatch)
27 .put(dispatch)
28 .head(dispatch),
29 )
30 .with_state(state)
31}
32
33async fn dispatch(
34 State(state): State<SmartState>,
35 Path((space, repo, path)): Path<(String, String, String)>,
36 req: Request,
37) -> Response {
38 let method = req.method().clone();
39 let headers = req.headers().clone();
40 let query = req.uri().query().map(parse_query).unwrap_or_default();
41 let body = req.into_body();
42
43 let repo_name = match repo.strip_suffix(".git") {
44 Some(name) if !name.is_empty() => name.to_string(),
45 _ => {
46 return respond_error(SmartError::BadRequest(
47 "expected '<repo>.git' in URL".into(),
48 ));
49 }
50 };
51
52 let result: SmartResult<Response> = async {
53 let path_lower = path.to_lowercase();
54 let is_get = method == Method::GET;
55 match (path_lower.as_str(), is_get) {
56 ("info/refs", true) => {
57 let service = query.get("service").map(String::as_str);
58 smart::info_refs(&state, &space, &repo_name, &headers, service).await
59 }
60 ("git-upload-pack", false) => {
61 smart::rpc(
62 &state,
63 &space,
64 &repo_name,
65 &headers,
66 body,
67 crate::protocol::GitService::UploadPack,
68 )
69 .await
70 }
71 ("git-receive-pack", false) => {
72 smart::rpc(
73 &state,
74 &space,
75 &repo_name,
76 &headers,
77 body,
78 crate::protocol::GitService::ReceivePack,
79 )
80 .await
81 }
82 (p, _) if p.starts_with("info/lfs/") => {
83 lfs::dispatch(
84 &state,
85 &space,
86 &repo_name,
87 &headers,
88 &path_lower,
89 method,
90 body,
91 )
92 .await
93 }
94 (_, true) => dumb::handle(&state, &space, &repo_name, &headers, &path_lower).await,
95 (_, _) => Ok(SmartError::BadRequest("method not allowed".into()).into_response()),
96 }
97 }
98 .await;
99
100 result.unwrap_or_else(respond_error)
101}
102
103fn respond_error(err: SmartError) -> Response {
104 tracing::warn!(error = %err, "gitsmart request failed");
105 if err.status() == StatusCode::INTERNAL_SERVER_ERROR {
106 tracing::error!(error = %err, "gitsmart internal error");
107 }
108 err.into_response()
109}
110
111fn parse_query(s: &str) -> HashMap<String, String> {
112 s.split('&')
113 .filter_map(|pair| {
114 let (k, v) = pair.split_once('=')?;
115 Some((k.to_string(), url_decode(v)))
116 })
117 .collect()
118}
119
120fn url_decode(s: &str) -> String {
121 // Minimal percent-decoding for the service query parameter.
122 let mut out = String::with_capacity(s.len());
123 let bytes = s.as_bytes();
124 let mut i = 0;
125 while i < bytes.len() {
126 if bytes[i] == b'%' && i + 2 < bytes.len() {
127 if let Ok(byte) = u8::from_str_radix(&s[i + 1..i + 3], 16) {
128 out.push(byte as char);
129 i += 3;
130 continue;
131 }
132 }
133 out.push(bytes[i] as char);
134 i += 1;
135 }
136 out
137}
138
139/// Authenticate the request. Returns the acting user (anonymous when no
140/// Authorization header is present).
141pub(crate) async fn authenticate(
142 state: &SmartService,
143 headers: &HeaderMap,
144) -> SmartResult<ActorContext> {
145 let Some(value) = headers.get(header::AUTHORIZATION) else {
146 return Ok(ActorContext::anonymous());
147 };
148 let header = value
149 .to_str()
150 .map_err(|_| SmartError::BadRequest("invalid authorization header".into()))?;
151 let creds: BasicCredentials = gitauth::parse_authorization(header)?;
152 let user = state
153 .auth
154 .authenticate(&creds.username, &creds.password)
155 .await?;
156 Ok(user.actor)
157}
158
159/// Authorize `required` access, returning the resolved repo.
160pub(crate) async fn authorize(
161 state: &SmartService,
162 space: &str,
163 repo: &str,
164 actor: &ActorContext,
165 required: RepoAccess,
166) -> SmartResult<crate::service::ResolvedRepo> {
167 state
168 .auth
169 .authorize(space, repo, actor, required)
170 .await
171 .map_err(SmartError::from)
172}
173
174/// Decompress a request body when `Content-Encoding: gzip` is present.
175pub(crate) async fn decode_request_body(
176 headers: &HeaderMap,
177 body: Body,
178 limit: usize,
179) -> SmartResult<Bytes> {
180 let bytes = axum::body::to_bytes(body, limit)
181 .await
182 .map_err(|e| SmartError::BadRequest(format!("failed to read request body: {e}")))?;
183 let encoding = headers
184 .get(header::CONTENT_ENCODING)
185 .and_then(|v| v.to_str().ok())
186 .map(|v| v.to_ascii_lowercase());
187 if encoding.as_deref() == Some("gzip") {
188 let mut decoder = GzDecoder::new(bytes.as_ref());
189 let mut out = Vec::new();
190 decoder
191 .read_to_end(&mut out)
192 .map_err(|e| SmartError::BadRequest(format!("failed to decompress gzip body: {e}")))?;
193 Ok(Bytes::from(out))
194 } else {
195 Ok(bytes)
196 }
197}
Added lib/gitsmart/handler/smart.rs View lib/gitsmart/handler/smart.rs
@@ -0,0 +1,104 @@
1use axum::body::Body;
2use axum::http::{HeaderMap, StatusCode, header};
3use axum::response::{IntoResponse, Response};
4use bytes::Bytes;
5
6use crate::error::{SmartError, SmartResult};
7use crate::handler::dumb;
8use crate::handler::{authenticate, authorize, decode_request_body};
9use crate::protocol::{GitService, pktline, rpc};
10use crate::service::{ResolvedRepo, SmartService};
11
12const MAX_RPC_BODY: usize = 1 << 30;
13
14pub async fn info_refs(
15 state: &SmartService,
16 space: &str,
17 repo: &str,
18 headers: &HeaderMap,
19 service: Option<&str>,
20) -> SmartResult<Response> {
21 let required = match service {
22 Some("git-upload-pack") => repo::actor::RepoAccess::Read,
23 Some("git-receive-pack") => repo::actor::RepoAccess::Write,
24 Some(other) => {
25 return Err(SmartError::BadRequest(format!(
26 "unsupported service: {other}"
27 )));
28 }
29 None => repo::actor::RepoAccess::Read,
30 };
31
32 let actor = authenticate(state, headers).await?;
33 let resolved = authorize(state, space, repo, &actor, required).await?;
34
35 match service {
36 Some(name) => {
37 let git_service = GitService::from_name(name)?;
38 smart_advertisement(state, &resolved, git_service).await
39 }
40 None => dumb::info_refs_text(state, &resolved).await,
41 }
42}
43
44async fn smart_advertisement(
45 state: &SmartService,
46 resolved: &ResolvedRepo,
47 service: GitService,
48) -> SmartResult<Response> {
49 let bare = state.bare_path(resolved);
50 let input = Bytes::new();
51 let output = rpc::run_rpc_bytes(&bare, service, true, input).await?;
52 let mut body = pktline::encode_header(service.dashed_name());
53 body.extend_from_slice(&output);
54
55 Ok((
56 StatusCode::OK,
57 [
58 (header::CONTENT_TYPE, service.advertisement_content_type()),
59 (header::CACHE_CONTROL, "no-cache"),
60 ],
61 body,
62 )
63 .into_response())
64}
65
66pub async fn rpc(
67 state: &SmartService,
68 space: &str,
69 repo: &str,
70 headers: &HeaderMap,
71 body: Body,
72 service: GitService,
73) -> SmartResult<Response> {
74 let actor = authenticate(state, headers).await?;
75 let resolved = authorize(state, space, repo, &actor, service.required_access()).await?;
76
77 let run = async {
78 let input = decode_request_body(headers, body, MAX_RPC_BODY).await?;
79 let bare = state.bare_path(&resolved);
80 let output = rpc::run_rpc_bytes(&bare, service, false, input).await?;
81 Ok::<_, SmartError>(
82 (
83 StatusCode::OK,
84 [
85 (header::CONTENT_TYPE, service.result_content_type()),
86 (header::CACHE_CONTROL, "no-cache"),
87 ],
88 output,
89 )
90 .into_response(),
91 )
92 };
93
94 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.
98 GitService::ReceivePack => {
99 let key = format!("push:{}:{}", resolved.channel.id, resolved.repo.id);
100 state.push_coordinator().run(&key, run).await
101 }
102 GitService::UploadPack => run.await,
103 }
104}
Added lib/gitsmart/protocol/mod.rs View lib/gitsmart/protocol/mod.rs
@@ -0,0 +1,6 @@
1pub mod pktline;
2pub mod rpc;
3pub mod service;
4
5pub use pktline::encode_header;
6pub use service::GitService;
Added lib/gitsmart/protocol/pktline.rs View lib/gitsmart/protocol/pktline.rs
@@ -0,0 +1,94 @@
1use std::io::Write;
2
3use crate::error::{SmartError, SmartResult};
4
5const MAX_PKT_SIZE: usize = 65_520;
6
7pub fn encode_line(line: &[u8]) -> Vec<u8> {
8 let len = line.len() + 4;
9 debug_assert!(len <= MAX_PKT_SIZE, "pkt-line too long: {len}");
10 let mut out = Vec::with_capacity(4 + line.len());
11 write!(out, "{len:04x}").expect("write hex len");
12 out.extend_from_slice(line);
13 out
14}
15
16pub fn flush_pkt() -> Vec<u8> {
17 b"0000".to_vec()
18}
19
20pub fn encode_header(service: &str) -> Vec<u8> {
21 let payload = format!("# service={service}\n");
22 let mut out = encode_line(payload.as_bytes());
23 out.extend_from_slice(&flush_pkt());
24 out
25}
26
27pub fn decode_stream(input: &[u8]) -> SmartResult<Vec<Vec<u8>>> {
28 let mut lines = Vec::new();
29 let mut pos = 0usize;
30 while pos + 4 <= input.len() {
31 let header = &input[pos..pos + 4];
32 let len = usize::from_str_radix(
33 std::str::from_utf8(header)
34 .map_err(|_| SmartError::BadRequest("invalid pkt-line length header".into()))?,
35 16,
36 )
37 .map_err(|_| SmartError::BadRequest("invalid pkt-line length header".into()))?;
38 pos += 4;
39 if len == 0 {
40 break;
41 }
42 if len < 4 || len > MAX_PKT_SIZE {
43 return Err(SmartError::BadRequest(format!(
44 "invalid pkt-line length: {len}"
45 )));
46 }
47 let payload_end = pos + len - 4;
48 if payload_end > input.len() {
49 return Err(SmartError::BadRequest("truncated pkt-line".into()));
50 }
51 lines.push(input[pos..payload_end].to_vec());
52 pos = payload_end;
53 }
54 Ok(lines)
55}
56
57#[cfg(test)]
58mod tests {
59 use super::*;
60
61 #[test]
62 fn roundtrip_lineless() {
63 let encoded = encode_line(b"hello");
64 assert_eq!(b"0009hello", encoded.as_slice());
65 let lines = decode_stream(&encoded).expect("decode");
66 assert_eq!(
67 vec![b"hello".as_slice()],
68 lines.iter().map(|l| l.as_slice()).collect::<Vec<_>>()
69 );
70 }
71
72 #[test]
73 fn flush_terminates() {
74 let mut buf = encode_line(b"a");
75 buf.extend_from_slice(&flush_pkt());
76 buf.extend_from_slice(&encode_line(b"ignored"));
77 let lines = decode_stream(&buf).expect("decode");
78 assert_eq!(1, lines.len());
79 }
80
81 #[test]
82 fn header_has_service_and_flush() {
83 let encoded = encode_header("git-upload-pack");
84 let lines = decode_stream(&encoded).expect("decode");
85 assert_eq!(1, lines.len());
86 assert_eq!(b"# service=git-upload-pack\n", lines[0].as_slice());
87 }
88
89 #[test]
90 fn rejects_bad_length() {
91 let err = decode_stream(b"zzzz").expect_err("should fail");
92 assert!(matches!(err, SmartError::BadRequest(_)));
93 }
94}
Added lib/gitsmart/protocol/rpc.rs View lib/gitsmart/protocol/rpc.rs
@@ -0,0 +1,150 @@
1use std::io;
2use std::path::Path;
3use std::process::Stdio;
4use std::sync::Arc;
5
6use bytes::Bytes;
7use futures::StreamExt;
8use tokio::io::{AsyncReadExt, AsyncWriteExt};
9use tokio::process::{ChildStdout, Command};
10use tokio::task::JoinHandle;
11
12use crate::error::{SmartError, SmartResult};
13
14use super::service::GitService;
15
16pub type RpcStream = futures::stream::BoxStream<'static, Result<Bytes, io::Error>>;
17
18type SharedResult = Arc<tokio::sync::Mutex<Option<Result<(), String>>>>;
19
20struct StreamState {
21 reader: ChildStdout,
22 reaper: Option<JoinHandle<()>>,
23 shared: SharedResult,
24 done: bool,
25}
26
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.
33pub async fn run_rpc(
34 bare_path: &Path,
35 service: GitService,
36 advertise: bool,
37 input: RpcStream,
38) -> SmartResult<RpcStream> {
39 let mut cmd = Command::new("git");
40 cmd.arg(service.name());
41 cmd.arg("--stateless-rpc");
42 if advertise {
43 cmd.arg("--advertise-refs");
44 }
45 cmd.arg(bare_path);
46 cmd.env("GIT_DIR", bare_path)
47 .env("GIT_PROTOCOL", "version=2")
48 .env("GIT_TERMINAL_PROMPT", "0")
49 .stdin(Stdio::piped())
50 .stdout(Stdio::piped())
51 .stderr(Stdio::piped());
52
53 let service_name = service.name();
54 let mut child = cmd
55 .spawn()
56 .map_err(|e| SmartError::Internal(format!("failed to spawn git {service_name}: {e}")))?;
57
58 let mut stdin = child.stdin.take().expect("stdin piped");
59 let stdin_task = tokio::spawn(async move {
60 let mut input = input;
61 while let Some(chunk) = input.next().await {
62 let chunk = chunk?;
63 stdin.write_all(&chunk).await?;
64 }
65 stdin.shutdown().await?;
66 io::Result::Ok(())
67 });
68
69 let stdout = child.stdout.take().expect("stdout piped");
70 let mut stderr = child.stderr.take().expect("stderr piped");
71 let shared: SharedResult = Arc::new(tokio::sync::Mutex::new(None));
72 let shared_clone = Arc::clone(&shared);
73 let reaper = tokio::spawn(async move {
74 let mut stderr_buf = Vec::new();
75 let mut chunk = vec![0u8; 4096];
76 loop {
77 match stderr.read(&mut chunk).await {
78 Ok(0) | Err(_) => break,
79 Ok(n) => stderr_buf.extend_from_slice(&chunk[..n]),
80 }
81 }
82 let result = match child.wait().await {
83 Ok(status) if status.success() => Ok(()),
84 Ok(status) => Err(format!(
85 "git {service_name} exited with {status}: {}",
86 String::from_utf8_lossy(&stderr_buf)
87 )),
88 Err(e) => Err(format!("waiting for git {service_name} failed: {e}")),
89 };
90 *shared_clone.lock().await = Some(result);
91 });
92
93 let state = StreamState {
94 reader: stdout,
95 reaper: Some(reaper),
96 shared,
97 done: false,
98 };
99
100 let stream = futures::stream::unfold(state, |mut st| async move {
101 if st.done {
102 return None;
103 }
104 let mut buf = vec![0u8; 8192];
105 match st.reader.read(&mut buf).await {
106 Ok(0) => {
107 st.done = true;
108 if let Some(reaper) = st.reaper.take() {
109 let _ = reaper.await;
110 }
111 let pending = st.shared.lock().await.clone();
112 match pending {
113 Some(Err(msg)) => Some((Err(io::Error::new(io::ErrorKind::Other, msg)), st)),
114 _ => None,
115 }
116 }
117 Ok(n) => Some((Ok(Bytes::copy_from_slice(&buf[..n])), st)),
118 Err(e) => {
119 st.done = true;
120 Some((Err(e), st))
121 }
122 }
123 });
124
125 let _ = stdin_task;
126 Ok(Box::pin(stream))
127}
128
129/// Buffered convenience wrapper used by tests: sends the whole body, collects stdout.
130pub async fn run_rpc_bytes(
131 bare_path: &Path,
132 service: GitService,
133 advertise: bool,
134 input: Bytes,
135) -> SmartResult<Bytes> {
136 let stream = run_rpc(
137 bare_path,
138 service,
139 advertise,
140 Box::pin(futures::stream::once(async move { Ok(input) })),
141 )
142 .await?;
143 let mut out = Vec::new();
144 let mut stream = stream;
145 while let Some(item) = stream.next().await {
146 let chunk = item.map_err(|e| SmartError::Internal(e.to_string()))?;
147 out.extend_from_slice(&chunk);
148 }
149 Ok(Bytes::from(out))
150}
Added lib/gitsmart/protocol/service.rs View lib/gitsmart/protocol/service.rs
@@ -0,0 +1,97 @@
1use std::fmt;
2
3use crate::error::{SmartError, SmartResult};
4
5#[derive(Debug, Clone, Copy, PartialEq, Eq)]
6pub enum GitService {
7 UploadPack,
8 ReceivePack,
9}
10
11impl GitService {
12 pub fn from_name(name: &str) -> SmartResult<Self> {
13 match name {
14 "git-upload-pack" => Ok(Self::UploadPack),
15 "git-receive-pack" => Ok(Self::ReceivePack),
16 _ => Err(SmartError::BadRequest(format!(
17 "unsupported git service: {name}"
18 ))),
19 }
20 }
21
22 pub fn name(self) -> &'static str {
23 match self {
24 Self::UploadPack => "upload-pack",
25 Self::ReceivePack => "receive-pack",
26 }
27 }
28
29 /// Dashed binary name used in the smart-HTTP `# service=` pkt-line header.
30 pub fn dashed_name(self) -> &'static str {
31 match self {
32 Self::UploadPack => "git-upload-pack",
33 Self::ReceivePack => "git-receive-pack",
34 }
35 }
36
37 pub fn advertisement_content_type(self) -> &'static str {
38 match self {
39 Self::UploadPack => "application/x-git-upload-pack-advertisement",
40 Self::ReceivePack => "application/x-git-receive-pack-advertisement",
41 }
42 }
43
44 pub fn result_content_type(self) -> &'static str {
45 match self {
46 Self::UploadPack => "application/x-git-upload-pack-result",
47 Self::ReceivePack => "application/x-git-receive-pack-result",
48 }
49 }
50
51 pub fn required_access(self) -> repo::actor::RepoAccess {
52 match self {
53 Self::UploadPack => repo::actor::RepoAccess::Read,
54 Self::ReceivePack => repo::actor::RepoAccess::Write,
55 }
56 }
57}
58
59impl fmt::Display for GitService {
60 fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
61 f.write_str(self.name())
62 }
63}
64
65#[cfg(test)]
66mod tests {
67 use super::*;
68
69 #[test]
70 fn parses_known_services() {
71 assert_eq!(
72 GitService::from_name("git-upload-pack").unwrap(),
73 GitService::UploadPack
74 );
75 assert_eq!(
76 GitService::from_name("git-receive-pack").unwrap(),
77 GitService::ReceivePack
78 );
79 }
80
81 #[test]
82 fn rejects_unknown_service() {
83 assert!(GitService::from_name("git-daemon").is_err());
84 }
85
86 #[test]
87 fn content_types() {
88 assert_eq!(
89 GitService::UploadPack.advertisement_content_type(),
90 "application/x-git-upload-pack-advertisement"
91 );
92 assert_eq!(
93 GitService::ReceivePack.result_content_type(),
94 "application/x-git-receive-pack-result"
95 );
96 }
97}
Added lib/gitsmart/tests/smart.rs View lib/gitsmart/tests/smart.rs
@@ -0,0 +1,703 @@
1use std::net::SocketAddr;
2use std::path::Path;
3use std::sync::Arc;
4use std::time::Duration;
5
6use db::channel::channel_members;
7use db::channel::channels;
8use db::conn::Db;
9use db::conn::sea_orm::{
10 ActiveModelTrait, ConnectionTrait, Database, DatabaseBackend, DatabaseConnection, EntityTrait,
11 Schema, Set,
12};
13use db::repo::repos;
14use db::space::spaces;
15use db::user::users;
16use gitauth::GitAuthService;
17use gitsmart::handler::router;
18use gitsmart::{SmartConfig, SmartService};
19use jwt::{JwtSettings, JwtState, SigningKeyCache};
20use password_hash::rand_core::OsRng;
21use password_hash::{PasswordHasher, SaltString};
22use storage::config::{LocalStorageSettings, StorageSettings};
23use tempfile::TempDir;
24use time::OffsetDateTime;
25use tokio::sync::RwLock;
26use uuid::Uuid;
27
28fn now() -> OffsetDateTime {
29 OffsetDateTime::now_utc()
30}
31
32fn hash_for(password: &str) -> String {
33 let salt = SaltString::generate(&mut OsRng);
34 argon2::Argon2::default()
35 .hash_password(password.as_bytes(), &salt)
36 .expect("hash")
37 .to_string()
38}
39
40async fn create_table<E: EntityTrait>(db: &DatabaseConnection) {
41 let stmt = Schema::new(DatabaseBackend::Sqlite).create_table_from_entity(E::default());
42 db.execute(&stmt).await.expect("create table");
43}
44
45async fn setup_db() -> DatabaseConnection {
46 let db = Database::connect("sqlite::memory:")
47 .await
48 .expect("connect sqlite memory");
49 create_table::<users::Entity>(&db).await;
50 create_table::<spaces::Entity>(&db).await;
51 create_table::<channels::Entity>(&db).await;
52 create_table::<channel_members::Entity>(&db).await;
53 create_table::<db::space::space_members::Entity>(&db).await;
54 create_table::<db::space::space_roles::Entity>(&db).await;
55 create_table::<db::space::space_member_roles::Entity>(&db).await;
56 create_table::<db::space::space_role_permissions::Entity>(&db).await;
57 create_table::<db::repo::repo_collaborations::Entity>(&db).await;
58 create_table::<repos::Entity>(&db).await;
59 db
60}
61
62async fn insert_user(
63 db: &impl ConnectionTrait,
64 uid: Uuid,
65 username: &str,
66 passwd_hash: Option<&str>,
67) {
68 users::ActiveModel {
69 uid: Set(uid),
70 username: Set(username.to_string()),
71 lower_name: Set(username.to_lowercase()),
72 display_name: Set(username.to_string()),
73 avatar: Set(String::new()),
74 is_active: Set(true),
75 is_admin: Set(false),
76 is_restricted: Set(false),
77 prohibit_login: Set(false),
78 visibility: Set(users::ProfileVisibility::Public),
79 bot: Set(false),
80 system: Set(false),
81 totp_enabled: Set(false),
82 version: Set(1),
83 passwd_hash: Set(passwd_hash.map(|s| s.to_string())),
84 created_at: Set(now()),
85 updated_at: Set(now()),
86 ..Default::default()
87 }
88 .insert(db)
89 .await
90 .expect("insert user");
91}
92
93async fn insert_space(
94 db: &impl ConnectionTrait,
95 id: Uuid,
96 lower_name: &str,
97 owner_id: Uuid,
98 is_public: bool,
99) {
100 spaces::ActiveModel {
101 id: Set(id),
102 lower_name: Set(lower_name.to_string()),
103 name: Set(lower_name.to_string()),
104 owner_id: Set(owner_id),
105 public: Set(is_public),
106 is_active: Set(true),
107 allow_join: Set(true),
108 allow_invite: Set(true),
109 member_count: Set(0),
110 allow_join_level: Set(spaces::AllowJoinLevel::None),
111 member_must_enable_2fa: Set(false),
112 created_at: Set(now()),
113 updated_at: Set(now()),
114 ..Default::default()
115 }
116 .insert(db)
117 .await
118 .expect("insert space");
119}
120
121async fn insert_channel(
122 db: &impl ConnectionTrait,
123 id: Uuid,
124 space_id: Uuid,
125 name: &str,
126 created_by: Uuid,
127) {
128 channels::ActiveModel {
129 id: Set(id),
130 space_id: Set(Some(space_id)),
131 name: Set(name.to_string()),
132 position: Set(0),
133 is_archived: Set(false),
134 created_at: Set(now()),
135 updated_at: Set(now()),
136 created_by: Set(created_by),
137 ..Default::default()
138 }
139 .insert(db)
140 .await
141 .expect("insert channel");
142}
143
144async fn insert_repo(
145 db: &impl ConnectionTrait,
146 id: Uuid,
147 channel_id: Uuid,
148 name: &str,
149 is_private: bool,
150) {
151 repos::ActiveModel {
152 id: Set(id),
153 channel_id: Set(channel_id),
154 lower_name: Set(name.to_lowercase()),
155 name: Set(name.to_string()),
156 default_branch: Set("main".to_string()),
157 nums_watch: Set(0),
158 nums_star: Set(0),
159 nums_pr: Set(0),
160 nums_commits: Set(0),
161 nums_branches: Set(0),
162 nums_tags: Set(0),
163 enable_lfs: Set(true),
164 is_private: Set(is_private),
165 is_archived: Set(false),
166 size: Set(0),
167 git_size: Set(0),
168 lfs_size: Set(0),
169 allow_rebase: Set(true),
170 allow_rebase_explicit: Set(true),
171 allow_squash_merge: Set(true),
172 allow_merge_commits: Set(true),
173 created_at: Set(now()),
174 object_format: Set("sha1".to_string()),
175 ..Default::default()
176 }
177 .insert(db)
178 .await
179 .expect("insert repo");
180}
181
182async fn insert_channel_member(
183 db: &impl ConnectionTrait,
184 channel_id: Uuid,
185 user_id: Uuid,
186 role: channel_members::ChannelMemberRole,
187) {
188 channel_members::ActiveModel {
189 channel_id: Set(channel_id),
190 user_id: Set(user_id),
191 role: Set(role),
192 muted: Set(false),
193 joined_at: Set(now()),
194 ..Default::default()
195 }
196 .insert(db)
197 .await
198 .expect("insert channel member");
199}
200
201async fn git(dir: &Path, args: &[&str]) -> String {
202 let out = tokio::process::Command::new("git")
203 .args(args)
204 .current_dir(dir)
205 .env("GIT_TERMINAL_PROMPT", "0")
206 .output()
207 .await
208 .expect("run git");
209 assert!(
210 out.status.success(),
211 "git {:?} failed: {}",
212 args,
213 String::from_utf8_lossy(&out.stderr)
214 );
215 String::from_utf8_lossy(&out.stdout).into_owned()
216}
217
218fn basic_auth(username: &str, password: &str) -> String {
219 use base64::Engine;
220 format!(
221 "Basic {}",
222 base64::engine::general_purpose::STANDARD.encode(format!("{username}:{password}"))
223 )
224}
225
226struct TestServer {
227 _tmp: TempDir,
228 addr: SocketAddr,
229 base: String,
230 bare: std::path::PathBuf,
231 db: DatabaseConnection,
232 git_root: std::path::PathBuf,
233 space_id: Uuid,
234 channel_id: Uuid,
235}
236
237async fn start_server() -> TestServer {
238 let tmp = TempDir::new().expect("tempdir");
239 let root = tmp.path().to_path_buf();
240
241 let git_root = root.join("git");
242 let lfs_root = root.join("lfs");
243 let storage_root = root.join("storage");
244 std::fs::create_dir_all(&git_root).expect("git root");
245 std::fs::create_dir_all(&lfs_root).expect("lfs root");
246 std::fs::create_dir_all(&storage_root).expect("storage root");
247
248 let conn = setup_db().await;
249 let db = Db::from_connection(conn.clone());
250
251 let alice = Uuid::new_v4();
252 let space_id = Uuid::new_v4();
253 let channel_id = Uuid::new_v4();
254 let repo_id = Uuid::new_v4();
255
256 insert_user(&conn, alice, "alice", Some(hash_for("secret").as_str())).await;
257 insert_space(&conn, space_id, "acme", alice, true).await;
258 insert_channel(&conn, channel_id, space_id, "dev", alice).await;
259 insert_repo(&conn, repo_id, channel_id, "widget", false).await;
260 insert_channel_member(
261 &conn,
262 channel_id,
263 alice,
264 channel_members::ChannelMemberRole::Owner,
265 )
266 .await;
267
268 let bare = git_root.join(format!("{}/{}.git", channel_id, repo_id));
269 std::fs::create_dir_all(bare.parent().expect("bare parent")).expect("bare dirs");
270 git(&root, &["init", "--bare", bare.to_str().expect("bare str")]).await;
271
272 let mut settings = StorageSettings::default();
273 settings.set_local(
274 LocalStorageSettings::builder()
275 .root_dir(storage_root.to_path_buf())
276 .build(),
277 );
278 let storage = storage::repo::from_config(&settings).expect("storage");
279
280 let jwt = JwtState::new(
281 conn.clone(),
282 JwtSettings::default(),
283 Arc::new(RwLock::new(SigningKeyCache::new())),
284 );
285 let auth = GitAuthService::new(db, None, jwt);
286 let config = SmartConfig {
287 git_storage_root: git_root.to_str().expect("git root str").to_string(),
288 lfs_storage_root: lfs_root.to_str().expect("lfs root str").to_string(),
289 object_cache_prefix: "gitsmart".to_string(),
290 object_cache_ttl: 300,
291 };
292 let svc = SmartService::new(auth, config, storage)
293 .await
294 .expect("smart service");
295 let app = router(Arc::new(svc));
296
297 let listener = tokio::net::TcpListener::bind("127.0.0.1:0")
298 .await
299 .expect("bind");
300 let addr = listener.local_addr().expect("addr");
301 tokio::spawn(async move {
302 axum::serve(listener, app).await.expect("serve");
303 });
304
305 let base = format!("http://{addr}/acme/widget.git");
306 TestServer {
307 _tmp: tmp,
308 addr,
309 base,
310 bare,
311 db: conn,
312 git_root,
313 space_id,
314 channel_id,
315 }
316}
317
318async fn seed_bare(root: &Path, bare: &Path) {
319 let src = root.join("seed-src");
320 std::fs::create_dir_all(&src).expect("src dir");
321 git(&src, &["init", "-q"]).await;
322 git(&src, &["config", "user.email", "alice@example.com"]).await;
323 git(&src, &["config", "user.name", "Alice"]).await;
324 std::fs::write(src.join("hello.txt"), b"hello world\n").expect("write file");
325 git(&src, &["add", "."]).await;
326 git(&src, &["commit", "-q", "-m", "init"]).await;
327 git(
328 &src,
329 &["remote", "add", "origin", bare.to_str().expect("bare str")],
330 )
331 .await;
332 git(&src, &["push", "-q", "origin", "HEAD:main"]).await;
333 git(
334 root,
335 &[
336 "--git-dir",
337 bare.to_str().expect("bare str"),
338 "update-server-info",
339 ],
340 )
341 .await;
342}
343
344#[tokio::test]
345async fn smart_clone_and_push() {
346 let srv = start_server().await;
347
348 let src = srv._tmp.path().join("src");
349 std::fs::create_dir_all(&src).expect("src dir");
350 git(&src, &["init", "-q"]).await;
351 git(&src, &["config", "user.email", "alice@example.com"]).await;
352 git(&src, &["config", "user.name", "Alice"]).await;
353 std::fs::write(src.join("hello.txt"), b"hello world\n").expect("write file");
354 git(&src, &["add", "."]).await;
355 git(&src, &["commit", "-q", "-m", "init"]).await;
356 git(
357 &src,
358 &[
359 "remote",
360 "add",
361 "origin",
362 &format!("http://alice:secret@{}/acme/widget.git", srv.addr),
363 ],
364 )
365 .await;
366 git(&src, &["push", "-q", "-u", "origin", "main"]).await;
367
368 let dst = srv._tmp.path().join("dst");
369 git(
370 &srv._tmp.path(),
371 &[
372 "clone",
373 "-q",
374 &format!("http://alice:secret@{}/acme/widget.git", srv.addr),
375 dst.to_str().expect("dst str"),
376 ],
377 )
378 .await;
379 let content = std::fs::read_to_string(dst.join("hello.txt")).expect("read cloned file");
380 assert_eq!(content, "hello world\n");
381}
382
383#[tokio::test]
384async fn dumb_info_refs_text() {
385 let srv = start_server().await;
386 seed_bare(&srv._tmp.path().to_path_buf(), &srv.bare).await;
387 let client = reqwest::Client::new();
388 let resp = client
389 .get(format!("{}/info/refs", srv.base))
390 .send()
391 .await
392 .expect("get info/refs");
393 assert_eq!(resp.status(), reqwest::StatusCode::OK, "dumb info/refs");
394 let body = resp.text().await.expect("body");
395 assert!(body.contains("refs/heads/main"), "body: {body}");
396}
397
398#[tokio::test]
399async fn lfs_put_get_roundtrip() {
400 let srv = start_server().await;
401 let client = reqwest::Client::new();
402 let data = b"large binary payload".to_vec();
403 let oid = gitlfs::LfsOid::compute(&data);
404 let url = format!("{}/info/lfs/objects/{}", srv.base, oid.as_str());
405
406 let put = client
407 .put(&url)
408 .header("Authorization", basic_auth("alice", "secret"))
409 .body(data.clone())
410 .send()
411 .await
412 .expect("lfs put");
413 assert_eq!(put.status(), reqwest::StatusCode::OK, "lfs put failed");
414
415 let get = client
416 .get(&url)
417 .header("Authorization", basic_auth("alice", "secret"))
418 .send()
419 .await
420 .expect("lfs get");
421 assert_eq!(get.status(), reqwest::StatusCode::OK);
422 let got = get.bytes().await.expect("lfs bytes");
423 assert_eq!(got.as_ref(), data.as_slice());
424}
425
426#[tokio::test]
427async fn lfs_batch_upload_action() {
428 let srv = start_server().await;
429 let client = reqwest::Client::new();
430 let data = b"batch test payload".to_vec();
431 let oid = gitlfs::LfsOid::compute(&data);
432 let body = serde_json::json!({
433 "operation": "upload",
434 "transfers": ["basic"],
435 "objects": [{"oid": oid.as_str(), "size": data.len() as i64}]
436 });
437 let resp = client
438 .post(format!("{}/info/lfs/objects/batch", srv.base))
439 .header("Authorization", basic_auth("alice", "secret"))
440 .header("Content-Type", "application/vnd.git-lfs+json")
441 .body(body.to_string())
442 .send()
443 .await
444 .expect("lfs batch");
445 assert_eq!(resp.status(), reqwest::StatusCode::OK, "lfs batch failed");
446 let json: serde_json::Value = resp.json().await.expect("batch json");
447 let href = json["objects"][0]["actions"]["upload"]["href"]
448 .as_str()
449 .expect("upload href");
450 assert!(
451 href.contains(&format!("/info/lfs/objects/{}", oid.as_str())),
452 "href: {href}"
453 );
454}
455
456#[tokio::test]
457async fn anonymous_cannot_read_private() {
458 let srv = start_server().await;
459 let client = reqwest::Client::new();
460
461 let repo_id = Uuid::new_v4();
462 insert_repo(&srv.db, repo_id, srv.channel_id, "secret", true).await;
463
464 let bare = srv
465 .git_root
466 .join(format!("{}/{}.git", srv.channel_id, repo_id));
467 std::fs::create_dir_all(bare.parent().expect("bare parent")).expect("bare dirs");
468 git(
469 &srv._tmp.path(),
470 &["init", "--bare", bare.to_str().expect("bare str")],
471 )
472 .await;
473
474 let resp = client
475 .get(format!(
476 "http://{}/acme/secret.git/info/refs?service=git-upload-pack",
477 srv.addr
478 ))
479 .send()
480 .await
481 .expect("private info refs");
482 assert_eq!(resp.status(), reqwest::StatusCode::UNAUTHORIZED);
483}
484
485#[tokio::test]
486async fn unknown_repo_404() {
487 let srv = start_server().await;
488 let client = reqwest::Client::new();
489 let resp = client
490 .get(format!(
491 "http://{}/acme/ghost.git/info/refs?service=git-upload-pack",
492 srv.addr
493 ))
494 .send()
495 .await
496 .expect("ghost info refs");
497 assert_eq!(resp.status(), reqwest::StatusCode::NOT_FOUND);
498}
499
500#[tokio::test]
501async fn smart_advertisement_responds() {
502 let srv = start_server().await;
503 let client = reqwest::Client::new();
504 let resp = client
505 .get(format!("{}/info/refs?service=git-upload-pack", srv.base))
506 .timeout(Duration::from_secs(10))
507 .send()
508 .await
509 .expect("info refs");
510 assert_eq!(resp.status(), reqwest::StatusCode::OK);
511 let body = resp.bytes().await.expect("bytes");
512 assert!(!body.is_empty());
513 assert!(
514 String::from_utf8_lossy(&body).contains("service=git-upload-pack"),
515 "advertisement header missing: {:?}",
516 String::from_utf8_lossy(&body)
517 );
518}
519
520// === Streaming LFS tests ===
521
522/// Generate a deterministic pseudo-random byte vector of `len` bytes,
523/// seeded by the supplied value so callers can produce distinct payloads.
524fn pseudo_random_bytes(len: usize, seed: u64) -> Vec<u8> {
525 let mut out = Vec::with_capacity(len);
526 let mut state: u64 = seed;
527 for _ in 0..len {
528 state ^= state << 13;
529 state ^= state >> 7;
530 state ^= state << 17;
531 out.push(state as u8);
532 }
533 out
534}
535
536/// Build a payload that gzip can actually compress (low-entropy text-like data).
537fn compressible_payload(len: usize) -> Vec<u8> {
538 let pattern = b"the quick brown fox jumps over the lazy dog\n";
539 let mut out = Vec::with_capacity(len);
540 while out.len() < len {
541 let take = (len - out.len()).min(pattern.len());
542 out.extend_from_slice(&pattern[..take]);
543 }
544 out
545}
546
547/// Compress `data` with gzip using flate2 (already in workspace deps).
548fn gzip_compress(data: &[u8]) -> Vec<u8> {
549 use flate2::Compression;
550 use flate2::write::GzEncoder;
551 use std::io::Write;
552 let mut encoder = GzEncoder::new(Vec::new(), Compression::default());
553 encoder.write_all(data).expect("gzip write");
554 encoder.finish().expect("gzip finish")
555}
556
557#[tokio::test]
558async fn lfs_upload_streams_large() {
559 let srv = start_server().await;
560 let client = reqwest::Client::new();
561
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.
565 let data = pseudo_random_bytes(5 * 1024 * 1024, 0xA1B2_C3D4_E5F6_0701);
566 let oid = gitlfs::LfsOid::compute(&data);
567 let url = format!("{}/info/lfs/objects/{}", srv.base, oid.as_str());
568
569 let put = client
570 .put(&url)
571 .header("Authorization", basic_auth("alice", "secret"))
572 .body(data.clone())
573 .timeout(Duration::from_secs(30))
574 .send()
575 .await
576 .expect("lfs put");
577 assert_eq!(put.status(), reqwest::StatusCode::OK, "lfs put failed");
578
579 // Round-trip: GET and verify sha256 + byte-equal content.
580 let got = client
581 .get(&url)
582 .header("Authorization", basic_auth("alice", "secret"))
583 .send()
584 .await
585 .expect("lfs get");
586 assert_eq!(got.status(), reqwest::StatusCode::OK);
587 let body = got.bytes().await.expect("lfs bytes");
588 assert_eq!(body.len(), data.len(), "size mismatch");
589 assert_eq!(gitlfs::LfsOid::compute(&body), oid, "sha256 mismatch");
590 assert_eq!(body.as_ref(), data.as_slice(), "content mismatch");
591}
592
593#[tokio::test]
594async fn lfs_upload_gzip() {
595 let srv = start_server().await;
596 let client = reqwest::Client::new();
597
598 let data = compressible_payload(2 * 1024 * 1024);
599 let oid = gitlfs::LfsOid::compute(&data);
600 let url = format!("{}/info/lfs/objects/{}", srv.base, oid.as_str());
601
602 let gz = gzip_compress(&data);
603 assert!(gz.len() < data.len(), "gzip did not compress random data");
604
605 let put = client
606 .put(&url)
607 .header("Authorization", basic_auth("alice", "secret"))
608 .header("Content-Encoding", "gzip")
609 .body(gz)
610 .timeout(Duration::from_secs(30))
611 .send()
612 .await
613 .expect("lfs put gzip");
614 assert_eq!(put.status(), reqwest::StatusCode::OK, "lfs gzip put failed");
615
616 let got = client
617 .get(&url)
618 .header("Authorization", basic_auth("alice", "secret"))
619 .send()
620 .await
621 .expect("lfs get");
622 assert_eq!(got.status(), reqwest::StatusCode::OK);
623 let body = got.bytes().await.expect("lfs bytes");
624 assert_eq!(body.as_ref(), data.as_slice(), "decoded mismatch");
625 assert_eq!(gitlfs::LfsOid::compute(&body), oid);
626}
627
628#[tokio::test]
629async fn lfs_upload_hash_mismatch() {
630 let srv = start_server().await;
631 let client = reqwest::Client::new();
632
633 // Compute OID for one payload, but PUT a different one.
634 let real = pseudo_random_bytes(64 * 1024, 0x1111_1111_1111_1111);
635 let fake = pseudo_random_bytes(64 * 1024, 0x2222_2222_2222_2222);
636 let oid = gitlfs::LfsOid::compute(&real);
637 assert_ne!(real, fake);
638 let url = format!("{}/info/lfs/objects/{}", srv.base, oid.as_str());
639
640 let put = client
641 .put(&url)
642 .header("Authorization", basic_auth("alice", "secret"))
643 .body(fake)
644 .timeout(Duration::from_secs(30))
645 .send()
646 .await
647 .expect("lfs put");
648 // Store auto-aborts multipart on hash mismatch; handler surfaces 4xx.
649 assert!(
650 put.status().is_client_error() || put.status().is_server_error(),
651 "expected error status, got {}",
652 put.status()
653 );
654
655 // Object must not have been stored.
656 let head = client
657 .get(&url)
658 .header("Authorization", basic_auth("alice", "secret"))
659 .send()
660 .await
661 .expect("lfs get");
662 assert_eq!(head.status(), reqwest::StatusCode::NOT_FOUND);
663}
664
665#[tokio::test]
666async fn lfs_download_streams_large() {
667 let srv = start_server().await;
668 let client = reqwest::Client::new();
669
670 let data = pseudo_random_bytes(3 * 1024 * 1024, 0xCAFE_BABE_DEAD_BEEF);
671 let oid = gitlfs::LfsOid::compute(&data);
672 let url = format!("{}/info/lfs/objects/{}", srv.base, oid.as_str());
673
674 // Seed the store.
675 let put = client
676 .put(&url)
677 .header("Authorization", basic_auth("alice", "secret"))
678 .body(data.clone())
679 .send()
680 .await
681 .expect("seed put");
682 assert_eq!(put.status(), reqwest::StatusCode::OK);
683
684 // Stream-download and verify, streaming the response body (no full
685 // buffering on client side either).
686 let resp = client
687 .get(&url)
688 .header("Authorization", basic_auth("alice", "secret"))
689 .send()
690 .await
691 .expect("lfs get");
692 assert_eq!(resp.status(), reqwest::StatusCode::OK);
693
694 use futures::StreamExt;
695 let mut stream = resp.bytes_stream();
696 let mut got = Vec::with_capacity(data.len());
697 while let Some(chunk) = stream.next().await {
698 got.extend_from_slice(&chunk.expect("chunk"));
699 }
700 assert_eq!(got.len(), data.len(), "size mismatch");
701 assert_eq!(gitlfs::LfsOid::compute(&got), oid, "sha256 mismatch");
702 assert_eq!(got, data, "content mismatch");
703}
Added lib/gitsmart/config.rs View lib/gitsmart/config.rs
@@ -0,0 +1,18 @@
1#[derive(Debug, Clone)]
2pub struct SmartConfig {
3 pub git_storage_root: String,
4 pub lfs_storage_root: String,
5 pub object_cache_prefix: String,
6 pub object_cache_ttl: u64,
7}
8
9impl Default for SmartConfig {
10 fn default() -> Self {
11 Self {
12 git_storage_root: "/var/lib/repo/git".to_string(),
13 lfs_storage_root: "/var/lib/repo/lfs".to_string(),
14 object_cache_prefix: "gitsmart".to_string(),
15 object_cache_ttl: 300,
16 }
17 }
18}
Added lib/gitsmart/error.rs View lib/gitsmart/error.rs
@@ -0,0 +1,94 @@
1use thiserror::Error;
2
3#[derive(Debug, Error)]
4pub enum SmartError {
5 #[error("{0}")]
6 BadRequest(String),
7 #[error("{0}")]
8 Unauthorized(String),
9 #[error("{0}")]
10 Forbidden(String),
11 #[error("{0}")]
12 NotFound(String),
13 #[error("{0}")]
14 Conflict(String),
15 #[error("{0}")]
16 Internal(String),
17 #[error("io error: {0}")]
18 Io(#[from] std::io::Error),
19 #[error("gitauth error: {0}")]
20 GitAuth(gitauth::GitAuthError),
21 #[error("git error: {0}")]
22 Git(#[from] git::Error),
23 #[error("lfs error: {0}")]
24 Lfs(gitlfs::LfsError),
25 #[error("http error: {0}")]
26 Http(#[from] http::Error),
27 #[error("redis error: {0}")]
28 Redis(#[from] cache::redis::RedisError),
29}
30
31pub type SmartResult<T> = Result<T, SmartError>;
32
33impl SmartError {
34 pub fn status(&self) -> http::StatusCode {
35 match self {
36 Self::BadRequest(_) => http::StatusCode::BAD_REQUEST,
37 Self::Unauthorized(_) => http::StatusCode::UNAUTHORIZED,
38 Self::Forbidden(_) => http::StatusCode::FORBIDDEN,
39 Self::NotFound(_) => http::StatusCode::NOT_FOUND,
40 Self::Conflict(_) => http::StatusCode::CONFLICT,
41 Self::Lfs(gitlfs::LfsError::InvalidOid(_)) => http::StatusCode::BAD_REQUEST,
42 Self::Lfs(gitlfs::LfsError::SizeMismatch { .. }) => http::StatusCode::BAD_REQUEST,
43 Self::Lfs(gitlfs::LfsError::HashMismatch { .. }) => http::StatusCode::BAD_REQUEST,
44 Self::Lfs(gitlfs::LfsError::PayloadTooLarge { .. }) => http::StatusCode::BAD_REQUEST,
45 Self::Lfs(gitlfs::LfsError::NotFound(_)) => http::StatusCode::NOT_FOUND,
46 _ => http::StatusCode::INTERNAL_SERVER_ERROR,
47 }
48 }
49
50 pub fn message(&self) -> String {
51 match self {
52 Self::BadRequest(m)
53 | Self::Unauthorized(m)
54 | Self::Forbidden(m)
55 | Self::NotFound(m)
56 | Self::Conflict(m)
57 | Self::Internal(m) => m.clone(),
58 other => other.to_string(),
59 }
60 }
61}
62
63impl From<gitauth::GitAuthError> for SmartError {
64 fn from(e: gitauth::GitAuthError) -> Self {
65 match e {
66 gitauth::GitAuthError::BadRequest(m) => Self::BadRequest(m),
67 gitauth::GitAuthError::Unauthorized(m) => Self::Unauthorized(m),
68 gitauth::GitAuthError::Forbidden(m) => Self::Forbidden(m),
69 gitauth::GitAuthError::NotFound(m) => Self::NotFound(m),
70 gitauth::GitAuthError::Conflict(m) => Self::Conflict(m),
71 other => Self::Internal(other.to_string()),
72 }
73 }
74}
75
76impl From<gitlfs::LfsError> for SmartError {
77 fn from(e: gitlfs::LfsError) -> Self {
78 Self::Lfs(e)
79 }
80}
81
82impl axum::response::IntoResponse for SmartError {
83 fn into_response(self) -> axum::response::Response {
84 let status = self.status();
85 let mut resp = (status, self.message()).into_response();
86 if status == http::StatusCode::UNAUTHORIZED {
87 resp.headers_mut().insert(
88 http::header::WWW_AUTHENTICATE,
89 http::HeaderValue::from_static("Basic realm=\"gitsmart\""),
90 );
91 }
92 resp
93 }
94}
Added lib/gitsmart/push.rs View lib/gitsmart/push.rs
@@ -0,0 +1,337 @@
1use std::sync::Arc;
2use std::time::Duration;
3
4use async_trait::async_trait;
5use cache::RedisCache;
6use uuid::Uuid;
7
8use crate::SmartError;
9use crate::error::SmartResult;
10
11/// Backend primitives for the per-repository push lock + push queue.
12#[async_trait]
13pub 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>;
17
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>;
20
21 /// Release the lock, but only if it is still owned by `token`.
22 async fn release(&self, key: &str, token: &[u8]) -> Result<(), SmartError>;
23
24 /// Append `member` to the wait queue for `key`.
25 async fn enqueue(&self, key: &str, member: &[u8]) -> Result<(), SmartError>;
26
27 /// Remove `member` from the wait queue for `key`.
28 async fn dequeue(&self, key: &str, member: &[u8]) -> Result<(), SmartError>;
29}
30
31/// Redis-backed implementation. Lock = `SET NX EX`, renewal = atomic CAS,
32/// release = token-checked delete, queue = list `RPUSH`/`LREM`.
33pub struct RedisPushBackend {
34 pub kv: Arc<RedisCache>,
35}
36
37#[async_trait]
38impl PushOperations for RedisPushBackend {
39 async fn acquire(&self, key: &str, member: &[u8], ttl_secs: u64) -> Result<bool, SmartError> {
40 Ok(self.kv.set_nx(key, member.to_vec(), ttl_secs).await?)
41 }
42
43 async fn renew(&self, key: &str, token: &[u8], ttl_secs: u64) -> Result<bool, SmartError> {
44 Ok(self
45 .kv
46 .compare_and_set_bytes(key, token, token.to_vec(), ttl_secs)
47 .await?)
48 }
49
50 async fn release(&self, key: &str, token: &[u8]) -> Result<(), SmartError> {
51 let _ = self.kv.compare_and_delete(key, token).await?;
52 Ok(())
53 }
54
55 async fn enqueue(&self, key: &str, member: &[u8]) -> Result<(), SmartError> {
56 let _ = self.kv.rpush(key, member.to_vec()).await?;
57 Ok(())
58 }
59
60 async fn dequeue(&self, key: &str, member: &[u8]) -> Result<(), SmartError> {
61 let _ = self.kv.lrem(key, 1, member).await?;
62 Ok(())
63 }
64}
65
66/// No-op implementation used when no Redis is configured: every push proceeds
67/// without locking or queueing.
68pub struct NoopPushBackend;
69
70#[async_trait]
71impl PushOperations for NoopPushBackend {
72 async fn acquire(
73 &self,
74 _key: &str,
75 _member: &[u8],
76 _ttl_secs: u64,
77 ) -> Result<bool, SmartError> {
78 Ok(true)
79 }
80
81 async fn renew(&self, _key: &str, _token: &[u8], _ttl_secs: u64) -> Result<bool, SmartError> {
82 Ok(true)
83 }
84
85 async fn release(&self, _key: &str, _token: &[u8]) -> Result<(), SmartError> {
86 Ok(())
87 }
88
89 async fn enqueue(&self, _key: &str, _member: &[u8]) -> Result<(), SmartError> {
90 Ok(())
91 }
92
93 async fn dequeue(&self, _key: &str, _member: &[u8]) -> Result<(), SmartError> {
94 Ok(())
95 }
96}
97
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.
101#[derive(Clone)]
102pub struct PushCoordinator {
103 backend: Arc<dyn PushOperations>,
104 lock_ttl_secs: u64,
105 renew_interval_secs: u64,
106 poll_interval: Duration,
107}
108
109const DEFAULT_LOCK_TTL_SECS: u64 = 120;
110const DEFAULT_RENEW_INTERVAL_SECS: u64 = 40;
111const DEFAULT_POLL_INTERVAL: Duration = Duration::from_millis(200);
112
113impl PushCoordinator {
114 pub fn new(backend: Arc<dyn PushOperations>) -> Self {
115 Self {
116 backend,
117 lock_ttl_secs: DEFAULT_LOCK_TTL_SECS,
118 renew_interval_secs: DEFAULT_RENEW_INTERVAL_SECS,
119 poll_interval: DEFAULT_POLL_INTERVAL,
120 }
121 }
122
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>(
126 &self,
127 key: &str,
128 f: impl std::future::Future<Output = SmartResult<T>>,
129 ) -> SmartResult<T> {
130 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;
142 if self
143 .backend
144 .acquire(key, token.as_bytes(), self.lock_ttl_secs)
145 .await?
146 {
147 self.backend.dequeue(key, token.as_bytes()).await?;
148 return self.run_locked(key, &token, f).await;
149 }
150 }
151 }
152
153 async fn run_locked<T>(
154 &self,
155 key: &str,
156 token: &str,
157 f: impl std::future::Future<Output = SmartResult<T>>,
158 ) -> SmartResult<T> {
159 let heartbeat = Heartbeat::spawn(
160 self.backend.clone(),
161 key.to_string(),
162 token.to_string(),
163 self.lock_ttl_secs,
164 self.renew_interval_secs,
165 );
166 let result = f.await;
167 heartbeat.stop().await;
168 self.backend.release(key, token.as_bytes()).await?;
169 result
170 }
171}
172
173/// Periodically renews the push lock while a push is in flight.
174struct Heartbeat {
175 handle: tokio::task::JoinHandle<()>,
176 stop: Arc<tokio::sync::Notify>,
177}
178
179impl Heartbeat {
180 fn spawn(
181 backend: Arc<dyn PushOperations>,
182 key: String,
183 token: String,
184 ttl: u64,
185 interval: u64,
186 ) -> Self {
187 let stop = Arc::new(tokio::sync::Notify::new());
188 let stop_clone = stop.clone();
189 let handle = tokio::spawn(async move {
190 loop {
191 tokio::select! {
192 _ = stop_clone.notified() => break,
193 _ = tokio::time::sleep(Duration::from_secs(interval)) => {
194 if backend
195 .renew(&key, token.as_bytes(), ttl)
196 .await
197 .is_err()
198 {
199 break;
200 }
201 }
202 }
203 }
204 });
205 Self { handle, stop }
206 }
207
208 async fn stop(self) {
209 self.stop.notify_one();
210 let _ = self.handle.await;
211 }
212}
213
214#[cfg(test)]
215mod tests {
216 use super::*;
217 use std::collections::HashMap;
218 use std::sync::Mutex;
219 use std::sync::atomic::{AtomicUsize, Ordering as AtomicOrdering};
220
221 #[derive(Default)]
222 struct FakeBackend {
223 locks: Arc<Mutex<HashMap<String, Vec<u8>>>>,
224 queue: Arc<Mutex<HashMap<String, Vec<Vec<u8>>>>>,
225 }
226
227 #[async_trait]
228 impl PushOperations for FakeBackend {
229 async fn acquire(
230 &self,
231 key: &str,
232 member: &[u8],
233 _ttl_secs: u64,
234 ) -> Result<bool, SmartError> {
235 let mut locks = self.locks.lock().unwrap();
236 if locks.contains_key(key) {
237 return Ok(false);
238 }
239 locks.insert(key.to_string(), member.to_vec());
240 Ok(true)
241 }
242
243 async fn renew(&self, key: &str, token: &[u8], _ttl_secs: u64) -> Result<bool, SmartError> {
244 let locks = self.locks.lock().unwrap();
245 Ok(locks.get(key) == Some(&token.to_vec()))
246 }
247
248 async fn release(&self, key: &str, token: &[u8]) -> Result<(), SmartError> {
249 let mut locks = self.locks.lock().unwrap();
250 if locks.get(key) == Some(&token.to_vec()) {
251 locks.remove(key);
252 }
253 Ok(())
254 }
255
256 async fn enqueue(&self, key: &str, member: &[u8]) -> Result<(), SmartError> {
257 self.queue
258 .lock()
259 .unwrap()
260 .entry(key.to_string())
261 .or_default()
262 .push(member.to_vec());
263 Ok(())
264 }
265
266 async fn dequeue(&self, key: &str, member: &[u8]) -> Result<(), SmartError> {
267 if let Some(queue) = self.queue.lock().unwrap().get_mut(key) {
268 queue.retain(|m| m != &member.to_vec());
269 }
270 Ok(())
271 }
272 }
273
274 #[tokio::test]
275 async fn noop_backend_passes_through() {
276 let coordinator = PushCoordinator::new(Arc::new(NoopPushBackend));
277 let result: SmartResult<i32> = coordinator.run("repo", async { Ok(42) }).await;
278 assert_eq!(result.unwrap(), 42);
279 }
280
281 #[tokio::test]
282 async fn concurrent_pushes_serialize() {
283 let coordinator = PushCoordinator::new(Arc::new(FakeBackend::default()));
284 let running = Arc::new(AtomicUsize::new(0));
285 let max = Arc::new(AtomicUsize::new(0));
286
287 let run_one = |coordinator: PushCoordinator,
288 running: Arc<AtomicUsize>,
289 max: Arc<AtomicUsize>| {
290 async move {
291 coordinator
292 .run("repo", async move {
293 let current = running.fetch_add(1, AtomicOrdering::SeqCst) + 1;
294 max.fetch_max(current, AtomicOrdering::SeqCst);
295 tokio::time::sleep(Duration::from_millis(20)).await;
296 running.fetch_sub(1, AtomicOrdering::SeqCst);
297 Ok::<(), SmartError>(())
298 })
299 .await
300 .unwrap();
301 }
302 };
303
304 tokio::join!(
305 run_one(coordinator.clone(), running.clone(), max.clone()),
306 run_one(coordinator.clone(), running.clone(), max.clone()),
307 run_one(coordinator.clone(), running.clone(), max.clone()),
308 );
309
310 assert_eq!(max.load(AtomicOrdering::SeqCst), 1);
311 }
312
313 #[tokio::test]
314 async fn waiter_acquires_after_release() {
315 let backend = Arc::new(FakeBackend::default());
316 let coordinator = PushCoordinator::new(backend.clone());
317
318 // Pre-lock so the first run() must wait in the queue.
319 backend.acquire("repo", b"stale-holder", 60).await.unwrap();
320
321 let coordinator_clone = coordinator.clone();
322 let backend_clone = backend.clone();
323 let waiter = tokio::spawn(async move {
324 let result: SmartResult<i32> = coordinator_clone.run("repo", async { Ok(7) }).await;
325 result.unwrap()
326 });
327
328 tokio::time::sleep(Duration::from_millis(50)).await;
329 backend_clone
330 .release("repo", b"stale-holder")
331 .await
332 .unwrap();
333
334 assert_eq!(waiter.await.unwrap(), 7);
335 assert!(!backend.locks.lock().unwrap().contains_key("repo"));
336 }
337}
Added lib/gitsmart/service.rs View lib/gitsmart/service.rs
@@ -0,0 +1,89 @@
1use std::path::PathBuf;
2use std::sync::Arc;
3
4use cache::RedisCache;
5use gitauth::GitAuthService;
6use storage::StorageRepository;
7
8use crate::config::SmartConfig;
9use crate::push::{NoopPushBackend, PushCoordinator, PushOperations, RedisPushBackend};
10
11pub use gitauth::ResolvedRepo;
12
13#[derive(Clone)]
14pub struct SmartService {
15 pub auth: GitAuthService,
16 pub config: SmartConfig,
17 pub storage: Arc<dyn StorageRepository>,
18 pub cache: Arc<cache::Cache>,
19 pub kv: Option<Arc<RedisCache>>,
20}
21
22impl SmartService {
23 pub async fn new(
24 auth: GitAuthService,
25 config: SmartConfig,
26 storage: Arc<dyn StorageRepository>,
27 ) -> crate::SmartResult<Self> {
28 let cache_cfg = cache::config::cache::AppCacheSettings::default();
29 let cache = cache::CacheBuilder::from_config(cache_cfg)
30 .build()
31 .await
32 .map_err(|e| crate::SmartError::Internal(format!("cache build: {e}")))?
33 .ok_or_else(|| crate::SmartError::Internal("cache disabled".to_string()))?;
34 Ok(Self {
35 auth,
36 config,
37 storage,
38 cache: Arc::new(cache),
39 kv: None,
40 })
41 }
42
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 {
46 self.kv = kv;
47 self
48 }
49
50 /// Build a push coordinator for per-repo lock + queue serialization.
51 pub fn push_coordinator(&self) -> PushCoordinator {
52 let backend: Arc<dyn PushOperations> = match &self.kv {
53 Some(kv) => Arc::new(RedisPushBackend { kv: kv.clone() }),
54 None => Arc::new(NoopPushBackend),
55 };
56 PushCoordinator::new(backend)
57 }
58
59 pub fn bare_path(&self, resolved: &ResolvedRepo) -> PathBuf {
60 PathBuf::from(&self.config.git_storage_root)
61 .join(resolved.repo.channel_id.to_string())
62 .join(format!("{}.git", resolved.repo.id))
63 }
64
65 pub fn lfs_prefix(&self, resolved: &ResolvedRepo) -> String {
66 format!(
67 "{}/{}/{}",
68 self.config.lfs_storage_root, resolved.repo.channel_id, resolved.repo.id
69 )
70 }
71
72 pub fn git_repository(
73 &self,
74 resolved: &ResolvedRepo,
75 ) -> crate::SmartResult<git::GitRepository> {
76 let oc = git::object::cache::ObjectCache {
77 prefix: self.config.object_cache_prefix.clone(),
78 cache: self.cache.clone(),
79 expire_ttl: self.config.object_cache_ttl,
80 };
81 Ok(git::GitRepository::new(
82 self.bare_path(resolved),
83 oc,
84 self.storage.clone(),
85 ))
86 }
87}
88
89pub type SmartState = Arc<SmartService>;