fix(git): defer push completion until promotion

The large-push timeout reported in nostr:nevent1qqsyjly2u925qdc59cmxlxarvwagnr7pd6gpnr776x2mnfz6mdhn33cpz3mhxue69uhhyetvv9ujumn8d96zuer9wcj596eq led f4828c63 to stream receive-pack stdout. That change correctly kept libgit2 alive while Git spent 40-90 seconds resolving deltas and checking connectivity, but it also forwarded the terminal 0000 flush before process_newly_available_git_data ran.

Git clients treat that flush as the semantic end of receive-pack and do not need to wait for HTTP EOF. A standard GRASP push could therefore return from nostr_push while its kind-30617 announcement and kind-30618 state were still in purgatory. Immediate clone or proposal setup then intermittently observed a repository that was not queryable yet. This is the server-side cause exposed by ngit git_push_merge setup.

Continue streaming every preceding progress byte, but retain the final four-byte flush. After a successful receive-pack, promote and save the matching events, align refs and HEAD, notify subscribers, and only then release the flush. Preserve the existing failure response path when Git itself fails.

The regression blocks announcement promotion and proves progress remains visible, the terminal flush remains hidden, and the event is queryable before completion is released. This applies to the standard announced-repository receive-pack path; the GRASP-06 /prs handler has its own completion and cleanup pipeline.

Regression-from: f4828c6393

Related: nostr:nevent1qqs279s9sxvcuf2g9kq6d25nxkg6x2thcrvg9lzfzrdxt0kzpnlcykqpz3mhxue69uhhyetvv9ujuerpd46hxtnfdumztv34

CHANGELOG.md

@@ -13,6 +13,7 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ### Fixed
 
+- Smart HTTP pushes now expose the terminal receive-pack flush only after GRASP has finished promoting the matching repository announcement and state from purgatory. Git progress remains streamed while large packs are resolved and checked, but an immediately following clone or proposal push can now rely on a completed push being queryable on the relay.
 - Batch the one-time deletion-request lifecycle migration so large production databases do not remain unavailable while LMDB commits every historical request in separate transactions.
 - Prevent invitation syncing from dropping a source relay while its initial repository history is still being downloaded.
 - Fix maintainership invitation syncing by retaining and refetching expired inviter announcement and state IDs across the maintainer relay chain before promotion.

src/git/handlers.rs

@@ -325,6 +325,70 @@ where
     }
 }
 
+/// Forward receive-pack progress while retaining its protocol terminator.
+///
+/// A successful receive-pack response ends with a `0000` flush pkt-line.
+/// Git clients use that flush as the semantic end of the push and need not wait
+/// for the HTTP body to reach EOF. Retaining the final four bytes lets the
+/// caller finish GRASP post-push processing before making success visible,
+/// without buffering the progress stream that keeps clients alive during
+/// expensive pack processing.
+async fn pump_receive_pack_stdout_to_channel<R>(
+    mut stdout: R,
+    tx: &mpsc::Sender<Result<Frame<Bytes>, io::Error>>,
+) -> (PumpResult, Option<Vec<u8>>)
+where
+    R: tokio::io::AsyncRead + Unpin,
+{
+    const FLUSH_PKT: &[u8; 4] = b"0000";
+
+    let mut read_buf = [0_u8; STREAM_CHUNK_SIZE];
+    let mut pending = Vec::with_capacity(4);
+    let mut sent_stdout = false;
+
+    loop {
+        match stdout.read(&mut read_buf).await {
+            Ok(0) => {
+                if pending.as_slice() == FLUSH_PKT {
+                    return (
+                        PumpResult::Eof { sent_stdout },
+                        Some(std::mem::take(&mut pending)),
+                    );
+                }
+
+                if !pending.is_empty() {
+                    sent_stdout = true;
+                    if send_body_bytes(tx, std::mem::take(&mut pending))
+                        .await
+                        .is_err()
+                    {
+                        return (PumpResult::ClientDisconnected, None);
+                    }
+                }
+
+                return (PumpResult::Eof { sent_stdout }, None);
+            }
+            Ok(n) => {
+                pending.extend_from_slice(&read_buf[..n]);
+                if pending.len() > FLUSH_PKT.len() {
+                    let retained = pending.split_off(pending.len() - FLUSH_PKT.len());
+                    sent_stdout = true;
+                    if send_body_bytes(tx, std::mem::replace(&mut pending, retained))
+                        .await
+                        .is_err()
+                    {
+                        return (PumpResult::ClientDisconnected, None);
+                    }
+                }
+            }
+            Err(e) => {
+                let _ = tx.send(Err(e)).await;
+                return (PumpResult::ReadError, None);
+            }
+        }
+    }
+}
+
 /// Drain Git stderr for later logging or Git protocol error synthesis.
 ///
 /// Streaming handlers run this concurrently with stdout pumping so the child
