From f4828c639300f027fb938dbd7fdbe47c2f74ff9a Mon Sep 17 00:00:00 2001 From: DanConwayDev Date: Mon, 29 Jun 2026 15:17:31 +0100 Subject: [PATCH] fix(http): stream git smart HTTP responses Return channel-backed response bodies for receive-pack and upload-pack so Git stdout reaches clients as soon as it is produced instead of after the subprocess exits. This fixes a production issue seen when pushing large repositories with ngit-cli: large receive-pack pushes could sit silent while Git resolved deltas and checked connectivity, causing libgit2 clients to hit their per-recv timeout instead of observing sideband progress. The streaming task now owns subprocess cleanup, stderr draining, lifecycle locks, metrics, and post-push purgatory promotion. Add dedicated /prs/ regression coverage to prove receive-pack stdout streams before EOF while zero-ref /prs/ cleanup waits until the in-flight push finishes. Fixes nostr:nevent1qqsyjly2u925qdc59cmxlxarvwagnr7pd6gpnr776x2mnfz6mdhn33cpz3mhxue69uhhyetvv9ujumn8d96zuer9wcj596eq --- src/git/handlers.rs | 521 +++++++++++++++++++++++-------- src/git/mod.rs | 26 ++ src/grasp06/fetch.rs | 79 ++++- src/grasp06/receive.rs | 509 ++++++++++++++++++------------ src/http/mod.rs | 98 ++++-- src/metrics/mod.rs | 7 +- src/purgatory/promotion_hooks.rs | 34 +- tests/git_response_streaming.rs | 300 ++++++++++++++++++ 8 files changed, 1198 insertions(+), 376 deletions(-) create mode 100644 tests/git_response_streaming.rs diff --git a/src/git/handlers.rs b/src/git/handlers.rs index d1470e1..5735776 100644 --- a/src/git/handlers.rs +++ b/src/git/handlers.rs @@ -2,25 +2,33 @@ //! //! This module implements the HTTP handlers for Git Smart HTTP protocol. -use http_body_util::Full; -use hyper::{body::Bytes, Response, StatusCode}; +use futures_util::stream; +use http_body_util::{BodyExt, StreamBody}; +use hyper::{body::Bytes, body::Frame, Response, StatusCode}; use nostr_relay_builder::LocalRelay; use nostr_sdk::prelude::*; use std::collections::HashSet; +use std::io; use std::path::PathBuf; use std::sync::Arc; use tokio::io::{AsyncReadExt, AsyncWriteExt}; +use tokio::sync::mpsc; use tracing::{debug, error, info, warn}; use super::protocol::{GitService, PktLine}; use super::subprocess::GitSubprocess; +use super::{full_body, GitResponseBody}; use crate::git::authorization::{authorize_push, parse_pushed_refs}; use crate::git::sync::{process_newly_available_git_data, PurgatoryPromotionHooks}; +use crate::metrics::Metrics; use crate::nostr::lifecycle::LifecycleReadGuard; use crate::nostr::SharedDatabase; use crate::purgatory::Purgatory; +pub(crate) const STREAM_CHANNEL_DEPTH: usize = 8; +const STREAM_CHUNK_SIZE: usize = 8 * 1024; + /// Handle GET /info/refs?service=git-{upload,receive}-pack /// /// This advertises the repository's refs to the client. @@ -28,7 +36,7 @@ pub async fn handle_info_refs( repo_path: PathBuf, service: GitService, git_protocol: Option<&str>, -) -> Result>, GitError> { +) -> Result, GitError> { debug!( "Handling info/refs for {:?} with service {:?}", repo_path, service @@ -96,7 +104,7 @@ pub async fn handle_info_refs( .status(StatusCode::OK) .header("content-type", service.advertisement_content_type()) .header("cache-control", "no-cache") - .body(Full::new(Bytes::from(response_body))) + .body(full_body(response_body)) .unwrap()) } @@ -178,7 +186,22 @@ pub(crate) fn build_git_protocol_error_response( service: GitService, error_message: &str, request_body: Option<&[u8]>, -) -> Response> { +) -> Response { + let err_pktline = encode_err_pktline(service, error_message, request_body); + + Response::builder() + .status(StatusCode::OK) + .header("content-type", service.result_content_type()) + .header("cache-control", "no-cache") + .body(full_body(err_pktline)) + .unwrap() +} + +fn encode_err_pktline( + service: GitService, + error_message: &str, + request_body: Option<&[u8]>, +) -> Vec { // Format: "ERR \n" let err_content = format!("ERR {}\n", error_message.trim()); @@ -189,21 +212,26 @@ pub(crate) fn build_git_protocol_error_response( .map(client_negotiated_sideband_64k) .unwrap_or(false); - let err_pktline = if use_sideband { + if use_sideband { let mut framed = Vec::with_capacity(1 + err_content.len()); framed.push(0x03); // band 3 = error framed.extend_from_slice(err_content.as_bytes()); PktLine::data(framed).encode() } else { PktLine::data(err_content.as_bytes()).encode() - }; + } +} - Response::builder() - .status(StatusCode::OK) - .header("content-type", service.result_content_type()) - .header("cache-control", "no-cache") - .body(Full::new(Bytes::from(err_pktline))) - .unwrap() +pub(crate) fn err_pktline_frame( + service: GitService, + error_message: &str, + request_body: Option<&[u8]>, +) -> Frame { + Frame::data(Bytes::from(encode_err_pktline( + service, + error_message, + request_body, + ))) } /// Check if a git process failure is a protocol error (vs transport error). @@ -222,12 +250,105 @@ pub(crate) fn is_git_protocol_error(exit_code: Option, stderr: &[u8]) -> bo exit_code == Some(128) && !stderr.is_empty() } +/// Build the common channel-backed Git response used by streaming handlers. +/// +/// The handler returns this response before the Git child has necessarily +/// exited. A background task sends `Frame` values into `rx`; dropping the +/// sender closes the HTTP body. Kept `pub(crate)` so `/prs/` receive-pack can +/// use the same wire shape as the standard endpoints. +pub(crate) fn streaming_response( + service: GitService, + rx: mpsc::Receiver, io::Error>>, +) -> Response { + let body_stream = stream::unfold(rx, |mut rx| async { + rx.recv().await.map(|item| (item, rx)) + }); + let body = BodyExt::boxed(StreamBody::new(body_stream)); + + Response::builder() + .status(StatusCode::OK) + .header("content-type", service.result_content_type()) + .header("cache-control", "no-cache") + .body(body) + .unwrap() +} + +/// Record Git metrics from either the request handler or the detached streaming +/// task without repeating the `Option` plumbing at every callsite. +pub(crate) fn record_git_operation(metrics: &Option>, operation: &str, status: &str) { + if let Some(metrics) = metrics { + metrics.record_git_operation(operation, status); + } +} + +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub(crate) enum PumpResult { + /// Stdout reached EOF. `sent_stdout` tells callers whether it is still safe + /// to synthesize a Git `ERR` pkt-line for a late process failure. + Eof { sent_stdout: bool }, + /// The response receiver was dropped, usually because the HTTP client + /// disconnected. Callers should stop the child process and clean up state. + ClientDisconnected, + /// Reading stdout failed. The read error has already been sent through the + /// body channel so Hyper can terminate the response stream. + ReadError, +} + +/// Forward Git stdout into the streaming response channel. +/// +/// This is intentionally only the stdout pump. The caller still owns process +/// termination, stderr collection, protocol-error classification, and any +/// endpoint-specific cleanup after the stream ends. +pub(crate) async fn pump_stdout_to_channel( + mut stdout: R, + tx: &mpsc::Sender, io::Error>>, +) -> PumpResult +where + R: tokio::io::AsyncRead + Unpin, +{ + let mut read_buf = [0_u8; STREAM_CHUNK_SIZE]; + let mut sent_stdout = false; + + loop { + match stdout.read(&mut read_buf).await { + Ok(0) => return PumpResult::Eof { sent_stdout }, + Ok(n) => { + sent_stdout = true; + if send_body_bytes(tx, read_buf[..n].to_vec()).await.is_err() { + return PumpResult::ClientDisconnected; + } + } + Err(e) => { + let _ = tx.send(Err(e)).await; + return PumpResult::ReadError; + } + } + } +} + +/// Drain Git stderr for later logging or Git protocol error synthesis. +/// +/// Streaming handlers run this concurrently with stdout pumping so the child +/// cannot block on a full stderr pipe while the HTTP body is still being read. +pub(crate) async fn read_stderr_to_end(mut stderr: R) -> Vec +where + R: tokio::io::AsyncRead + Unpin, +{ + let mut stderr_output = Vec::new(); + if let Err(e) = stderr.read_to_end(&mut stderr_output).await { + warn!("Failed to read git subprocess stderr: {}", e); + } + stderr_output +} + /// Handle POST /git-upload-pack (clone/fetch) pub async fn handle_upload_pack( repo_path: PathBuf, request_body: Bytes, git_protocol: Option<&str>, -) -> Result>, GitError> { + repo_lifecycle_guard: Option, + metrics: Option>, +) -> Result, GitError> { debug!("Handling upload-pack for {:?}", repo_path); if !repo_path.exists() { @@ -248,60 +369,101 @@ pub async fn handle_upload_pack( drop(stdin); } - // Read response from git's stdout - let mut output = Vec::new(); - let mut stderr_output = Vec::new(); - - if let Some(stdout) = git.take_stdout() { - let mut stdout = stdout; - stdout.read_to_end(&mut output).await.map_err(|e| { - error!("Failed to read git upload-pack stdout: {}", e); - GitError::IoError(e) - })?; - } - - if let Some(stderr) = git.take_stderr() { - let mut stderr = stderr; - stderr.read_to_end(&mut stderr_output).await.map_err(|e| { - error!("Failed to read git upload-pack stderr: {}", e); - GitError::IoError(e) - })?; - } - - // Wait for process - let status = git.wait().await.map_err(|e| { - error!("Failed to wait for git upload-pack process: {}", e); - GitError::IoError(e) + let stdout = git.take_stdout().ok_or_else(|| { + GitError::IoError(io::Error::new( + io::ErrorKind::BrokenPipe, + "git upload-pack stdout unavailable", + )) })?; + let stderr = git.take_stderr(); - if !status.success() { + // Stream upload-pack stdout as Git produces it instead of buffering the + // whole fetch response in memory. The detached task below owns the child + // until EOF so the request future can return the HTTP body immediately. + let (tx, rx) = mpsc::channel::, io::Error>>(STREAM_CHANNEL_DEPTH); + + tokio::spawn(async move { + stream_upload_pack_output(git, stdout, stderr, tx, repo_lifecycle_guard, metrics).await; + }); + + Ok(streaming_response(GitService::UploadPack, rx)) +} + +async fn stream_upload_pack_output( + mut git: GitSubprocess, + stdout: S, + stderr: Option, + tx: mpsc::Sender, io::Error>>, + repo_lifecycle_guard: Option, + metrics: Option>, +) where + S: tokio::io::AsyncRead + Unpin + Send + 'static, + E: tokio::io::AsyncRead + Unpin + Send + 'static, +{ + let stderr_task = stderr.map(|stderr| tokio::spawn(read_stderr_to_end(stderr))); + let pump_result = pump_stdout_to_channel(stdout, &tx).await; + + if !matches!(pump_result, PumpResult::Eof { .. }) { + let _ = git.kill().await; + } + + let status = match git.wait().await { + Ok(status) => status, + Err(e) => { + let _ = tx.send(Err(e)).await; + drop(repo_lifecycle_guard); + record_git_operation(&metrics, "clone", "error"); + return; + } + }; + + // Keep the repository lifecycle read lock until git-upload-pack exits so + // deletion/archive/restore cannot remove or replace the bare repository + // while Git is still reading it in the detached streaming task. + drop(repo_lifecycle_guard); + + let stderr_output = match stderr_task { + Some(task) => task.await.unwrap_or_default(), + None => Vec::new(), + }; + + if !status.success() && matches!(pump_result, PumpResult::Eof { sent_stdout: false }) { + record_git_operation(&metrics, "clone", "error"); let stderr_str = String::from_utf8_lossy(&stderr_output); + let msg = if stderr_str.trim().is_empty() { + format!("git upload-pack failed with code {:?}", status.code()) + } else { + stderr_str.to_string() + }; - // Check if this is a git protocol error (exit code 128 with stderr) - // Protocol errors should be returned as HTTP 200 with ERR pkt-line if is_git_protocol_error(status.code(), &stderr_output) { warn!( "Git upload-pack protocol error (returning ERR pkt-line): {}", stderr_str ); - return Ok(build_git_protocol_error_response( - GitService::UploadPack, - &stderr_str, - None, - )); + } else { + error!("Git upload-pack failed: {}", stderr_str); } - // Transport errors (spawn failures, signals, etc.) remain as HTTP 500 - error!("Git upload-pack failed: {}", stderr_str); - return Err(GitError::GitFailed(status.code())); + // The streaming response headers have already been sent. If Git failed + // before producing any stdout, surface a protocol-visible ERR pkt-line + // instead of silently completing an empty HTTP 200 body. + let _ = tx + .send(Ok(err_pktline_frame(GitService::UploadPack, &msg, None))) + .await; + } else if !status.success() { + record_git_operation(&metrics, "clone", "error"); + let stderr_str = String::from_utf8_lossy(&stderr_output); + error!( + "Git upload-pack failed after streaming stdout: {}", + stderr_str + ); + } else if matches!(pump_result, PumpResult::Eof { .. }) { + debug!("Git upload-pack stream completed successfully"); + record_git_operation(&metrics, "clone", "success"); + } else { + record_git_operation(&metrics, "clone", "error"); } - - Ok(Response::builder() - .status(StatusCode::OK) - .header("content-type", GitService::UploadPack.result_content_type()) - .header("cache-control", "no-cache") - .body(Full::new(Bytes::from(output))) - .unwrap()) } /// Handle POST /git-receive-pack (push) @@ -338,8 +500,9 @@ pub async fn handle_receive_pack( git_data_path: &str, git_protocol: Option<&str>, repo_lifecycle_guard: Option, - promotion_hooks: Option<&dyn PurgatoryPromotionHooks>, -) -> Result>, GitError> { + promotion_hooks: Option>, + metrics: Option>, +) -> Result, GitError> { debug!("Handling receive-pack for {:?}", repo_path); if !repo_path.exists() { @@ -366,6 +529,7 @@ pub async fn handle_receive_pack( Ok(auth_result) => { if !auth_result.authorized { warn!("Push rejected for {}: {}", identifier, auth_result.reason); + record_git_operation(&metrics, "push", "error"); return Ok(build_git_protocol_error_response( GitService::ReceivePack, &format!("authorisation failed: {}", auth_result.reason), @@ -383,6 +547,7 @@ pub async fn handle_receive_pack( } Err(e) => { warn!("Authorization check failed for {}: {}", identifier, e); + record_git_operation(&metrics, "push", "error"); return Ok(build_git_protocol_error_response( GitService::ReceivePack, &format!("authorisation failed: {}", e), @@ -404,55 +569,147 @@ pub async fn handle_receive_pack( drop(stdin); } - // Read response from git's stdout - let mut output = Vec::new(); - let mut stderr_output = Vec::new(); - - if let Some(stdout) = git.take_stdout() { - let mut stdout = stdout; - stdout.read_to_end(&mut output).await.map_err(|e| { - error!("Failed to read git receive-pack stdout: {}", e); - GitError::IoError(e) - })?; - } - - if let Some(stderr) = git.take_stderr() { - let mut stderr = stderr; - stderr.read_to_end(&mut stderr_output).await.map_err(|e| { - error!("Failed to read git receive-pack stderr: {}", e); - GitError::IoError(e) - })?; - } - - // Wait for process - let status = git.wait().await.map_err(|e| { - error!("Failed to wait for git receive-pack process: {}", e); - GitError::IoError(e) + let stdout = git.take_stdout().ok_or_else(|| { + GitError::IoError(io::Error::new( + io::ErrorKind::BrokenPipe, + "git receive-pack stdout unavailable", + )) })?; + let stderr = git.take_stderr(); + + // Do not buffer receive-pack stdout: for large pushes Git may spend tens of + // seconds resolving deltas/checking connectivity after the client finishes + // uploading. Streaming sideband progress keeps libgit2 clients from + // hitting their per-recv timeout during that otherwise-silent window. + let (tx, rx) = mpsc::channel::, io::Error>>(STREAM_CHANNEL_DEPTH); + let pushed_refs = parse_pushed_refs(&request_body); + let new_oids: HashSet = pushed_refs + .iter() + .filter(|(_, new_oid, _)| new_oid != "0000000000000000000000000000000000000000") + .map(|(_, new_oid, _)| new_oid.clone()) + .collect(); + let request_body_for_errors = request_body.clone(); + let identifier = identifier.to_owned(); + let git_data_path = git_data_path.to_owned(); + + tokio::spawn(async move { + stream_receive_pack_output( + git, + stdout, + stderr, + tx, + repo_path, + new_oids, + database, + relay, + identifier, + purgatory, + git_data_path, + request_body_for_errors, + repo_lifecycle_guard, + promotion_hooks, + metrics, + ) + .await; + }); + + Ok(streaming_response(GitService::ReceivePack, rx)) +} + +#[allow(clippy::too_many_arguments)] +async fn stream_receive_pack_output( + mut git: GitSubprocess, + stdout: S, + stderr: Option, + tx: mpsc::Sender, io::Error>>, + repo_path: PathBuf, + new_oids: HashSet, + database: SharedDatabase, + relay: LocalRelay, + identifier: String, + purgatory: Arc, + git_data_path: String, + request_body: Bytes, + repo_lifecycle_guard: Option, + promotion_hooks: Option>, + metrics: Option>, +) where + S: tokio::io::AsyncRead + Unpin + Send + 'static, + E: tokio::io::AsyncRead + Unpin + Send + 'static, +{ + let stderr_task = stderr.map(|stderr| tokio::spawn(read_stderr_to_end(stderr))); + let pump_result = pump_stdout_to_channel(stdout, &tx).await; + + if !matches!(pump_result, PumpResult::Eof { .. }) { + let _ = git.kill().await; + } + + let status = match git.wait().await { + Ok(status) => status, + Err(e) => { + let _ = tx.send(Err(e)).await; + drop(repo_lifecycle_guard); + record_git_operation(&metrics, "push", "error"); + return; + } + }; + + let stderr_output = match stderr_task { + Some(task) => task.await.unwrap_or_default(), + None => Vec::new(), + }; + + let sent_stdout = match pump_result { + PumpResult::Eof { sent_stdout } => sent_stdout, + PumpResult::ClientDisconnected | PumpResult::ReadError => { + drop(repo_lifecycle_guard); + record_git_operation(&metrics, "push", "error"); + return; + } + }; if !status.success() { drop(repo_lifecycle_guard); + record_git_operation(&metrics, "push", "error"); let stderr_str = String::from_utf8_lossy(&stderr_output); - // Check if this is a git protocol error (exit code 128 with stderr) - // Protocol errors should be returned as HTTP 200 with ERR pkt-line if is_git_protocol_error(status.code(), &stderr_output) { warn!( "Git receive-pack protocol error (returning ERR pkt-line): {}", stderr_str ); - return Ok(build_git_protocol_error_response( - GitService::ReceivePack, - &stderr_str, - Some(&request_body), - )); + if !sent_stdout { + let _ = tx + .send(Ok(err_pktline_frame( + GitService::ReceivePack, + &stderr_str, + Some(&request_body), + ))) + .await; + } + } else { + error!("Git receive-pack failed: {}", stderr_str); + if !sent_stdout { + let msg = if stderr_str.trim().is_empty() { + format!("git receive-pack failed with code {:?}", status.code()) + } else { + stderr_str.to_string() + }; + let _ = tx + .send(Ok(err_pktline_frame( + GitService::ReceivePack, + &msg, + Some(&request_body), + ))) + .await; + } } - - // Transport errors (spawn failures, signals, etc.) remain as HTTP 500 - error!("Git receive-pack failed: {}", stderr_str); - return Err(GitError::GitFailed(status.code())); + return; } + debug!("Git receive-pack stream completed successfully"); + record_git_operation(&metrics, "push", "success"); + // Release the repository lifecycle read lock once git-receive-pack itself // has finished. The lock's purpose is to keep deletion/archive/restore from // removing or replacing the bare repository while Git is actively reading or @@ -462,33 +719,19 @@ pub async fn handle_receive_pack( // the subprocess boundary would deadlock. drop(repo_lifecycle_guard); - // Process newly available git data using the unified function - // This handles: - // - Discovering satisfiable events from purgatory (state events and PR events) - // - Syncing OIDs to authorized owner repos - // - Aligning refs (+ setting HEAD) in all owner repos - // - Saving events to database - // - Notifying WebSocket subscribers - // - Removing from purgatory - // - // Parse pushed refs to collect new OIDs - let pushed_refs = parse_pushed_refs(&request_body); - let new_oids: HashSet = pushed_refs - .iter() - .filter(|(_, new_oid, _)| new_oid != "0000000000000000000000000000000000000000") - .map(|(_, new_oid, _)| new_oid.clone()) - .collect(); - - let git_data_path_buf = std::path::Path::new(git_data_path); - + // Git's receive-pack response has already been streamed verbatim, including + // its final flush. Run GRASP follow-up work before closing the HTTP body so + // push completion still implies local refs/HEAD are aligned, but treat those + // follow-up failures as internal/log-only rather than client-visible push + // rejections. match process_newly_available_git_data( &repo_path, &new_oids, &database, Some(&relay), &purgatory, - git_data_path_buf, - promotion_hooks, + std::path::Path::new(&git_data_path), + promotion_hooks.as_deref(), ) .await { @@ -516,16 +759,13 @@ pub async fn handle_receive_pack( ); } } +} - Ok(Response::builder() - .status(StatusCode::OK) - .header( - "content-type", - GitService::ReceivePack.result_content_type(), - ) - .header("cache-control", "no-cache") - .body(Full::new(Bytes::from(output))) - .unwrap()) +async fn send_body_bytes( + tx: &mpsc::Sender, io::Error>>, + bytes: Vec, +) -> Result<(), mpsc::error::SendError, io::Error>>> { + tx.send(Ok(Frame::data(Bytes::from(bytes)))).await } /// Errors that can occur in Git handlers @@ -591,7 +831,7 @@ mod tests { out } - async fn response_body_bytes(resp: Response>) -> Vec { + async fn response_body_bytes(resp: Response) -> Vec { resp.into_body() .collect() .await @@ -698,6 +938,39 @@ mod tests { assert_eq!(raw[4], b'E'); } + #[tokio::test] + async fn streaming_body_yields_stdout_before_eof() { + use tokio::io::AsyncWriteExt; + use tokio::time::{timeout, Duration}; + + let (mut writer, reader) = tokio::io::duplex(64); + let (tx, rx) = mpsc::channel::, io::Error>>(STREAM_CHANNEL_DEPTH); + + let pump_task = tokio::spawn(async move { pump_stdout_to_channel(reader, &tx).await }); + writer.write_all(b"progress chunk").await.unwrap(); + + let mut body = streaming_response(GitService::ReceivePack, rx).into_body(); + let frame = timeout(Duration::from_secs(1), body.frame()) + .await + .expect("stream should yield first stdout chunk before EOF") + .expect("stream should still be open") + .expect("stdout chunk should not be a body error"); + + let data = frame.into_data().expect("frame should contain data"); + assert_eq!(data, Bytes::from_static(b"progress chunk")); + + assert!( + !pump_task.is_finished(), + "stdout pump should still be waiting for EOF after yielding first chunk" + ); + + drop(writer); + assert_eq!( + pump_task.await.unwrap(), + PumpResult::Eof { sent_stdout: true } + ); + } + #[tokio::test] async fn receive_pack_err_without_body_is_not_wrapped() { // No request body → conservative path: do not wrap. diff --git a/src/git/mod.rs b/src/git/mod.rs index f82b9e9..f912865 100644 --- a/src/git/mod.rs +++ b/src/git/mod.rs @@ -24,10 +24,36 @@ pub mod protocol; pub mod subprocess; pub mod sync; +use std::convert::Infallible; use std::path::{Path, PathBuf}; use std::process::Command; + +use http_body_util::combinators::BoxBody; +use http_body_util::{BodyExt, Full}; +use hyper::body::Bytes; use tracing::{debug, info}; +/// Unified HTTP response body for every route served by ngit-grasp. +/// +/// This boxed body can hold either a buffered [`Full`] payload (for +/// small responses such as advertisements, protocol-error pkt-lines, JSON +/// pages, and static assets) or a streaming body backed by a channel (for live +/// `git-receive-pack` / `git-upload-pack` stdout). Using one body type keeps +/// the Hyper service response type unified. +pub type GitResponseBody = BoxBody; + +/// Wrap an in-memory payload as a one-shot [`GitResponseBody`]. +pub fn full_body(bytes: impl Into) -> GitResponseBody { + Full::new(bytes.into()) + .map_err(|never: Infallible| match never {}) + .boxed() +} + +/// Explicit empty response body. +pub fn empty_body() -> GitResponseBody { + full_body(Bytes::new()) +} + /// Parse a Git repository path from URL components /// /// Converts //.git/* to a filesystem path diff --git a/src/grasp06/fetch.rs b/src/grasp06/fetch.rs index 2f96cce..fb3a55d 100644 --- a/src/grasp06/fetch.rs +++ b/src/grasp06/fetch.rs @@ -12,18 +12,20 @@ //! //! Receive-pack lives in [`crate::grasp06::receive`]. -use http_body_util::Full; use hyper::body::Bytes; use hyper::Response; use std::path::Path; use std::process::Command; +use std::sync::Arc; use tempfile::TempDir; use tracing::{debug, warn}; use crate::git::handlers::{handle_info_refs, handle_upload_pack, GitError}; use crate::git::protocol::GitService; +use crate::git::GitResponseBody; use crate::grasp06::endpoint::PrsUrl; use crate::grasp06::paths::prs_repo_path; +use crate::metrics::Metrics; /// Handle `GET /prs//.git/info/refs?service=...`. /// @@ -38,7 +40,7 @@ pub async fn handle_prs_info_refs( git_data_path: &str, service: GitService, git_protocol: Option<&str>, -) -> Result>, GitError> { +) -> Result, GitError> { let real_repo = prs_repo_path( Path::new(git_data_path), &prs.submitter.to_hex(), @@ -75,7 +77,8 @@ pub async fn handle_prs_upload_pack( git_data_path: &str, body: Bytes, git_protocol: Option<&str>, -) -> Result>, GitError> { + metrics: Option>, +) -> Result, GitError> { let real_repo = prs_repo_path( Path::new(git_data_path), &prs.submitter.to_hex(), @@ -87,7 +90,7 @@ pub async fn handle_prs_upload_pack( "/prs/ upload-pack: real repo found at {} — delegating", real_repo.display() ); - return handle_upload_pack(real_repo, body, git_protocol).await; + return handle_upload_pack(real_repo, body, git_protocol, None, metrics).await; } debug!( @@ -96,10 +99,74 @@ pub async fn handle_prs_upload_pack( prs.identifier ); let temp = init_empty_bare_repo()?; + handle_prs_upload_pack_buffered(temp, body, git_protocol).await +} + +async fn handle_prs_upload_pack_buffered( + temp: TempDir, + request_body: Bytes, + git_protocol: Option<&str>, +) -> Result, GitError> { + use crate::git::full_body; + use crate::git::handlers::{build_git_protocol_error_response, is_git_protocol_error}; + use crate::git::subprocess::GitSubprocess; + use hyper::StatusCode; + use tokio::io::{AsyncReadExt, AsyncWriteExt}; + let repo_path = temp.path().to_path_buf(); - let response = handle_upload_pack(repo_path, body, git_protocol).await; + let mut git = GitSubprocess::spawn(GitService::UploadPack, &repo_path, false, git_protocol) + .map_err(GitError::ProcessSpawnFailed)?; + + if let Some(mut stdin) = git.take_stdin() { + stdin + .write_all(&request_body) + .await + .map_err(GitError::IoError)?; + drop(stdin); + } + + let mut output = Vec::new(); + let mut stderr_output = Vec::new(); + + if let Some(mut stdout) = git.take_stdout() { + stdout + .read_to_end(&mut output) + .await + .map_err(GitError::IoError)?; + } + + if let Some(mut stderr) = git.take_stderr() { + stderr + .read_to_end(&mut stderr_output) + .await + .map_err(GitError::IoError)?; + } + + let status = git.wait().await.map_err(GitError::IoError)?; + + // Keep the TempDir alive until git has fully exited. The standard + // upload-pack handler streams in a detached task; using that for this + // synthesized repo would race with dropping `temp` and deleting the repo. drop(temp); - response + + if !status.success() { + let stderr_str = String::from_utf8_lossy(&stderr_output); + if is_git_protocol_error(status.code(), &stderr_output) { + return Ok(build_git_protocol_error_response( + GitService::UploadPack, + &stderr_str, + None, + )); + } + return Err(GitError::GitFailed(status.code())); + } + + Ok(Response::builder() + .status(StatusCode::OK) + .header("content-type", GitService::UploadPack.result_content_type()) + .header("cache-control", "no-cache") + .body(full_body(output)) + .unwrap()) } /// Create a fresh empty bare repo in a temp directory. diff --git a/src/grasp06/receive.rs b/src/grasp06/receive.rs index e55e0c2..b634957 100644 --- a/src/grasp06/receive.rs +++ b/src/grasp06/receive.rs @@ -27,50 +27,65 @@ //! own ref locking handles intra-push concurrency, and the //! `in_flight` counter is what off-push cleanup paths consult to know //! a push is active. -//! 4. Runs `git-receive-pack` against the repo, mirroring the subprocess -//! plumbing in [`crate::git::handlers::handle_receive_pack`]. -//! 5. For each accepted `refs/nostr/` ref, re-runs the shared -//! pre-validation as a **race safety net** — an event for one of the -//! pushed ids may have arrived via WebSocket during the receive-pack -//! window. On mismatch the ref is deleted (and any populated purgatory -//! entry is dropped). When neither the DB nor purgatory knows about -//! the event a scoped PR placeholder is added so the standard -//! 30-minute purgatory sweep can clean it up if the event never +//! 4. Starts `git-receive-pack`, writes the full request body to the child, and +//! immediately returns an HTTP response whose body is backed by a bounded +//! channel. From this point on Hyper can stream stdout to the client while a +//! detached task owns the subprocess, stderr, and all post-push work. +//! 5. In the detached task, each stdout chunk is forwarded as it is read. If Git +//! exits with a protocol-level error before writing stdout, the task sends a +//! Git `ERR` pkt-line through the same stream so clients still see a normal +//! receive-pack failure. If stdout has already been sent, the task cannot +//! safely append a late protocol error without corrupting the Git stream, so +//! it logs and performs cleanup instead. +//! 6. After a successful receive-pack, for each accepted +//! `refs/nostr/` ref, re-runs the shared pre-validation as a +//! **race safety net** — an event for one of the pushed ids may have arrived +//! via WebSocket during the receive-pack window. On mismatch the ref is +//! deleted (and any populated purgatory entry is dropped). When neither the +//! DB nor purgatory knows about the event a scoped PR placeholder is added so +//! the standard 30-minute purgatory sweep can clean it up if the event never //! arrives. -//! 6. Re-acquires the per-path mutex briefly, decrements `in_flight`, -//! and — if no other push is in flight and the repo has zero refs -//! left — removes the bare directory. Always runs, including on -//! receive-pack protocol errors, so a failed push that has just -//! initialised an empty repo does not leak it. -//! 7. Triggers the standard purgatory-release path via -//! [`crate::git::sync::process_newly_available_git_data`] so PR events -//! already in purgatory waiting for these commits get promoted. +//! 7. The detached task re-acquires the per-path mutex briefly, decrements +//! `in_flight`, and — if no other push is in flight and the repo has zero +//! refs left — removes the bare directory. This runs on success, protocol +//! errors, client disconnects, and subprocess I/O failures, so a failed push +//! that has just initialised an empty repo does not leak it. +//! 8. Finally, on successful pushes where the repo still exists, the detached +//! task triggers the standard purgatory-release path via +//! [`crate::git::sync::process_newly_available_git_data`] so PR events already +//! in purgatory waiting for these commits get promoted. use std::collections::HashSet; +use std::io; use std::path::{Path, PathBuf}; use std::sync::atomic::{AtomicUsize, Ordering}; use std::sync::Arc; use dashmap::DashMap; -use http_body_util::Full; -use hyper::body::Bytes; -use hyper::{Response, StatusCode}; +use hyper::body::{Bytes, Frame}; +use hyper::Response; use nostr_relay_builder::LocalRelay; use nostr_sdk::prelude::*; use std::sync::Mutex; -use tokio::io::{AsyncReadExt, AsyncWriteExt}; +use tokio::io::AsyncWriteExt; +use tokio::sync::mpsc; use tracing::{debug, error, info, warn}; use crate::git::authorization::{ parse_pushed_refs, pre_validate_refs_nostr_push, NostrRefPreValidation, PrsUrlConstraints, }; -use crate::git::handlers::{build_git_protocol_error_response, is_git_protocol_error, GitError}; +use crate::git::handlers::{ + build_git_protocol_error_response, err_pktline_frame, is_git_protocol_error, + pump_stdout_to_channel, read_stderr_to_end, record_git_operation, streaming_response, GitError, + PumpResult, STREAM_CHANNEL_DEPTH, +}; use crate::git::protocol::GitService; use crate::git::subprocess::GitSubprocess; use crate::git::sync::process_newly_available_git_data; -use crate::git::{delete_ref, list_refs}; +use crate::git::{delete_ref, list_refs, GitResponseBody}; use crate::grasp06::endpoint::PrsUrl; use crate::grasp06::paths::prs_repo_path; +use crate::metrics::Metrics; use crate::nostr::builder::Nip34WritePolicy; use crate::nostr::SharedDatabase; use crate::purgatory::promotion_hooks::NostrPurgatoryPromotionHooks; @@ -85,9 +100,9 @@ use crate::sync::rejected_index::RejectedEventsIndex; /// /// * by the receive handler to perform `git init --bare` and register /// the request as in-flight (`fetch_add` on `in_flight`), -/// * by the receive handler again at end-of-push to decrement -/// `in_flight` and, if no other push is in flight and the repo has -/// zero refs, remove the bare directory, +/// * by the detached receive-pack streaming task at end-of-push to decrement +/// `in_flight` and, if no other push is in flight and the repo has zero refs, +/// remove the bare directory, /// * by off-push cleanup paths (PR-event validation discard, purgatory /// expiry) for the duration of one `delete_ref` + optional /// `remove_dir_all`. @@ -161,7 +176,8 @@ pub async fn handle_prs_receive_pack( git_protocol: Option<&str>, repo_init_locks: RepoInitLocks, domain: &str, -) -> Result>, GitError> { + metrics: Option>, +) -> Result, GitError> { // 1. Pre-scan refs and reject the whole push if any ref name is not // `refs/nostr/<64-lowercase-hex>`. We use the same parser as the // standard receive-pack path so behaviour stays in lock-step. @@ -172,6 +188,7 @@ pub async fn handle_prs_receive_pack( prs.submitter.to_hex(), prs.identifier ); + record_git_operation(&metrics, "push", "error"); return Ok(build_git_protocol_error_response( GitService::ReceivePack, "no ref updates found in push", @@ -187,6 +204,7 @@ pub async fn handle_prs_receive_pack( prs.identifier, reason ); + record_git_operation(&metrics, "push", "error"); return Ok(build_git_protocol_error_response( GitService::ReceivePack, &format!( @@ -226,6 +244,7 @@ pub async fn handle_prs_receive_pack( prs.identifier, reason ); + record_git_operation(&metrics, "push", "error"); return Ok(build_git_protocol_error_response( GitService::ReceivePack, &format!("GRASP-06: {}", reason), @@ -258,128 +277,82 @@ pub async fn handle_prs_receive_pack( repo_path.display(), e ); + record_git_operation(&metrics, "push", "error"); return Err(e); } state.in_flight.fetch_add(1, Ordering::Relaxed); } - // 4 + 5: run receive-pack and re-validate refs without the per-path - // mutex held. Wrapping in an async block lets us catch every - // exit path with the same end-of-push cleanup below. - let process_result: Result>, GitError> = async { - let response = run_receive_pack(&repo_path, &request_body, git_protocol).await?; - - // If the push itself failed with a protocol error (e.g. a stale - // OID or a corrupt pack) we return that ERR pkt-line straight - // back to the client without doing any post-push validation. The - // pre-scan in step 1 already gated ref-name shape, so we only - // land here for git-level failures. - if response.status() != StatusCode::OK { - return Ok(response); - } - - // 5. Race safety net. The pre-validation in step 2 was performed - // *before* `git-receive-pack` ran, so an event with one of the - // pushed ids may have arrived via WebSocket during the - // receive-pack window. Re-run the same shared check now and - // delete the ref on any mismatch. The `Unknown` branch is the - // common case — neither DB nor purgatory have heard of this - // event yet — and creates a scoped placeholder so the - // 30-minute purgatory sweep can clean it up if the event - // never arrives. - let post_push_constraints = PrsUrlConstraints { - submitter: &prs.submitter, - identifier: &prs.identifier, - domain, - }; - for (_, new_oid, ref_name) in &pushed_refs { - let event_id_hex = ref_name - .strip_prefix("refs/nostr/") - .expect("ref shape validated above"); - post_push_validate( - &database, - &purgatory, - &repo_path, - post_push_constraints, - event_id_hex, - new_oid, - ref_name, - ) - .await; - } - - Ok(response) - } - .await; - - // 6. End-of-push cleanup. Always runs — including on receive-pack - // protocol errors and `?`-propagated transport errors — so a - // failed push that has just initialised an empty repo does not - // leak it. Decrements `in_flight`; if no other push is in flight - // and the repo has zero refs left, removes the bare directory. - { - let _g = state.mu.lock().expect("prs path mutex poisoned"); - state.in_flight.fetch_sub(1, Ordering::Relaxed); - if state.in_flight.load(Ordering::Relaxed) == 0 { - if let Ok(refs) = list_refs(&repo_path) { - if refs.is_empty() { - if let Err(e) = std::fs::remove_dir_all(&repo_path) { - warn!( - "/prs/ receive-pack: failed to clean up empty repo {}: {}", - repo_path.display(), - e - ); - } else { - debug!( - "/prs/ receive-pack: removed empty repo {} (no refs after push)", - repo_path.display() - ); - } - } - } - } - } - - let response = process_result?; - - // 7. Drive the standard purgatory-release pipeline so PR events - // already waiting on these commits can be promoted out of - // purgatory. Only fires on a successful push, and only if the - // repo still exists (it may have been removed in step 6). - if response.status() == StatusCode::OK && repo_path.exists() { - let new_oids: HashSet = pushed_refs - .iter() - .filter(|(_, new_oid, _)| new_oid != "0000000000000000000000000000000000000000") - .map(|(_, new_oid, _)| new_oid.clone()) - .collect(); - - let promotion_hooks = NostrPurgatoryPromotionHooks::git_push( - &write_policy, - &rejected_events_index, - Some(&relay), - ); - - if let Err(e) = process_newly_available_git_data( - &repo_path, - &new_oids, - &database, - Some(&relay), - &purgatory, - Path::new(git_data_path), - Some(&promotion_hooks), - ) - .await + // 4. The pre-validation and repo creation work above must stay buffered so + // rejections can be returned as a complete Git `ERR` response before any + // repository state is touched. Once we start `git-receive-pack`, switch + // to the same streaming shape as the standard receive-pack endpoint: + // write stdin here, then hand stdout/stderr and all follow-up state to a + // detached task that feeds the response body channel. + let mut git = + match GitSubprocess::spawn(GitService::ReceivePack, &repo_path, false, git_protocol) + .map_err(GitError::ProcessSpawnFailed) { - warn!( - "/prs/ receive-pack: post-push processing failed for {}/{}: {}", - prs.submitter.to_hex(), - prs.identifier, - e - ); + Ok(git) => git, + Err(e) => { + finish_prs_receive_pack(&state, &repo_path); + record_git_operation(&metrics, "push", "error"); + return Err(e); + } + }; + + if let Some(mut stdin) = git.take_stdin() { + if let Err(e) = stdin.write_all(&request_body).await { + let _ = git.kill().await; + finish_prs_receive_pack(&state, &repo_path); + record_git_operation(&metrics, "push", "error"); + return Err(GitError::IoError(e)); } + drop(stdin); } - Ok(response) + let stdout = match git.take_stdout() { + Some(stdout) => stdout, + None => { + let _ = git.kill().await; + finish_prs_receive_pack(&state, &repo_path); + record_git_operation(&metrics, "push", "error"); + return Err(GitError::IoError(io::Error::new( + io::ErrorKind::BrokenPipe, + "git receive-pack stdout unavailable", + ))); + } + }; + let stderr = git.take_stderr(); + let (tx, rx) = mpsc::channel::, io::Error>>(STREAM_CHANNEL_DEPTH); + + // The request future returns as soon as the channel-backed response is + // built. Everything that previously happened after the buffered + // `run_receive_pack` call must therefore move into this task: forwarding + // stdout, classifying Git failures, decrementing `in_flight`, race-window + // validation, empty-repo cleanup, purgatory promotion, and metrics. + tokio::spawn(stream_prs_receive_pack_output( + git, + stdout, + stderr, + tx, + repo_path, + state, + pushed_refs, + database, + relay, + purgatory, + write_policy, + rejected_events_index, + git_data_path.to_string(), + request_body, + prs.submitter, + prs.identifier.clone(), + domain.to_string(), + metrics, + )); + + Ok(streaming_response(GitService::ReceivePack, rx)) } /// Return `Some(reason)` if `ref_name` is not exactly @@ -448,75 +421,218 @@ fn ensure_repo_initialised(repo_path: &Path) -> Result<(), GitError> { Ok(()) } -/// Spawn `git-receive-pack`, stream stdin/stdout/stderr, and convert the -/// result into an HTTP response — protocol errors become 200 + ERR -/// pkt-line, transport errors bubble up as [`GitError`]. Mirrors -/// [`crate::git::handlers::handle_receive_pack`]'s subprocess plumbing. -async fn run_receive_pack( - repo_path: &Path, - request_body: &Bytes, - git_protocol: Option<&str>, -) -> Result>, GitError> { - let mut git = GitSubprocess::spawn(GitService::ReceivePack, repo_path, false, git_protocol) - .map_err(GitError::ProcessSpawnFailed)?; +/// End-of-push cleanup. Always runs — including on receive-pack protocol +/// errors and client disconnects — so a failed push that has just initialised +/// an empty repo does not leak it. Decrements `in_flight`; if no other push is +/// in flight and the repo has zero refs left, removes the bare directory. +fn finish_prs_receive_pack(state: &PrsPathState, repo_path: &Path) { + let _g = state.mu.lock().expect("prs path mutex poisoned"); + state.in_flight.fetch_sub(1, Ordering::Relaxed); + if state.in_flight.load(Ordering::Relaxed) == 0 { + if let Ok(refs) = list_refs(repo_path) { + if refs.is_empty() { + if let Err(e) = std::fs::remove_dir_all(repo_path) { + warn!( + "/prs/ receive-pack: failed to clean up empty repo {}: {}", + repo_path.display(), + e + ); + } else { + debug!( + "/prs/ receive-pack: removed empty repo {} (no refs after push)", + repo_path.display() + ); + } + } + } + } +} - if let Some(mut stdin) = git.take_stdin() { - stdin - .write_all(request_body) - .await - .map_err(GitError::IoError)?; - drop(stdin); +/// Own the live `/prs/` receive-pack after the request handler has returned its +/// streaming response. +/// +/// This task is the continuation of [`handle_prs_receive_pack`], not just a +/// stdout pump. It must preserve the old buffered handler's guarantees while the +/// HTTP body is already streaming: +/// +/// * if stdout reaches EOF cleanly, inspect Git's exit status and stderr; +/// * if Git failed before sending stdout, synthesize a final Git `ERR` pkt-line +/// through the response channel; +/// * always run `finish_prs_receive_pack` on every terminal path after +/// `in_flight` was incremented; +/// * only run race-window validation and purgatory promotion after a successful +/// receive-pack. +#[allow(clippy::too_many_arguments)] +async fn stream_prs_receive_pack_output( + mut git: GitSubprocess, + stdout: S, + stderr: Option, + tx: mpsc::Sender, io::Error>>, + repo_path: PathBuf, + state: Arc, + pushed_refs: Vec<(String, String, String)>, + database: SharedDatabase, + relay: LocalRelay, + purgatory: Arc, + write_policy: Arc, + rejected_events_index: Arc, + git_data_path: String, + request_body: Bytes, + submitter: PublicKey, + identifier: String, + domain: String, + metrics: Option>, +) where + S: tokio::io::AsyncRead + Unpin + Send + 'static, + E: tokio::io::AsyncRead + Unpin + Send + 'static, +{ + // Drain stderr concurrently with stdout. Git can write enough diagnostics to + // fill its stderr pipe while still producing stdout; draining both prevents + // child-process deadlock and preserves stderr for protocol-error reporting. + let stderr_task = stderr.map(|stderr| tokio::spawn(read_stderr_to_end(stderr))); + let pump_result = pump_stdout_to_channel(stdout, &tx).await; + + // If the client goes away or stdout read fails, stop Git rather than letting + // it continue writing into a response nobody can receive. Cleanup below will + // still release `in_flight` after `wait()` observes process termination. + if !matches!(pump_result, PumpResult::Eof { .. }) { + let _ = git.kill().await; } - let mut output = Vec::new(); - let mut stderr_output = Vec::new(); + let status = match git.wait().await { + Ok(status) => status, + Err(e) => { + let _ = tx.send(Err(e)).await; + finish_prs_receive_pack(&state, &repo_path); + record_git_operation(&metrics, "push", "error"); + return; + } + }; - if let Some(stdout) = git.take_stdout() { - let mut stdout = stdout; - stdout - .read_to_end(&mut output) - .await - .map_err(GitError::IoError)?; - } - if let Some(stderr) = git.take_stderr() { - let mut stderr = stderr; - stderr - .read_to_end(&mut stderr_output) - .await - .map_err(GitError::IoError)?; - } + let stderr_output = match stderr_task { + Some(task) => task.await.unwrap_or_default(), + None => Vec::new(), + }; - let status = git.wait().await.map_err(GitError::IoError)?; + let sent_stdout = match pump_result { + PumpResult::Eof { sent_stdout } => sent_stdout, + PumpResult::ClientDisconnected | PumpResult::ReadError => { + finish_prs_receive_pack(&state, &repo_path); + record_git_operation(&metrics, "push", "error"); + return; + } + }; if !status.success() { + record_git_operation(&metrics, "push", "error"); let stderr_str = String::from_utf8_lossy(&stderr_output); if is_git_protocol_error(status.code(), &stderr_output) { warn!( "/prs/ git-receive-pack protocol error (returning ERR pkt-line): {}", stderr_str.trim() ); - return Ok(build_git_protocol_error_response( - GitService::ReceivePack, - &stderr_str, - Some(request_body), - )); + if !sent_stdout { + let _ = tx + .send(Ok(err_pktline_frame( + GitService::ReceivePack, + &stderr_str, + Some(&request_body), + ))) + .await; + } + } else { + error!( + "/prs/ git-receive-pack failed (transport): {}", + stderr_str.trim() + ); + if !sent_stdout { + let msg = if stderr_str.trim().is_empty() { + format!("git receive-pack failed with code {:?}", status.code()) + } else { + stderr_str.to_string() + }; + let _ = tx + .send(Ok(err_pktline_frame( + GitService::ReceivePack, + &msg, + Some(&request_body), + ))) + .await; + } } - error!( - "/prs/ git-receive-pack failed (transport): {}", - stderr_str.trim() - ); - return Err(GitError::GitFailed(status.code())); + // Whether or not an ERR frame could be sent, the HTTP headers are + // already committed as a streaming 200 response. The only remaining + // safe action is to close the body after cleanup. + finish_prs_receive_pack(&state, &repo_path); + return; } - Ok(Response::builder() - .status(StatusCode::OK) - .header( - "content-type", - GitService::ReceivePack.result_content_type(), + debug!("/prs/ git-receive-pack stream completed successfully"); + record_git_operation(&metrics, "push", "success"); + + // Race safety net. The pre-validation in `handle_prs_receive_pack` was + // performed before `git-receive-pack` ran, so an event with one of the + // pushed ids may have arrived via WebSocket during the receive-pack window. + // Re-run the same shared check now and delete the ref on any mismatch. + for (_, new_oid, ref_name) in &pushed_refs { + let event_id_hex = ref_name + .strip_prefix("refs/nostr/") + .expect("ref shape validated above"); + let post_push_constraints = PrsUrlConstraints { + submitter: &submitter, + identifier: &identifier, + domain: &domain, + }; + post_push_validate( + &database, + &purgatory, + &repo_path, + post_push_constraints, + event_id_hex, + new_oid, + ref_name, ) - .header("cache-control", "no-cache") - .body(Full::new(Bytes::from(output))) - .unwrap()) + .await; + } + + finish_prs_receive_pack(&state, &repo_path); + + // Drive the standard purgatory-release pipeline so PR events already + // waiting on these commits can be promoted out of purgatory. Only fires on + // a successful push, and only if the repo still exists (it may have been + // removed by end-of-push cleanup). + if repo_path.exists() { + let new_oids: HashSet = pushed_refs + .iter() + .filter(|(_, new_oid, _)| new_oid != "0000000000000000000000000000000000000000") + .map(|(_, new_oid, _)| new_oid.clone()) + .collect(); + + let promotion_hooks = NostrPurgatoryPromotionHooks::git_push( + write_policy, + rejected_events_index, + Some(relay.clone()), + ); + + if let Err(e) = process_newly_available_git_data( + &repo_path, + &new_oids, + &database, + Some(&relay), + &purgatory, + Path::new(&git_data_path), + Some(&promotion_hooks), + ) + .await + { + warn!( + "/prs/ receive-pack: post-push processing failed for {}/{}: {}", + submitter.to_hex(), + identifier, + e + ); + } + } } /// Race safety net for the `/prs/` receive-pack post-push phase. @@ -534,10 +650,11 @@ async fn run_receive_pack( /// - was held in purgatory with a populated entry that also mismatches /// (delete the ref AND drop the purgatory entry — its event is wrong). /// -/// Anything left here is best-effort: errors deleting refs are logged -/// and the push response is not changed. The end-of-push cleanup in -/// [`handle_prs_receive_pack`] removes the bare repo if every ref ends -/// up deleted. +/// Anything left here is best-effort: errors deleting refs are logged and the +/// push response is not changed. By the time this runs the response body has +/// already streamed Git's success output, so the only safe correction is to fix +/// repository state before `finish_prs_receive_pack` decides whether the bare +/// repo is now empty. async fn post_push_validate( database: &SharedDatabase, purgatory: &Purgatory, diff --git a/src/http/mod.rs b/src/http/mod.rs index c88a778..6115f44 100644 --- a/src/http/mod.rs +++ b/src/http/mod.rs @@ -10,7 +10,7 @@ use std::pin::Pin; use std::sync::Arc; use base64::Engine; -use http_body_util::{BodyExt, Full}; +use http_body_util::BodyExt; use hyper::body::{Bytes, Incoming}; use hyper::header::{CONNECTION, SEC_WEBSOCKET_ACCEPT, UPGRADE}; use hyper::server::conn::http1; @@ -25,6 +25,7 @@ use tokio::net::TcpListener; use crate::config::Config; use crate::git; +use crate::git::{empty_body, full_body, GitResponseBody}; use crate::grasp06::receive::RepoInitLocks; use crate::metrics::Metrics; use crate::nostr::builder::Nip34WritePolicy; @@ -34,6 +35,8 @@ use crate::purgatory::promotion_hooks::NostrPurgatoryPromotionHooks; use crate::purgatory::Purgatory; use crate::sync::rejected_index::RejectedEventsIndex; +type HttpBody = GitResponseBody; + /// CORS headers required by GRASP-01 specification (lines 40-47) const CORS_ALLOW_ORIGIN: &str = "*"; const CORS_ALLOW_METHODS: &str = "GET, POST"; @@ -159,7 +162,7 @@ impl HttpService { } impl Service> for HttpService { - type Response = Response>; + type Response = Response; type Error = String; type Future = Pin> + Send>>; @@ -183,7 +186,7 @@ impl Service> for HttpService { Ok( add_cors_headers(Response::builder().header("server", "ngit-grasp")) .status(204) - .body(Full::new(Bytes::new())) + .body(empty_body()) .unwrap(), ) }); @@ -245,7 +248,7 @@ impl Service> for HttpService { raw_body }; - let result: Result>, git::handlers::GitError> = + let result: Result, git::handlers::GitError> = match (method_clone.as_ref(), subpath.as_str()) { // GET|HEAD /info/refs?service=git-{upload,receive}-pack // HEAD must mirror GET headers with no body (RFC 9110 §9.3.2) @@ -284,16 +287,33 @@ impl Service> for HttpService { // POST /git-upload-pack — clone/fetch. (m, "git-upload-pack") if m == Method::POST => { + let streams_real_repo = crate::grasp06::paths::prs_repo_path( + std::path::Path::new(&git_data_path), + &prs.submitter.to_hex(), + &prs.identifier, + ) + .exists(); let r = crate::grasp06::fetch::handle_prs_upload_pack( &prs, &git_data_path, body_bytes, git_protocol.as_deref(), + metrics_clone.clone(), ) .await; if let Some(ref m) = metrics_clone { let status = if r.is_ok() { "success" } else { "error" }; - m.record_git_operation("clone", status); + if streams_real_repo && r.is_ok() { + // Real /prs/ repos delegate to the standard streaming + // upload-pack handler. The streaming task records the final + // git-upload-pack success/failure when Git exits; this + // `Ok(Response)` only means Hyper received a body stream. + } else { + // Synthesized empty repos are still buffered, so `Ok` here + // means git-upload-pack has already completed. For streamed + // setup errors, there is no task to record the failure. + m.record_git_operation("clone", status); + } } r } @@ -302,6 +322,11 @@ impl Service> for HttpService { // refs/nostr/, reject anything else, // per GRASP-06 06.md line 15. (m, "git-receive-pack") if m == Method::POST => { + // Metrics are recorded inside the `/prs/` + // handler and its detached streaming task. For + // successful setup this `await` only means the + // response stream was created, not that Git has + // finished receiving the push. let r = crate::grasp06::receive::handle_prs_receive_pack( &prs, body_bytes, @@ -314,12 +339,9 @@ impl Service> for HttpService { git_protocol.as_deref(), repo_init_locks.clone(), &config_clone.domain, + metrics_clone.clone(), ) .await; - if let Some(ref m) = metrics_clone { - let status = if r.is_ok() { "success" } else { "error" }; - m.record_git_operation("push", status); - } r } @@ -332,7 +354,7 @@ impl Service> for HttpService { // RFC 9110 §9.3.2: HEAD response must have same headers as GET // but no body. let body = if method_clone == Method::HEAD { - Full::new(Bytes::new()) + empty_body() } else { body }; @@ -360,7 +382,7 @@ impl Service> for HttpService { let error_msg = format!("Git error: {}", e); Ok(add_cors_headers(Response::builder()) .status(e.status_code()) - .body(Full::new(Bytes::from(error_msg))) + .body(full_body(error_msg)) .unwrap()) } } @@ -482,7 +504,7 @@ impl Service> for HttpService { // POST /git-upload-pack (clone/fetch) (m, "git-upload-pack") if m == Method::POST => { - let _repo_lifecycle_guard = match PublicKey::parse(&npub) { + let repo_lifecycle_guard = match PublicKey::parse(&npub) { Ok(owner_pk) => Some( lifecycle .read_repository(&owner_pk.to_hex(), &identifier) @@ -494,11 +516,17 @@ impl Service> for HttpService { repo_path, body_bytes, git_protocol.as_deref(), + repo_lifecycle_guard, + metrics_clone.clone(), ) .await; if let Some(ref m) = metrics_clone { - let status = if result.is_ok() { "success" } else { "error" }; - m.record_git_operation("clone", status); + if result.is_err() { + // On `Ok(Response)`, the streaming task owns the actual + // git-upload-pack outcome metric. The synchronous handler result + // only says whether stream setup succeeded. + m.record_git_operation("clone", "error"); + } } result } @@ -516,7 +544,7 @@ impl Service> for HttpService { } return Ok(add_cors_headers(Response::builder()) .status(hyper::StatusCode::BAD_REQUEST) - .body(Full::new(Bytes::from(format!("Invalid npub: {}", e)))) + .body(full_body(format!("Invalid npub: {}", e))) .unwrap()); } }; @@ -528,9 +556,9 @@ impl Service> for HttpService { ); let promotion_hooks = NostrPurgatoryPromotionHooks::git_push( - &write_policy, - &rejected_events_index, - Some(&relay), + write_policy.clone(), + rejected_events_index.clone(), + Some(relay.clone()), ); let result = git::handlers::handle_receive_pack( @@ -544,13 +572,19 @@ impl Service> for HttpService { &git_data_path, git_protocol.as_deref(), repo_lifecycle_guard, - Some(&promotion_hooks), + Some(Arc::new(promotion_hooks)), + metrics_clone.clone(), ) .await; if let Some(ref m) = metrics_clone { - let status = if result.is_ok() { "success" } else { "error" }; - m.record_git_operation("push", status); + if result.is_err() { + // On `Ok(Response)`, the operation may be a buffered protocol + // rejection already recorded by `handle_receive_pack`, or a stream + // whose final git-receive-pack outcome will be recorded by the + // streaming task. Do not treat this setup result as success. + m.record_git_operation("push", "error"); + } } result @@ -566,7 +600,7 @@ impl Service> for HttpService { // RFC 9110 §9.3.2: HEAD response must have same headers as GET // but no body. let body = if method == Method::HEAD { - Full::new(Bytes::new()) + empty_body() } else { body }; @@ -595,7 +629,7 @@ impl Service> for HttpService { let error_msg = format!("Git error: {}", e); Ok(add_cors_headers(Response::builder()) .status(e.status_code()) - .body(Full::new(Bytes::from(error_msg))) + .body(full_body(error_msg)) .unwrap()) } } @@ -625,7 +659,7 @@ impl Service> for HttpService { add_cors_headers(Response::builder().header("server", "ngit-grasp")) .status(200) .header("content-type", "application/nostr+json") - .body(Full::new(Bytes::from(json))) + .body(full_body(json)) .unwrap(), ) }); @@ -656,7 +690,7 @@ impl Service> for HttpService { add_cors_headers(Response::builder().header("server", "ngit-grasp")) .status(200) .header("content-type", "text/html; charset=utf-8") - .body(Full::new(Bytes::from(html))) + .body(full_body(html)) .unwrap(), ) } else { @@ -666,7 +700,7 @@ impl Service> for HttpService { add_cors_headers(Response::builder().header("server", "ngit-grasp")) .status(404) .header("content-type", "text/html; charset=utf-8") - .body(Full::new(Bytes::from(html))) + .body(full_body(html)) .unwrap(), ) } @@ -722,7 +756,7 @@ impl Service> for HttpService { .header(CONNECTION, "upgrade") .header(UPGRADE, "websocket") .header(SEC_WEBSOCKET_ACCEPT, derived.unwrap()) - .body(Full::new(Bytes::new())) + .body(empty_body()) .unwrap()) }); } @@ -738,7 +772,7 @@ impl Service> for HttpService { add_cors_headers(Response::builder().header("server", "ngit-grasp")) .status(200) .header("content-type", "text/plain; version=0.0.4; charset=utf-8") - .body(Full::new(Bytes::from(output))) + .body(full_body(output)) .unwrap(), ) }); @@ -748,7 +782,7 @@ impl Service> for HttpService { Ok( add_cors_headers(Response::builder().header("server", "ngit-grasp")) .status(404) - .body(Full::new(Bytes::from("Metrics disabled"))) + .body(full_body("Metrics disabled")) .unwrap(), ) }); @@ -763,7 +797,7 @@ impl Service> for HttpService { .status(200) .header("content-type", "image/png") .header("cache-control", "public, max-age=86400") - .body(Full::new(Bytes::from_static(ICON_PNG))) + .body(full_body(Bytes::from_static(ICON_PNG))) .unwrap(), ) }); @@ -779,7 +813,7 @@ impl Service> for HttpService { add_cors_headers(Response::builder().header("server", "ngit-grasp")) .status(200) .header("content-type", "text/html; charset=utf-8") - .body(Full::new(Bytes::from(html))) + .body(full_body(html)) .unwrap(), ) } else { @@ -789,7 +823,7 @@ impl Service> for HttpService { add_cors_headers(Response::builder().header("server", "ngit-grasp")) .status(404) .header("content-type", "text/html; charset=utf-8") - .body(Full::new(Bytes::from(html))) + .body(full_body(html)) .unwrap(), ) } diff --git a/src/metrics/mod.rs b/src/metrics/mod.rs index 4b619fa..16df4bc 100644 --- a/src/metrics/mod.rs +++ b/src/metrics/mod.rs @@ -398,7 +398,12 @@ impl Metrics { // === Git Operation Recording Methods === - /// Record a git operation completion + /// Record a git operation event. + /// + /// Buffered handlers record completion from the synchronous request path. + /// Streaming handlers record the final `success`/`error` from the background + /// task that owns the Git subprocess; their synchronous `Ok(Response)` only + /// means stream setup succeeded. pub fn record_git_operation(&self, operation: &str, status: &str) { self.inner .git_operations_total diff --git a/src/purgatory/promotion_hooks.rs b/src/purgatory/promotion_hooks.rs index 9716408..4854bef 100644 --- a/src/purgatory/promotion_hooks.rs +++ b/src/purgatory/promotion_hooks.rs @@ -21,19 +21,19 @@ use crate::nostr::lifecycle::DeletionService; use crate::nostr::persistence::SaveContext; use crate::sync::rejected_index::{EventType, RejectedEventsIndex}; -pub struct NostrPurgatoryPromotionHooks<'a> { - deletion: &'a DeletionService, - write_policy: Option<&'a Nip34WritePolicy>, - rejected_events_index: Option<&'a Arc>, - local_relay: Option<&'a LocalRelay>, +pub struct NostrPurgatoryPromotionHooks { + deletion: DeletionService, + write_policy: Option>, + rejected_events_index: Option>, + local_relay: Option, } -impl<'a> NostrPurgatoryPromotionHooks<'a> { +impl NostrPurgatoryPromotionHooks { /// Run deletion recovery and accepted-event persistence hooks only. - pub fn recovery_only(write_policy: &'a Nip34WritePolicy) -> Self { + pub fn recovery_only(write_policy: &Nip34WritePolicy) -> Self { Self { - deletion: write_policy.deletion(), - write_policy: Some(write_policy), + deletion: write_policy.deletion().clone(), + write_policy: Some(Arc::new(write_policy.clone())), rejected_events_index: None, local_relay: None, } @@ -41,12 +41,12 @@ impl<'a> NostrPurgatoryPromotionHooks<'a> { /// Git-push path: deletion recovery plus immediate hot-cache reprocessing. pub fn git_push( - write_policy: &'a Nip34WritePolicy, - rejected_events_index: &'a Arc, - local_relay: Option<&'a LocalRelay>, + write_policy: Arc, + rejected_events_index: Arc, + local_relay: Option, ) -> Self { Self { - deletion: write_policy.deletion(), + deletion: write_policy.deletion().clone(), write_policy: Some(write_policy), rejected_events_index: Some(rejected_events_index), local_relay, @@ -55,7 +55,7 @@ impl<'a> NostrPurgatoryPromotionHooks<'a> { } #[async_trait] -impl PurgatoryPromotionHooks for NostrPurgatoryPromotionHooks<'_> { +impl PurgatoryPromotionHooks for NostrPurgatoryPromotionHooks { async fn before_event_saved(&self, event: &Event, context: PurgatorySaveContext) { tracing::trace!( event_id = %event.id, @@ -77,9 +77,9 @@ impl PurgatoryPromotionHooks for NostrPurgatoryPromotionHooks<'_> { async fn after_announcement_saved(&self, event: &Event) { let (Some(write_policy), Some(rejected_events_index), Some(relay)) = ( - self.write_policy, - self.rejected_events_index, - self.local_relay, + self.write_policy.as_deref(), + self.rejected_events_index.as_ref(), + self.local_relay.as_ref(), ) else { return; }; diff --git a/tests/git_response_streaming.rs b/tests/git_response_streaming.rs new file mode 100644 index 0000000..ec8df95 --- /dev/null +++ b/tests/git_response_streaming.rs @@ -0,0 +1,300 @@ +//! Integration coverage for Git Smart HTTP response streaming. + +use std::path::Path; +use std::sync::Arc; +use std::time::Duration; + +use clap::Parser; +use http_body_util::BodyExt; +use hyper::body::{Bytes, Frame}; +use ngit_grasp::config::Config; +use ngit_grasp::git::handlers::handle_receive_pack; +use ngit_grasp::grasp06::endpoint::PrsUrl; +use ngit_grasp::grasp06::paths::prs_repo_path; +use ngit_grasp::grasp06::receive::{handle_prs_receive_pack, new_repo_init_locks}; +use ngit_grasp::nostr::builder::Nip34WritePolicy; +use ngit_grasp::nostr::lifecycle::{ + HoldingStore, ReplaceableHistoryStore, RepositoryLifecycle, Tombstones, +}; +use ngit_grasp::nostr::SharedDatabase; +use ngit_grasp::purgatory::Purgatory; +use ngit_grasp::sync::rejected_index::RejectedEventsIndex; +use nostr_relay_builder::prelude::LocalRelayBuilder; +use nostr_sdk::prelude::*; +use tokio::time::timeout; + +static PATH_ENV_LOCK: tokio::sync::Mutex<()> = tokio::sync::Mutex::const_new(()); + +struct PathOverride { + original: Option, +} + +impl PathOverride { + fn prepend(dir: &Path) -> Self { + let original = std::env::var_os("PATH"); + let mut paths = vec![dir.to_path_buf()]; + if let Some(existing) = original.as_ref() { + paths.extend(std::env::split_paths(existing)); + } + let joined = std::env::join_paths(paths).expect("join PATH entries"); + std::env::set_var("PATH", joined); + Self { original } + } +} + +impl Drop for PathOverride { + fn drop(&mut self) { + if let Some(original) = self.original.take() { + std::env::set_var("PATH", original); + } else { + std::env::remove_var("PATH"); + } + } +} + +#[tokio::test] +async fn receive_pack_response_streams_stdout_before_subprocess_exit() { + let _env_lock = PATH_ENV_LOCK.lock().await; + + let fake_bin = tempfile::tempdir().expect("fake git bin tempdir"); + write_fake_git(fake_bin.path()); + let _path = PathOverride::prepend(fake_bin.path()); + + let repo = tempfile::tempdir().expect("repo tempdir"); + let git_data = tempfile::tempdir().expect("git data tempdir"); + let database: SharedDatabase = Arc::new(nostr_memory::MemoryDatabase::unbounded()); + let relay = LocalRelayBuilder::default().build(); + let purgatory = Arc::new(Purgatory::new(git_data.path().to_path_buf())); + let owner_pubkey = "0".repeat(64); + let request_body = receive_pack_request_body(); + + let response = handle_receive_pack( + repo.path().to_path_buf(), + request_body, + database, + relay, + "streaming-test", + &owner_pubkey, + purgatory, + git_data.path().to_str().expect("utf-8 temp path"), + None, + None, + None, + None, + ) + .await + .expect("receive-pack handler should start fake subprocess"); + + let mut body = response.into_body(); + + let first = timeout(Duration::from_secs(1), body.frame()) + .await + .expect("first stdout chunk should arrive before fake git exits") + .expect("body should still be open") + .expect("first frame should not be an HTTP body error"); + assert_eq!(frame_data(first), Bytes::from_static(b"first-progress\n")); + + let no_second_yet = timeout(Duration::from_millis(250), body.frame()).await; + assert!( + no_second_yet.is_err(), + "body produced another frame while fake git was still sleeping; \ + this test needs the first frame to be observed before subprocess EOF" + ); + + let second = timeout(Duration::from_secs(3), body.frame()) + .await + .expect("second stdout chunk should arrive after fake git wakes") + .expect("body should still be open for second chunk") + .expect("second frame should not be an HTTP body error"); + assert_eq!(frame_data(second), Bytes::from_static(b"second-progress\n")); +} + +#[tokio::test] +async fn prs_receive_pack_streams_stdout_before_cleanup_removes_empty_repo() { + let _env_lock = PATH_ENV_LOCK.lock().await; + + let fake_bin = tempfile::tempdir().expect("fake git bin tempdir"); + write_fake_git(fake_bin.path()); + let _path = PathOverride::prepend(fake_bin.path()); + + let keys = Keys::generate(); + let git_data = tempfile::tempdir().expect("git data tempdir"); + let database: SharedDatabase = Arc::new(nostr_memory::MemoryDatabase::unbounded()); + let relay = LocalRelayBuilder::default().build(); + let purgatory = Arc::new(Purgatory::new(git_data.path().to_path_buf())); + let repo_init_locks = new_repo_init_locks(); + let write_policy = Arc::new(test_write_policy( + database.clone(), + purgatory.clone(), + repo_init_locks.clone(), + git_data.path(), + )); + let rejected_events_index = Arc::new(RejectedEventsIndex::new( + Duration::from_secs(120), + Duration::from_secs(7 * 24 * 60 * 60), + )); + let prs = PrsUrl { + submitter: keys.public_key(), + identifier: "prs-streaming-cleanup".to_string(), + subpath: "git-receive-pack".to_string(), + }; + let repo_path = prs_repo_path(git_data.path(), &prs.submitter.to_hex(), &prs.identifier); + + let response = handle_prs_receive_pack( + &prs, + receive_pack_request_body(), + database, + relay, + purgatory, + write_policy, + rejected_events_index, + git_data.path().to_str().expect("utf-8 temp path"), + None, + repo_init_locks, + "streaming-test.example", + None, + ) + .await + .expect("/prs/ receive-pack handler should start fake subprocess"); + + assert!( + repo_path.exists(), + "/prs/ repo should exist while fake receive-pack is in flight" + ); + + let mut body = response.into_body(); + + let first = timeout(Duration::from_secs(1), body.frame()) + .await + .expect("first /prs/ stdout chunk should arrive before fake git exits") + .expect("/prs/ body should still be open") + .expect("first /prs/ frame should not be an HTTP body error"); + assert_eq!(frame_data(first), Bytes::from_static(b"first-progress\n")); + + assert!( + repo_path.exists(), + "/prs/ cleanup must not remove the repo before receive-pack exits" + ); + + let no_second_yet = timeout(Duration::from_millis(250), body.frame()).await; + assert!( + no_second_yet.is_err(), + "/prs/ body produced another frame while fake git was still sleeping; \ + this test needs the first frame to be observed before subprocess EOF" + ); + + let second = timeout(Duration::from_secs(3), body.frame()) + .await + .expect("second /prs/ stdout chunk should arrive after fake git wakes") + .expect("/prs/ body should still be open for second chunk") + .expect("second /prs/ frame should not be an HTTP body error"); + assert_eq!(frame_data(second), Bytes::from_static(b"second-progress\n")); + + let eof = timeout(Duration::from_secs(1), body.frame()) + .await + .expect("/prs/ body should close after cleanup"); + assert!( + eof.is_none(), + "/prs/ body should be closed after fake git exits" + ); + + assert!( + !repo_path.exists(), + "/prs/ cleanup should remove zero-ref repo after receive-pack exits" + ); +} + +fn frame_data(frame: Frame) -> Bytes { + frame.into_data().expect("frame should contain data") +} + +fn receive_pack_request_body() -> Bytes { + let old_oid = "0".repeat(40); + let new_oid = "1".repeat(40); + let event_id = "a".repeat(64); + let mut payload = Vec::new(); + payload.extend_from_slice(format!("{old_oid} {new_oid} refs/nostr/{event_id}").as_bytes()); + payload.push(0); + payload.extend_from_slice(b"report-status side-band-64k\n"); + + let mut request = Vec::new(); + request.extend_from_slice(format!("{:04x}", payload.len() + 4).as_bytes()); + request.extend_from_slice(&payload); + request.extend_from_slice(b"0000"); + Bytes::from(request) +} + +fn test_write_policy( + database: SharedDatabase, + purgatory: Arc, + repo_init_locks: ngit_grasp::grasp06::receive::RepoInitLocks, + git_data_path: &Path, +) -> Nip34WritePolicy { + let config = Config::parse_from([ + "ngit-grasp-test", + "--domain", + "streaming-test.example", + "--grasp06-enable", + ]); + + Nip34WritePolicy::new( + database, + Tombstones::in_memory(), + HoldingStore::in_memory(), + RepositoryLifecycle::in_memory(), + ReplaceableHistoryStore::in_memory(), + git_data_path.to_path_buf(), + purgatory, + config, + repo_init_locks, + ) +} + +fn write_fake_git(bin_dir: &Path) { + let git_path = bin_dir.join("git"); + std::fs::write( + &git_path, + r#"#!/usr/bin/env bash +set -euo pipefail + +is_receive_pack=0 +for arg in "$@"; do + if [ "$arg" = "receive-pack" ]; then + is_receive_pack=1 + fi +done + +if [ "$is_receive_pack" = "1" ]; then +cat >/dev/null +printf 'first-progress\n' +sleep 2 +printf 'second-progress\n' +exit 0 +fi + +if [ "$1" = "init" ]; then + repo="${@: -1}" + mkdir -p "$repo" + exit 0 +fi + +if [ "$1" = "for-each-ref" ]; then + exit 0 +fi + +echo "fake git only supports init, for-each-ref, and receive-pack" >&2 +exit 1 +"#, + ) + .expect("write fake git executable"); + + #[cfg(unix)] + { + use std::os::unix::fs::PermissionsExt; + let mut perms = std::fs::metadata(&git_path) + .expect("fake git metadata") + .permissions(); + perms.set_mode(0o755); + std::fs::set_permissions(&git_path, perms).expect("chmod fake git"); + } +}