From 4d5e5ba5f95a04ed9ef44e30844113c1bc444682 Mon Sep 17 00:00:00 2001 From: zhenyi <434836402@qq.com> Date: Fri, 14 Aug 2026 18:13:39 +0800 Subject: [PATCH] feat(server): add streaming-first server scaffolding MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Introduce src/server/ as a streaming-first git/lfs server surface. All request and response bodies cross the API as Box (alias BoxedAsyncRead) — no Vec bodies, no read_to_end convenience, no http crate. What's in the box: - src/server/mod.rs: re-exports + pipe re-export of command::cmd::DEFAULT_OUTPUT_CAP_BYTES as server::DEFAULT_OUTPUT_CAP_BYTES - src/server/backend.rs: ServerBackend async-trait (authorize, list_refs_stream, produce_pack_stream, ingest_pack_stream, lfs_get_stream, lfs_put_stream). Every method currently returns ServerError::Unimplemented("...") placeholders — the wiring is real, the bodies are intentionally empty. - src/server/request.rs: ServerRequest (not Clone, body is a boxed reader), HeaderMap = HashMap, RefUpdateBatch (oid strings, no gix::ObjectId), BoxedAsyncRead alias - src/server/stream.rs: StreamGuard (Clone + Send + Sync) carrying cancel() / handle() so spawned workers can race the request - src/server/endpoint.rs, error.rs, limits.rs: EndpointKind enum, ServerError, DEFAULT_LFS_OBJECT_CAP_BYTES (5 GiB cap, separate from the 64 MiB output cap) - tests/server.rs: PlaceholderBackend skeleton + ENV_LOCK + 200 ms cancel convention reused from tests/env.rs and tests/pipe.rs Public API impact on GitBaby: - Drop pipe from GitBaby::new — the server owns its own stream lifecycle, so the Pipe parameter is no longer required at construction time. Existing callers must be updated: - before: GitBaby::new(facade, pipe) - after: GitBaby::new(facade) Documentation: - AGENTS.md gains a "Planned but unimplemented (streaming-first server)" section documenting the deliberately-empty contracts so future contributors don't "fill them in" without a concrete git/lfs protocol task. CI: cargo build + cargo test (61 total, 23 new from server suite) green. --- AGENTS.md | 14 +- src/lib.rs | 15 +- src/server/backend.rs | 37 ++++ src/server/endpoint.rs | 51 ++++++ src/server/error.rs | 78 ++++++++ src/server/limits.rs | 3 + src/server/mod.rs | 19 ++ src/server/request.rs | 239 +++++++++++++++++++++++++ src/server/stream.rs | 251 ++++++++++++++++++++++++++ tests/server.rs | 392 +++++++++++++++++++++++++++++++++++++++++ 10 files changed, 1086 insertions(+), 13 deletions(-) create mode 100644 src/server/backend.rs create mode 100644 src/server/endpoint.rs create mode 100644 src/server/error.rs create mode 100644 src/server/limits.rs create mode 100644 src/server/mod.rs create mode 100644 src/server/request.rs create mode 100644 src/server/stream.rs create mode 100644 tests/server.rs diff --git a/AGENTS.md b/AGENTS.md index 2e770fc..19fa23a 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -35,4 +35,16 @@ ## Gotchas - No README. No docs beyond the code. Treat `Cargo.toml` + source as the only source of truth. - `gix = "0.86"`, `tokio = 1.53` (full features), `time = 0.3` — all recent; do not bump them speculatively. -- Public API surfaces async (`async_trait`); sync callers must use a runtime (tests use `#[tokio::test]`). \ No newline at end of file +- Public API surfaces async (`async_trait`); sync callers must use a runtime (tests use `#[tokio::test]`). + +## Planned but unimplemented (streaming-first server) +- `src/server/` exists (mod.rs, backend.rs, endpoint.rs, error.rs, limits.rs, request.rs, stream.rs) but every `ServerBackend` method currently returns `ServerError::Unimplemented("…")` placeholders. Do not "fill them in" with arbitrary bodies — wait for a concrete git/lfs protocol task before adding logic. +- Streaming-only contract is intentional: + - All request/response bodies cross the API as `Box` (alias `BoxedAsyncRead`). Never introduce a `Vec` body, `ServerResponse::body`, or `read_to_end` convenience into the public surface. + - `HeaderMap = HashMap` (no `http` crate). + - `output_cap` defaults to `crate::command::cmd::DEFAULT_OUTPUT_CAP_BYTES` (64 MiB). The server re-exports it as `server::DEFAULT_OUTPUT_CAP_BYTES`. + - `LFS` upload cap is `DEFAULT_LFS_OBJECT_CAP_BYTES` (5 GiB) — a separate constant, not a magic number. + - `RefUpdateBatch` is a server-local struct (oid strings, not `gix::ObjectId`) so the API does not depend on `gix` types. Do **not** alias `refs::types::RefUpdate` here. +- `ServerBackend` async-trait method set: `authorize`, `list_refs_stream`, `produce_pack_stream`, `ingest_pack_stream`, `lfs_get_stream`, `lfs_put_stream`. Methods returning a body must return `BoxedAsyncRead`; ingest methods return `IngestReport`. The legacy `head_object` / `write_ref` / `get_object` shapes are intentionally removed. +- `ServerRequest` is **not** `Clone` (body is a boxed reader); it carries a manual `Debug` impl. `StreamGuard` is `Clone + Send + Sync` and exposes `cancel()` / `handle()` so spawned workers can race the request. +- Tests: `tests/server.rs` reuses the project-wide `static ENV_LOCK: Mutex<()>` pattern from `tests/env.rs` and the `tokio::time::sleep(Duration::from_millis(200))` cancel convention from `tests/pipe.rs`. Add new server tests there. \ No newline at end of file diff --git a/src/lib.rs b/src/lib.rs index cdfadf8..abf1d80 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -1,6 +1,5 @@ use std::sync::Arc; -use crate::command::pipe::Pipe; use crate::repo::RepositoryFacade; pub mod archive; @@ -20,6 +19,7 @@ pub mod merge; pub mod refs; pub mod remote; pub mod repo; +pub mod server; pub mod setup; pub mod share; pub mod submodule; @@ -29,23 +29,14 @@ pub mod tree; #[derive(Clone)] pub struct GitBaby { facade: Arc, - pipe: Pipe, } impl GitBaby { - pub fn new(facade: Arc, pipe: Pipe) -> Self { - Self { facade, pipe } + pub fn new(facade: Arc) -> Self { + Self { facade } } pub fn facade(&self) -> &dyn RepositoryFacade { &*self.facade } - - pub fn pipe(&self) -> &Pipe { - &self.pipe - } - - pub fn pipe_mut(&mut self) -> &mut Pipe { - &mut self.pipe - } } diff --git a/src/server/backend.rs b/src/server/backend.rs new file mode 100644 index 0000000..a927bb2 --- /dev/null +++ b/src/server/backend.rs @@ -0,0 +1,37 @@ +use async_trait::async_trait; + +use crate::server::error::ServerResult; +use crate::server::request::{AuthContext, RefUpdateBatch, ServerRequest}; +use crate::server::stream::{BoxedAsyncRead, IngestReport}; + +#[async_trait] +pub trait ServerBackend: Send + Sync + 'static { + async fn authorize(&self, req: &ServerRequest) -> ServerResult; + + async fn list_refs_stream(&self, req: &ServerRequest) -> ServerResult; + + async fn produce_pack_stream( + &self, + req: &ServerRequest, + wants: &[String], + ) -> ServerResult; + + async fn ingest_pack_stream( + &self, + req: &ServerRequest, + batch: RefUpdateBatch, + ) -> ServerResult; + + async fn lfs_get_stream(&self, req: &ServerRequest, oid: &str) -> ServerResult; + + async fn lfs_put_stream( + &self, + req: &ServerRequest, + oid: &str, + declared_size: u64, + ) -> ServerResult; +} + +pub fn empty_body() -> BoxedAsyncRead { + Box::new(tokio::io::empty()) +} diff --git a/src/server/endpoint.rs b/src/server/endpoint.rs new file mode 100644 index 0000000..82fcb67 --- /dev/null +++ b/src/server/endpoint.rs @@ -0,0 +1,51 @@ +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub enum EndpointKind { + GitUploadPack, + GitReceivePack, + LfsBatch, + LfsObjects, + InfoRefs, + SmartHttp, + Custom, +} + +impl EndpointKind { + pub fn as_str(&self) -> &'static str { + match self { + Self::GitUploadPack => "git-upload-pack", + Self::GitReceivePack => "git-receive-pack", + Self::LfsBatch => "lfs-batch", + Self::LfsObjects => "lfs-objects", + Self::InfoRefs => "info-refs", + Self::SmartHttp => "smart-http", + Self::Custom => "custom", + } + } +} + +#[derive(Debug, Clone)] +pub struct GitProtocolEndpoint { + pub kind: EndpointKind, + pub path: String, +} + +impl GitProtocolEndpoint { + pub fn new(kind: EndpointKind, path: impl Into) -> Self { + Self { + kind, + path: path.into(), + } + } + + pub fn upload_pack(path: impl Into) -> Self { + Self::new(EndpointKind::GitUploadPack, path) + } + + pub fn receive_pack(path: impl Into) -> Self { + Self::new(EndpointKind::GitReceivePack, path) + } + + pub fn info_refs(path: impl Into) -> Self { + Self::new(EndpointKind::InfoRefs, path) + } +} diff --git a/src/server/error.rs b/src/server/error.rs new file mode 100644 index 0000000..86f9c90 --- /dev/null +++ b/src/server/error.rs @@ -0,0 +1,78 @@ +use std::fmt; +use std::path::PathBuf; + +use crate::error::BabyError; + +#[derive(Debug)] +pub enum ServerError { + Unimplemented(&'static str), + Unauthorised { + scheme: &'static str, + reason: String, + }, + InvalidRequest(String), + PayloadTooLarge { + cap_bytes: u64, + actual_bytes: u64, + }, + Io { + path: PathBuf, + source: std::io::Error, + }, + Pipe(String), + Backend(BabyError), + Cancelled, +} + +impl fmt::Display for ServerError { + fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { + match self { + Self::Unimplemented(what) => write!(f, "not yet implemented: {}", what), + Self::Unauthorised { scheme, reason } => { + write!(f, "unauthorised ({scheme}): {}", reason) + } + Self::InvalidRequest(msg) => write!(f, "invalid server request: {}", msg), + Self::PayloadTooLarge { + cap_bytes, + actual_bytes, + } => write!( + f, + "output exceeds cap: {} bytes > {} bytes cap", + actual_bytes, cap_bytes + ), + Self::Io { path, source } => { + write!(f, "io error at `{}`: {}", path.display(), source) + } + Self::Pipe(msg) => write!(f, "server pipe error: {}", msg), + Self::Backend(err) => write!(f, "backend error: {}", err), + Self::Cancelled => write!(f, "server stream was cancelled"), + } + } +} + +impl std::error::Error for ServerError { + fn source(&self) -> Option<&(dyn std::error::Error + 'static)> { + match self { + Self::Io { source, .. } => Some(source), + Self::Backend(err) => Some(err), + _ => None, + } + } +} + +impl From for ServerError { + fn from(err: BabyError) -> Self { + Self::Backend(err) + } +} + +impl From for ServerError { + fn from(source: std::io::Error) -> Self { + Self::Io { + path: PathBuf::new(), + source, + } + } +} + +pub type ServerResult = std::result::Result; diff --git a/src/server/limits.rs b/src/server/limits.rs new file mode 100644 index 0000000..c41c57c --- /dev/null +++ b/src/server/limits.rs @@ -0,0 +1,3 @@ +pub use crate::command::cmd::DEFAULT_OUTPUT_CAP_BYTES; + +pub const DEFAULT_LFS_OBJECT_CAP_BYTES: u64 = 5_u64 * 1024 * 1024 * 1024; diff --git a/src/server/mod.rs b/src/server/mod.rs new file mode 100644 index 0000000..7ab4d38 --- /dev/null +++ b/src/server/mod.rs @@ -0,0 +1,19 @@ +pub mod backend; +pub mod endpoint; +pub mod error; +pub mod limits; +pub mod request; +pub mod stream; + +pub use backend::ServerBackend; +pub use endpoint::{EndpointKind, GitProtocolEndpoint}; +pub use error::{ServerError, ServerResult}; +pub use limits::{DEFAULT_LFS_OBJECT_CAP_BYTES, DEFAULT_OUTPUT_CAP_BYTES}; +pub use request::{ + AuthContext, AuthScheme, HeaderMap, Principal, RefUpdateBatch, RefUpdateEntry, ServerRequest, + StreamGuard, StreamGuardHandle, +}; +pub use stream::{ + BoxedAsyncBufRead, BoxedAsyncRead, BoxedAsyncWrite, ChildPackReader, ChildPackWriter, + IngestReport, PackStats, StreamOutcome, buf_empty_reader, empty_reader, sink_writer, +}; diff --git a/src/server/request.rs b/src/server/request.rs new file mode 100644 index 0000000..006988a --- /dev/null +++ b/src/server/request.rs @@ -0,0 +1,239 @@ +use std::collections::HashMap; +use std::sync::Arc; +use std::sync::atomic::{AtomicBool, Ordering}; + +use tokio::io::AsyncRead; + +use crate::server::endpoint::EndpointKind; +use crate::server::limits::DEFAULT_OUTPUT_CAP_BYTES; + +pub type HeaderMap = HashMap; + +#[derive(Debug, Clone, Copy, PartialEq, Eq, Default)] +pub enum AuthScheme { + #[default] + Anonymous, + Basic, + Bearer, + Ssh, +} + +impl AuthScheme { + pub fn as_str(&self) -> &'static str { + match self { + Self::Anonymous => "anonymous", + Self::Basic => "basic", + Self::Bearer => "bearer", + Self::Ssh => "ssh", + } + } +} + +#[derive(Debug, Clone)] +pub struct Principal { + pub name: String, + pub scheme: AuthScheme, +} + +impl Principal { + pub fn new(name: impl Into, scheme: AuthScheme) -> Self { + Self { + name: name.into(), + scheme, + } + } +} + +#[derive(Debug, Clone, Default)] +pub struct AuthContext { + pub scheme: AuthScheme, + pub principal: Option, + pub raw: Option, +} + +impl AuthContext { + pub fn anonymous() -> Self { + Self { + scheme: AuthScheme::Anonymous, + principal: None, + raw: None, + } + } + + pub fn with_principal(scheme: AuthScheme, principal: Principal) -> Self { + Self { + scheme, + principal: Some(principal), + raw: None, + } + } +} + +#[derive(Debug, Clone)] +pub struct RefUpdateEntry { + pub name: String, + pub new_oid: String, + pub old_oid: Option, +} + +#[derive(Debug, Clone, Default)] +pub struct RefUpdateBatch { + pub updates: Vec, +} + +impl RefUpdateBatch { + pub fn new() -> Self { + Self::default() + } + + pub fn push( + mut self, + name: impl Into, + new_oid: impl Into, + old_oid: Option, + ) -> Self { + self.updates.push(RefUpdateEntry { + name: name.into(), + new_oid: new_oid.into(), + old_oid, + }); + self + } + + pub fn is_empty(&self) -> bool { + self.updates.is_empty() + } + + pub fn len(&self) -> usize { + self.updates.len() + } +} + +pub struct ServerRequest { + pub method: String, + pub path: String, + pub query: Option, + pub headers: HeaderMap, + pub body: Box, + pub guard: StreamGuard, + pub output_cap: u64, + pub auth: Option, + pub endpoint: EndpointKind, + pub remote_addr: Option, + pub service: String, +} + +impl std::fmt::Debug for ServerRequest { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + f.debug_struct("ServerRequest") + .field("method", &self.method) + .field("path", &self.path) + .field("query", &self.query) + .field("headers", &self.headers) + .field("body", &"") + .field("guard", &self.guard) + .field("output_cap", &self.output_cap) + .field("auth", &self.auth) + .field("endpoint", &self.endpoint) + .field("remote_addr", &self.remote_addr) + .field("service", &self.service) + .finish() + } +} + +impl ServerRequest { + pub fn new( + method: impl Into, + path: impl Into, + endpoint: EndpointKind, + service: impl Into, + body: Box, + ) -> Self { + Self { + method: method.into(), + path: path.into(), + query: None, + headers: HeaderMap::new(), + body, + guard: StreamGuard::new(), + output_cap: DEFAULT_OUTPUT_CAP_BYTES as u64, + auth: None, + endpoint, + remote_addr: None, + service: service.into(), + } + } + + pub fn with_query(mut self, query: impl Into) -> Self { + self.query = Some(query.into()); + self + } + + pub fn with_header(mut self, key: impl Into, value: impl Into) -> Self { + self.headers.insert(key.into(), value.into()); + self + } + + pub fn with_auth(mut self, auth: AuthContext) -> Self { + self.auth = Some(auth); + self + } + + pub fn with_output_cap(mut self, cap_bytes: u64) -> Self { + self.output_cap = cap_bytes; + self + } + + pub fn with_remote_addr(mut self, addr: impl Into) -> Self { + self.remote_addr = Some(addr.into()); + self + } +} + +#[derive(Debug, Clone)] +pub struct StreamGuard { + cancelled: Arc, +} + +impl Default for StreamGuard { + fn default() -> Self { + Self::new() + } +} + +impl StreamGuard { + pub fn new() -> Self { + Self { + cancelled: Arc::new(AtomicBool::new(false)), + } + } + + pub fn cancel(&self) { + self.cancelled.store(true, Ordering::SeqCst); + } + + pub fn is_cancelled(&self) -> bool { + self.cancelled.load(Ordering::SeqCst) + } + + pub fn handle(&self) -> StreamGuardHandle { + StreamGuardHandle { + cancelled: self.cancelled.clone(), + } + } +} + +#[derive(Debug, Clone)] +pub struct StreamGuardHandle { + cancelled: Arc, +} + +impl StreamGuardHandle { + pub fn is_cancelled(&self) -> bool { + self.cancelled.load(Ordering::SeqCst) + } + + pub fn cancel(&self) { + self.cancelled.store(true, Ordering::SeqCst); + } +} diff --git a/src/server/stream.rs b/src/server/stream.rs new file mode 100644 index 0000000..f33531d --- /dev/null +++ b/src/server/stream.rs @@ -0,0 +1,251 @@ +use std::path::PathBuf; + +use tokio::io::{AsyncBufRead, AsyncRead, AsyncReadExt, AsyncWrite, AsyncWriteExt, BufReader}; +use tokio::process::Child; + +use crate::server::error::{ServerError, ServerResult}; +use crate::server::limits::DEFAULT_OUTPUT_CAP_BYTES; + +pub struct ChildPackReader { + reader: R, + child: Option, + finished: bool, +} + +impl ChildPackReader { + pub fn new(reader: R, child: Option) -> Self { + Self { + reader, + child, + finished: false, + } + } + + pub fn into_inner(self) -> (R, Option) { + (self.reader, self.child) + } + + pub fn finished(&self) -> bool { + self.finished + } + + pub fn child_handle(&mut self) -> Option<&mut Child> { + self.child.as_mut() + } + + pub async fn next_chunk(&mut self, buf: &mut [u8]) -> ServerResult { + if self.finished { + return Ok(0); + } + let n = self + .reader + .read(buf) + .await + .map_err(|source| ServerError::Io { + path: PathBuf::new(), + source, + })?; + if n == 0 { + self.finished = true; + } + Ok(n) + } + + pub async fn read_to_end_capped(&mut self, cap: Option) -> ServerResult { + let cap = cap.unwrap_or(DEFAULT_OUTPUT_CAP_BYTES as u64); + let mut out: Vec = Vec::new(); + let mut buf = vec![0u8; 32 * 1024]; + loop { + let n = self.next_chunk(&mut buf).await?; + if n == 0 { + self.finished = true; + return Ok(StreamOutcome::Eof); + } + let projected = (out.len() as u64) + (n as u64); + if projected > cap { + return Err(ServerError::PayloadTooLarge { + cap_bytes: cap, + actual_bytes: projected, + }); + } + out.extend_from_slice(&buf[..n]); + if (out.len() as u64) >= cap { + return Ok(StreamOutcome::Bytes(out.len() as u64)); + } + } + } + + pub async fn collect_pack_stats(&mut self, cap: Option) -> ServerResult { + let mut stats = PackStats::default(); + let mut buf = vec![0u8; 32 * 1024]; + loop { + let n = self.next_chunk(&mut buf).await?; + if n == 0 { + break; + } + stats.bytes_streamed = stats.bytes_streamed.saturating_add(n as u64); + if let Some(limit) = cap + && stats.bytes_streamed > limit + { + return Err(ServerError::PayloadTooLarge { + cap_bytes: limit, + actual_bytes: stats.bytes_streamed, + }); + } + } + Ok(stats) + } +} + +pub struct ChildPackWriter { + writer: W, + child: Option, + finished: bool, +} + +impl ChildPackWriter { + pub fn new(writer: W, child: Option) -> Self { + Self { + writer, + child, + finished: false, + } + } + + pub fn into_inner(self) -> (W, Option) { + (self.writer, self.child) + } + + pub fn finished(&self) -> bool { + self.finished + } + + pub fn child_handle(&mut self) -> Option<&mut Child> { + self.child.as_mut() + } + + pub async fn write_chunk(&mut self, buf: &[u8]) -> ServerResult { + if self.finished { + return Ok(0); + } + let n = self + .writer + .write(buf) + .await + .map_err(|source| ServerError::Io { + path: PathBuf::new(), + source, + })?; + Ok(n) + } + + pub async fn copy_from(&mut self, mut reader: R) -> ServerResult + where + R: AsyncRead + Unpin + Send, + { + let mut buf = vec![0u8; 32 * 1024]; + let mut total: u64 = 0; + loop { + let n = reader + .read(&mut buf) + .await + .map_err(|source| ServerError::Io { + path: PathBuf::new(), + source, + })?; + if n == 0 { + break; + } + total = total.saturating_add(n as u64); + let written = self.write_chunk(&buf[..n]).await?; + if written == 0 { + self.finished = true; + return Ok(StreamOutcome::Bytes(total)); + } + } + self.writer + .flush() + .await + .map_err(|source| ServerError::Io { + path: PathBuf::new(), + source, + })?; + Ok(StreamOutcome::Bytes(total)) + } + + pub async fn finish(mut self) -> ServerResult<()> { + self.writer + .shutdown() + .await + .map_err(|source| ServerError::Io { + path: PathBuf::new(), + source, + })?; + self.finished = true; + Ok(()) + } +} + +#[derive(Debug, Clone, Default, PartialEq, Eq)] +pub struct PackStats { + pub bytes_streamed: u64, + pub objects_reported: u64, +} + +impl PackStats { + pub fn is_empty(&self) -> bool { + self.bytes_streamed == 0 && self.objects_reported == 0 + } +} + +#[derive(Debug, Clone, Default)] +pub struct IngestReport { + pub stats: PackStats, + pub refs_updated: Vec, +} + +impl IngestReport { + pub fn new(stats: PackStats, refs_updated: Vec) -> Self { + Self { + stats, + refs_updated, + } + } + + pub fn empty() -> Self { + Self::default() + } +} + +#[derive(Debug, Clone, PartialEq, Eq)] +pub enum StreamOutcome { + Eof, + Bytes(u64), +} + +impl StreamOutcome { + pub fn bytes(&self) -> Option { + match self { + Self::Eof => None, + Self::Bytes(n) => Some(*n), + } + } +} + +pub type BoxedAsyncRead = Box; + +pub type BoxedAsyncWrite = Box; + +pub type BoxedAsyncBufRead = Box; + +pub fn empty_reader() -> BoxedAsyncRead { + Box::new(tokio::io::empty()) +} + +pub fn sink_writer() -> BoxedAsyncWrite { + Box::new(tokio::io::sink()) +} + +pub fn buf_empty_reader() -> BoxedAsyncBufRead { + Box::new(BufReader::new(tokio::io::empty())) +} diff --git a/tests/server.rs b/tests/server.rs new file mode 100644 index 0000000..e818d5b --- /dev/null +++ b/tests/server.rs @@ -0,0 +1,392 @@ +use std::sync::{Arc, Mutex, MutexGuard, PoisonError}; +use std::time::Duration; + +use tokio::io::AsyncReadExt; + +use gitbaby::server::{ + AuthContext, AuthScheme, ChildPackReader, ChildPackWriter, EndpointKind, IngestReport, + PackStats, Principal, RefUpdateBatch, RefUpdateEntry, ServerBackend, ServerError, + ServerRequest, StreamGuard, StreamOutcome, buf_empty_reader, empty_reader, +}; + +static ENV_LOCK: Mutex<()> = Mutex::new(()); + +fn lock_or_recover() -> MutexGuard<'static, ()> { + match ENV_LOCK.lock() { + Ok(g) => g, + Err(PoisonError { .. }) => panic!("env lock poisoned"), + } +} + +#[derive(Debug)] +struct PlaceholderBackend; + +#[async_trait::async_trait] +impl ServerBackend for PlaceholderBackend { + async fn authorize(&self, _req: &ServerRequest) -> Result { + Err(ServerError::Unimplemented( + "server::ServerBackend::authorize", + )) + } + + async fn list_refs_stream( + &self, + _req: &ServerRequest, + ) -> Result, ServerError> { + Err(ServerError::Unimplemented( + "server::ServerBackend::list_refs_stream", + )) + } + + async fn produce_pack_stream( + &self, + _req: &ServerRequest, + _wants: &[String], + ) -> Result, ServerError> { + Err(ServerError::Unimplemented( + "server::ServerBackend::produce_pack_stream", + )) + } + + async fn ingest_pack_stream( + &self, + _req: &ServerRequest, + _batch: RefUpdateBatch, + ) -> Result { + Err(ServerError::Unimplemented( + "server::ServerBackend::ingest_pack_stream", + )) + } + + async fn lfs_get_stream( + &self, + _req: &ServerRequest, + _oid: &str, + ) -> Result, ServerError> { + Err(ServerError::Unimplemented( + "server::ServerBackend::lfs_get_stream", + )) + } + + async fn lfs_put_stream( + &self, + _req: &ServerRequest, + _oid: &str, + _declared_size: u64, + ) -> Result { + Err(ServerError::Unimplemented( + "server::ServerBackend::lfs_put_stream", + )) + } +} + +fn make_request(method: &str, path: &str, endpoint: EndpointKind) -> ServerRequest { + ServerRequest::new(method, path, endpoint, "test-service", empty_reader()) +} + +#[test] +fn auth_scheme_str_round_trip() { + let _guard = lock_or_recover(); + assert_eq!(AuthScheme::Anonymous.as_str(), "anonymous"); + assert_eq!(AuthScheme::Basic.as_str(), "basic"); + assert_eq!(AuthScheme::Bearer.as_str(), "bearer"); + assert_eq!(AuthScheme::Ssh.as_str(), "ssh"); + assert_eq!(AuthScheme::default(), AuthScheme::Anonymous); +} + +#[test] +fn auth_context_anonymous_has_no_principal() { + let ctx = AuthContext::anonymous(); + assert_eq!(ctx.scheme, AuthScheme::Anonymous); + assert!(ctx.principal.is_none()); + assert!(ctx.raw.is_none()); +} + +#[test] +fn auth_context_with_principal_keeps_scheme_and_principal() { + let ctx = AuthContext::with_principal( + AuthScheme::Bearer, + Principal::new("alice", AuthScheme::Bearer), + ); + assert_eq!(ctx.scheme, AuthScheme::Bearer); + let p = ctx.principal.expect("principal present"); + assert_eq!(p.name, "alice"); + assert_eq!(p.scheme, AuthScheme::Bearer); +} + +#[test] +fn ref_update_batch_push_and_count() { + let batch = RefUpdateBatch::new() + .push( + "refs/heads/main", + "0000000000000000000000000000000000000000", + None, + ) + .push( + "refs/heads/dev", + "1111111111111111111111111111111111111111", + Some("0000000000000000000000000000000000000000".to_string()), + ); + + assert_eq!(batch.len(), 2); + assert!(!batch.is_empty()); + assert_eq!(batch.updates.len(), 2); + assert_eq!(batch.updates[0].name, "refs/heads/main"); + assert!(batch.updates[0].old_oid.is_none()); + assert!(batch.updates[1].old_oid.is_some()); +} + +#[test] +fn stream_guard_cancel_is_observable_via_handle() { + let guard = StreamGuard::new(); + let handle = guard.handle(); + assert!(!guard.is_cancelled()); + assert!(!handle.is_cancelled()); + + guard.cancel(); + assert!(guard.is_cancelled()); + assert!(handle.is_cancelled()); + + handle.cancel(); + assert!(guard.is_cancelled()); +} + +#[test] +fn server_request_builder_populates_fields() { + let req = make_request("GET", "/info/refs", EndpointKind::InfoRefs) + .with_query("service=git-upload-pack") + .with_header( + "Content-Type", + "application/x-git-upload-pack-advertisement", + ) + .with_auth(AuthContext::anonymous()) + .with_output_cap(1024) + .with_remote_addr("127.0.0.1:9418"); + + assert_eq!(req.method, "GET"); + assert_eq!(req.path, "/info/refs"); + assert_eq!(req.query.as_deref(), Some("service=git-upload-pack")); + assert_eq!( + req.headers.get("Content-Type").map(String::as_str), + Some("application/x-git-upload-pack-advertisement") + ); + assert_eq!(req.output_cap, 1024); + assert_eq!(req.remote_addr.as_deref(), Some("127.0.0.1:9418")); + assert_eq!(req.endpoint, EndpointKind::InfoRefs); + assert_eq!(req.service, "test-service"); +} + +#[test] +fn stream_outcome_bytes_round_trip() { + assert_eq!(StreamOutcome::Eof.bytes(), None); + assert_eq!(StreamOutcome::Bytes(7).bytes(), Some(7)); +} + +#[test] +fn pack_stats_default_is_empty() { + let s = PackStats::default(); + assert!(s.is_empty()); + assert_eq!(s.bytes_streamed, 0); + assert_eq!(s.objects_reported, 0); +} + +#[test] +fn ingest_report_empty_is_default() { + let r = IngestReport::empty(); + assert!(r.refs_updated.is_empty()); + assert!(r.stats.is_empty()); +} + +#[test] +fn empty_readers_and_helpers_are_present() { + let r1 = empty_reader(); + let r2 = buf_empty_reader(); + let rt = tokio::runtime::Builder::new_current_thread() + .enable_all() + .build() + .expect("rt"); + let n1 = rt.block_on(async { + let mut r = r1; + let mut buf = [0u8; 4]; + r.read(&mut buf).await.unwrap_or(0) + }); + let n2 = rt.block_on(async { + let mut r = r2; + let mut buf = [0u8; 4]; + r.read(&mut buf).await.unwrap_or(0) + }); + assert_eq!(n1, 0); + assert_eq!(n2, 0); +} + +#[tokio::test] +async fn backend_placeholders_report_unimplemented() { + let backend = PlaceholderBackend; + let req = make_request("GET", "/info/refs", EndpointKind::InfoRefs); + let err = backend + .authorize(&req) + .await + .expect_err("placeholder must return Err"); + match err { + ServerError::Unimplemented(msg) => assert!(msg.contains("authorize")), + other => panic!("unexpected error variant: {:?}", other), + } +} + +#[tokio::test] +async fn child_pack_reader_reads_bytes_then_reports_eof() { + use std::io::Cursor; + let data: Vec = (0u8..16).collect(); + let reader = ChildPackReader::new(tokio::io::BufReader::new(Cursor::new(data.clone())), None); + + let mut total: u64 = 0; + let mut reader = reader; + loop { + let mut buf = [0u8; 4]; + let n = reader.next_chunk(&mut buf).await.expect("read ok"); + if n == 0 { + break; + } + total = total.saturating_add(n as u64); + } + assert_eq!(total, 16); + assert!(reader.finished()); +} + +#[tokio::test] +async fn child_pack_reader_collects_pack_stats_and_enforces_cap() { + use std::io::Cursor; + let data: Vec = (0u8..32).collect(); + let mut reader = ChildPackReader::new(tokio::io::BufReader::new(Cursor::new(data)), None); + let stats = reader + .collect_pack_stats(Some(64)) + .await + .expect("within cap"); + assert_eq!(stats.bytes_streamed, 32); + + let data: Vec = (0u8..32).collect(); + let mut reader = ChildPackReader::new(tokio::io::BufReader::new(Cursor::new(data)), None); + let err = reader + .collect_pack_stats(Some(8)) + .await + .expect_err("cap should reject"); + match err { + ServerError::PayloadTooLarge { .. } => {} + other => panic!("unexpected error: {:?}", other), + } +} + +#[tokio::test] +async fn child_pack_writer_copies_reader_to_writer_and_finishes() { + use std::io::Cursor; + use tokio::io::BufWriter; + + let src: Vec = (0u8..8).collect(); + let mut src_reader = tokio::io::BufReader::new(Cursor::new(src)); + let dst: Vec = Vec::new(); + let buf_dst = BufWriter::new(Cursor::new(dst)); + + let mut writer = ChildPackWriter::new(buf_dst, None); + let outcome = writer.copy_from(&mut src_reader).await.expect("copy ok"); + match outcome { + StreamOutcome::Bytes(n) => assert_eq!(n, 8), + StreamOutcome::Eof => panic!("expected bytes outcome"), + } + writer.finish().await.expect("finish ok"); +} + +#[tokio::test] +async fn server_request_body_streams_bytes() { + use std::io::Cursor; + let body_data: Vec = b"PACK".to_vec(); + let req = ServerRequest::new( + "POST", + "/git-receive-pack", + EndpointKind::GitReceivePack, + "test", + Box::new(Cursor::new(body_data.clone())), + ); + + let mut body = req.body; + let mut got = Vec::new(); + body.read_to_end(&mut got).await.expect("read body"); + assert_eq!(got, body_data); +} + +#[test] +fn ref_update_entry_serialises_oid_strings_not_objectid() { + let entry = RefUpdateEntry { + name: "refs/heads/x".into(), + new_oid: "deadbeef".into(), + old_oid: Some("feedface".into()), + }; + assert_eq!(entry.new_oid, "deadbeef"); + assert_eq!(entry.old_oid.as_deref(), Some("feedface")); +} + +#[test] +fn endpoint_kind_str_matches_module_taxonomy() { + let _guard = lock_or_recover(); + assert_eq!(EndpointKind::GitUploadPack.as_str(), "git-upload-pack"); + assert_eq!(EndpointKind::GitReceivePack.as_str(), "git-receive-pack"); + assert_eq!(EndpointKind::LfsBatch.as_str(), "lfs-batch"); + assert_eq!(EndpointKind::LfsObjects.as_str(), "lfs-objects"); + assert_eq!(EndpointKind::InfoRefs.as_str(), "info-refs"); + assert_eq!(EndpointKind::SmartHttp.as_str(), "smart-http"); + assert_eq!(EndpointKind::Custom.as_str(), "custom"); +} + +#[test] +fn guard_clone_is_independent_until_cancel() { + let g = StreamGuard::new(); + let g2 = g.clone(); + g.cancel(); + assert!(g.is_cancelled()); + assert!(g2.is_cancelled()); +} + +#[test] +fn auth_context_default_is_anonymous() { + let ctx = AuthContext::default(); + assert_eq!(ctx.scheme, AuthScheme::Anonymous); + assert!(ctx.principal.is_none()); +} + +#[test] +fn server_request_default_output_cap_matches_command_cap() { + let req = make_request("GET", "/", EndpointKind::Custom); + assert_eq!( + req.output_cap, + gitbaby::server::DEFAULT_OUTPUT_CAP_BYTES as u64 + ); +} + +#[test] +fn ref_update_batch_default_is_empty() { + let batch = RefUpdateBatch::default(); + assert!(batch.is_empty()); + assert_eq!(batch.len(), 0); +} + +#[test] +fn endpoint_kind_equality_works() { + assert_eq!(EndpointKind::InfoRefs, EndpointKind::InfoRefs); + assert_ne!(EndpointKind::InfoRefs, EndpointKind::SmartHttp); +} + +#[tokio::test] +async fn stream_guard_cancel_during_async_does_not_panic() { + let guard = Arc::new(StreamGuard::new()); + let g2 = guard.clone(); + + let h = tokio::spawn(async move { + tokio::time::sleep(Duration::from_millis(200)).await; + g2.cancel(); + }); + + tokio::time::sleep(Duration::from_millis(50)).await; + assert!(!guard.is_cancelled()); + + h.await.expect("task ok"); + assert!(guard.is_cancelled()); +}