@@ -637,7 +701,7 @@ async fn stream_receive_pack_output<S, E>(
     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;
+    let (pump_result, terminal_flush) = pump_receive_pack_stdout_to_channel(stdout, &tx).await;
 
     if !matches!(pump_result, PumpResult::Eof { .. }) {
         let _ = git.kill().await;
@@ -658,7 +722,7 @@ async fn stream_receive_pack_output<S, E>(
         None => Vec::new(),
     };
 
-    let sent_stdout = match pump_result {
+    let mut sent_stdout = match pump_result {
         PumpResult::Eof { sent_stdout } => sent_stdout,
         PumpResult::ClientDisconnected | PumpResult::ReadError => {
             drop(repo_lifecycle_guard);
@@ -668,6 +732,15 @@ async fn stream_receive_pack_output<S, E>(
     };
 
     if !status.success() {
+        if let Some(flush) = terminal_flush {
+            sent_stdout = true;
+            if send_body_bytes(&tx, flush).await.is_err() {
+                drop(repo_lifecycle_guard);
+                record_git_operation(&metrics, "push", "error");
+                return;
+            }
+        }
+
         drop(repo_lifecycle_guard);
         record_git_operation(&metrics, "push", "error");
         let stderr_str = String::from_utf8_lossy(&stderr_output);
@@ -707,7 +780,6 @@ async fn stream_receive_pack_output<S, E>(
     }
 
     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
@@ -718,11 +790,11 @@ async fn stream_receive_pack_output<S, E>(
     // the subprocess boundary would deadlock.
     drop(repo_lifecycle_guard);
 
-    // 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.
+    // Git's receive-pack progress has already been streamed verbatim, but its
+    // final flush remains withheld. Run GRASP follow-up work before releasing
+    // that client-visible success boundary so push completion implies local
+    // refs/HEAD and promoted events are ready. Follow-up failures remain
+    // internal/log-only rather than client-visible push rejections.
     match process_newly_available_git_data(
         &repo_path,
         &new_oids,
@@ -758,6 +830,18 @@ async fn stream_receive_pack_output<S, E>(
             );
         }
     }
+
+    // The final receive-pack flush is the client-visible success boundary.
+    // Release it only after promoted events have been saved and subscribers
+    // notified, so a completed push implies that local GRASP state is ready.
+    if let Some(flush) = terminal_flush {
+        if send_body_bytes(&tx, flush).await.is_err() {
+            record_git_operation(&metrics, "push", "error");
+            return;
+        }
+    }
+
+    record_git_operation(&metrics, "push", "success");
 }
 
 async fn send_body_bytes(

tests/git_response_streaming.rs

@@ -1,14 +1,17 @@
 //! Integration coverage for Git Smart HTTP response streaming.
 
+use std::collections::HashSet;
 use std::path::Path;
 use std::sync::Arc;
 use std::time::Duration;
 
+use async_trait::async_trait;
 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::git::sync::PurgatoryPromotionHooks;
 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};
@@ -21,6 +24,7 @@ use ngit_grasp::purgatory::Purgatory;
 use ngit_grasp::sync::rejected_index::RejectedEventsIndex;
 use nostr_relay_builder::prelude::LocalRelayBuilder;
 use nostr_sdk::prelude::*;
+use tokio::sync::Semaphore;
 use tokio::time::timeout;
 
 static PATH_ENV_LOCK: tokio::sync::Mutex<()> = tokio::sync::Mutex::const_new(());
@@ -92,7 +96,11 @@ async fn receive_pack_response_streams_stdout_before_subprocess_exit() {
         .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 first = frame_data(first);
+    assert!(
+        !first.is_empty(),
+        "first progress frame should contain data"
+    );
 
     let no_second_yet = timeout(Duration::from_millis(250), body.frame()).await;
     assert!(
@@ -101,12 +109,165 @@ async fn receive_pack_response_streams_stdout_before_subprocess_exit() {
          this test needs the first frame to be observed before subprocess EOF"
     );
 
-    let second = timeout(Duration::from_secs(3), body.frame())
+    let mut streamed = first.to_vec();
+    loop {
+        let frame = timeout(Duration::from_secs(3), body.frame())
+            .await
+            .expect("remaining stdout should arrive after fake git wakes");
+        let Some(frame) = frame else {
+            break;
+        };
+        streamed.extend_from_slice(
+            &frame
+                .expect("remaining frame should not be an HTTP body error")
+                .into_data()
+                .expect("remaining frame should contain data"),
+        );
+    }
+    assert_eq!(streamed, b"first-progress\nsecond-progress\n");
+}
+
+#[derive(Clone)]
+struct BlockingAnnouncementPromotion {
+    entered: Arc<Semaphore>,
+    release: Arc<Semaphore>,
+}
+
+#[async_trait]
+impl PurgatoryPromotionHooks for BlockingAnnouncementPromotion {
+    async fn before_announcement_promote(&self, _event: &Event, _identifier: &str) {
+        self.entered.add_permits(1);
+        self.release
+            .acquire()
+            .await
+            .expect("promotion release semaphore should remain open")
+            .forget();
+    }
+}
+
+#[tokio::test]
+async fn receive_pack_terminal_flush_waits_for_purgatory_promotion() {
+    let _env_lock = PATH_ENV_LOCK.lock().await;
+
+    let fake_bin = tempfile::tempdir().expect("fake git bin tempdir");
+    write_fake_git_with_terminal_flush(fake_bin.path());
+    let _path = PathOverride::prepend(fake_bin.path());
+
+    let keys = Keys::generate();
+    let owner_npub = keys.public_key().to_bech32().expect("encode owner npub");
+    let identifier = "push-readiness";
+    let git_data = tempfile::tempdir().expect("git data tempdir");
+    let repo_path = git_data
+        .path()
+        .join(&owner_npub)
+        .join(format!("{identifier}.git"));
+    std::fs::create_dir_all(&repo_path).expect("create fake bare repo path");
+
+    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 announcement = EventBuilder::new(Kind::GitRepoAnnouncement, "")
+        .tags(vec![Tag::identifier(identifier)])
+        .finalize(&keys)
+        .expect("build announcement");
+    purgatory.add_announcement(
+        announcement.clone(),
+        identifier.to_string(),
+        keys.public_key(),
+        repo_path.clone(),
+        HashSet::new(),
+    );
+
+    let entered = Arc::new(Semaphore::new(0));
+    let release = Arc::new(Semaphore::new(0));
+    let hooks = BlockingAnnouncementPromotion {
+        entered: entered.clone(),
+        release: release.clone(),
+    };
+
+    let response = handle_receive_pack(
+        repo_path,
+        receive_pack_request_body(),
+        database.clone(),
+        relay,
+        identifier,
+        &keys.public_key().to_hex(),
+        purgatory,
+        git_data.path().to_str().expect("utf-8 temp path"),
+        None,
+        None,
+        Some(Arc::new(hooks)),
+        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("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"));
+        .expect("receive-pack progress should stream before promotion")
+        .expect("body should contain progress")
+        .expect("progress frame should not be an HTTP body error");
+    let mut streamed = frame_data(first).to_vec();
+
+    timeout(Duration::from_secs(3), entered.acquire())
+        .await
+        .expect("post-push promotion hook should run")
+        .expect("promotion semaphore should remain open")
+        .forget();
+
+    let before_save = database
+        .query(Filter::new().id(announcement.id))
+        .await
+        .expect("query announcement before promotion");
+    assert!(
+        before_save.is_empty(),
+        "announcement must not be queryable while promotion is blocked"
+    );
+
+    while let Ok(Some(frame)) = timeout(Duration::from_millis(25), body.frame()).await {
+        streamed.extend_from_slice(
+            &frame
+                .expect("progress frame should not be an HTTP body error")
+                .into_data()
+                .expect("progress frame should contain data"),
+        );
+    }
+    assert!(
+        !streamed.ends_with(b"0000"),
+        "receive-pack terminal flush must remain hidden while promotion is blocked"
+    );
+
+    release.add_permits(1);
+
+    loop {
+        let frame = timeout(Duration::from_secs(1), body.frame())
+            .await
+            .expect("response should finish after promotion is released");
+        let Some(frame) = frame else {
+            break;
+        };
+        streamed.extend_from_slice(
+            &frame
+                .expect("terminal frame should not be an HTTP body error")
+                .into_data()
+                .expect("terminal frame should contain data"),
+        );
+    }
+
+    assert_eq!(
+        streamed, b"first-progress\nsecond-progress\n0000",
+        "terminal flush should be the final client-visible bytes"
+    );
+    let after_save = database
+        .query(Filter::new().id(announcement.id))
+        .await
+        .expect("query announcement after promotion");
+    assert_eq!(
+        after_save.len(),
+        1,
+        "announcement must be queryable before push success is exposed"
+    );
 }
 
 #[tokio::test]
@@ -251,7 +412,20 @@ fn test_write_policy(
 }
 
 fn write_fake_git(bin_dir: &Path) {
+    write_fake_git_script(bin_dir, false);
+}
+
+fn write_fake_git_with_terminal_flush(bin_dir: &Path) {
+    write_fake_git_script(bin_dir, true);
+}
+
+fn write_fake_git_script(bin_dir: &Path, terminal_flush: bool) {
     let git_path = bin_dir.join("git");
+    let terminal_flush = if terminal_flush {
+        "printf '0000'\n"
+    } else {
+        ""
+    };
     std::fs::write(
         &git_path,
         r#"#!/usr/bin/env bash
@@ -269,7 +443,7 @@ cat >/dev/null
 printf 'first-progress\n'
 sleep 2
 printf 'second-progress\n'
-exit 0
+__TERMINAL_FLUSH__exit 0
 fi
 
 if [ "$1" = "init" ]; then
@@ -284,7 +458,8 @@ fi
 
 echo "fake git only supports init, for-each-ref, and receive-pack" >&2
 exit 1
-"#,
+"#
+        .replace("__TERMINAL_FLUSH__", terminal_flush),
     )
     .expect("write fake git executable");