mirror of
https://relay.ngit.dev/npub15qydau2hjma6ngxkl2cyar74wzyjshvl65za5k5rl69264ar2exs5cyejr/ngit-grasp.git
synced 2026-10-05 23:18:24 +00:00
fix(grasp06): narrow /prs/ per-path lock, gate cleanup on in-flight count
The previous design held the per-(submitter, identifier) mutex for the
entire receive-pack pipeline — `git init --bare`, the pack upload from
the client, `git-receive-pack`, per-ref validation, and the zero-ref
cleanup. Two concurrent pushes to the same `/prs/<submitter>/<id>.git`
path were therefore fully serialised end-to-end, with the second push's
HTTP connection blocked on the mutex for the entire duration of the
first push (including its pack upload over the wire). With multiple
agents or CI jobs sharing a contributor identity this could produce
HTTP/proxy timeouts and apparently-stuck pushes for no good reason —
git's own ref locking would normally handle intra-push concurrency on
the same path just fine.
Switch to a `PrsPathState { mu: Mutex<()>, in_flight: AtomicUsize }`
per path. The receive handler now takes the mutex only briefly:
* at the start of the request to run `git init --bare` and bump
`in_flight` from 0 to 1, then release the mutex,
* at the end of the request to drop `in_flight` and, if it lands on
zero with the repo at zero refs, `rm -rf` the bare directory.
`git-receive-pack` and per-ref validation run with no per-path lock
held, so concurrent pushes by different agents to the same path proceed
in parallel.
Off-push cleanup paths (the PR-event policy when it discards a scoped
placeholder whose incoming event mismatches, and the purgatory expiry
sweep when a scoped placeholder times out) take the same mutex briefly,
delete the dangling ref, then remove the bare directory only when
`in_flight.load() == 0` and `list_refs` is empty. With both reads
performed under the mutex that gates `in_flight` mutations, a repo
deletion can never race a push that is mid-receive: either the push is
still in init/register and we wait on the mutex, or the push has
already incremented `in_flight` (so we read non-zero and skip), or the
push has finished and decremented (so the directory is genuinely idle).
The end-of-push cleanup now also runs when receive-pack returns a
protocol-error response (200 with ERR pkt-line), which the previous
code's early-return skipped — a probe push that failed git-level
validation after `ensure_repo_initialised` had created the directory
used to leak an empty `.git` dir, which now gets removed in the same
critical section.
A small `path_state(&locks, path)` helper centralises the
get-or-insert-Arc dance the three sites used to duplicate.
Docs updated:
* docs/explanation/grasp-06-contributor-pr-submission.md — replace
the "lock held for the entire push pipeline" paragraph with the
new mutex + `in_flight` discipline; update the three-sites
enumeration to describe the `in_flight == 0` guard.
* docs/explanation/architecture.md — same shape, one-paragraph.
* docs/how-to/enable-grasp-06.md — call out that the mutex is held
only briefly so concurrent pushes don't serialise.
This commit is contained in:
DanConwayDev
@@ -554,7 +554,7 @@ Optional endpoint at `/prs/<npub>/<identifier>.git`, gated on `NGIT_GRASP06_ENAB
|
||||
- [`src/grasp06/endpoint.rs`](../../src/grasp06/endpoint.rs) — URL parsing.
|
||||
- [`src/grasp06/paths.rs`](../../src/grasp06/paths.rs) — on-disk path conventions under `<git_data_path>/prs/<hex>/<identifier>.git`.
|
||||
- [`src/grasp06/fetch.rs`](../../src/grasp06/fetch.rs) — empty-repo synthesis for `info/refs` and `git-upload-pack` against repos that don't yet exist on disk.
|
||||
- [`src/grasp06/receive.rs`](../../src/grasp06/receive.rs) — `git-receive-pack` with init-on-push, strict `refs/nostr/<event-id>` ref-name validation, and per-ref post-push validation against the database and purgatory. Holds a per-`(submitter, identifier)` lock for the whole pipeline (init → receive-pack → validation → zero-ref cleanup) so a concurrent push to the same path cannot delete the bare repo mid-receive. The same lock is taken (with `try_lock` in synchronous contexts) by the PR-event policy and the purgatory expiry sweep before either of them deletes a `/prs/` ref or removes a zero-ref bare repo.
|
||||
- [`src/grasp06/receive.rs`](../../src/grasp06/receive.rs) — `git-receive-pack` with init-on-push, strict `refs/nostr/<event-id>` ref-name validation, and per-ref post-push validation against the database and purgatory. Uses a per-`(submitter, identifier)` `PrsPathState` (mutex + `in_flight` counter) so concurrent pushes to the same path proceed in parallel: the mutex is held only for init+register and decrement+end-of-push cleanup, not across `git-receive-pack` itself. The same state is consulted (with `try_lock` in synchronous contexts) by the PR-event policy and the purgatory expiry sweep, which only remove a `/prs/` ref or zero-ref bare repo when `in_flight == 0`.
|
||||
- [`src/grasp06/policy.rs`](../../src/grasp06/policy.rs) — strict clone-tag URL comparator used by the PR-event acceptance relaxation.
|
||||
|
||||
`/prs/` repos are intentionally isolated from other subsystems: empty-repo cleanup skips the `/prs/` subtree, the proactive-sync subsystem never discovers them because subscriptions are built from DB-resident announcements, and the standard repo landing page guards against ever matching a `/prs/` path. Full design: [GRASP-06 Contributor Pull Request Submission](grasp-06-contributor-pr-submission.md). Operator how-to: [Enable GRASP-06](../how-to/enable-grasp-06.md).
|
||||
|
||||
@@ -107,7 +107,7 @@ The flow mirrors the existing `refs/nostr/<event-id>` path at the standard endpo
|
||||
|
||||
#### On-demand bare repo creation
|
||||
|
||||
The first push to `/prs/<submitter>/<identifier>.git` creates the bare repo on disk. A per-`(submitter, identifier)` `tokio::sync::Mutex` (kept in a `DashMap` on the `HttpService`, see [`crate::grasp06::receive::RepoInitLocks`](../../src/grasp06/receive.rs)) is held for the entire push pipeline — on-demand `git init --bare`, `git-receive-pack`, per-ref validation, and the zero-ref cleanup at the end. Pushes to different `(submitter, identifier)` paths still run in parallel. The same lock map is also taken (via `try_lock` from synchronous contexts) by the PR-event policy and the purgatory expiry sweep before either of them inspects or removes a `/prs/` repo, so no off-push code path can delete the bare repo while a push is in flight.
|
||||
The first push to `/prs/<submitter>/<identifier>.git` creates the bare repo on disk. A per-`(submitter, identifier)` [`PrsPathState`](../../src/grasp06/receive.rs) (kept in a `DashMap` on the `HttpService`, see [`crate::grasp06::receive::RepoInitLocks`](../../src/grasp06/receive.rs)) provides a `tokio::sync::Mutex` and an `in_flight: AtomicUsize` counter. The mutex is held only briefly — for `git init --bare` plus the `in_flight` increment at the start of a push, and for the `in_flight` decrement plus end-of-push zero-ref cleanup at the end. The pack upload and per-ref validation run *without* the mutex held, so concurrent pushes to the same path proceed in parallel; git's own ref locking handles intra-push concurrency. Off-push cleanup paths (PR-event policy, purgatory expiry) take the same mutex briefly and only `rm -rf` the bare repo when they see `in_flight == 0` and `list_refs` is empty — so no off-push code path can delete the bare repo while a push is in flight, and no push is serialised behind another push to the same identity.
|
||||
|
||||
### Event acceptance relaxation
|
||||
|
||||
@@ -158,9 +158,9 @@ Placeholder entries created at `/prs/` should be validated against `(submitter =
|
||||
|
||||
A `/prs/<submitter>/<identifier>.git` bare repo can become zero-ref through three independent paths. Each is cleaned up inline at the site where the last ref is removed; there is no separate periodic sweep over `<git_data_path>/prs/`:
|
||||
|
||||
1. **Probe push leaves no valid refs.** The `/prs/` receive handler validates each pushed `refs/nostr/<event-id>` against the database and purgatory. If every ref fails validation, the bare repo is empty at the end of the push. The handler removes it before returning, under the per-path lock it has been holding since `git init --bare`.
|
||||
2. **Scoped placeholder fails validation against a later-arriving event.** When a PR event arrives whose `(signer, identifier, commit)` does not match the placeholder a `/prs/` push registered, the PR-event policy ([`src/nostr/policy/pr_event.rs`](../../src/nostr/policy/pr_event.rs)) deletes the corresponding `refs/nostr/<event-id>` ref and discards the placeholder. If that leaves the bare repo with zero refs, the policy also removes the directory. It takes the same per-path lock as the receive handler first, so it can never race with an in-flight push that is still writing other refs to the same path.
|
||||
3. **Scoped placeholder expires without a matching event.** The standard purgatory sweep ([`src/purgatory/mod.rs`](../../src/purgatory/mod.rs), every 60 seconds) walks expiring `PrPurgatoryEntry` rows. For entries with a `prs_scope`, the sweep best-effort deletes the dangling `refs/nostr/<event-id>` ref and, if that leaves the repo with zero refs, the bare repo itself. The sweep is synchronous, so it uses `try_lock` on the per-path mutex: if a push is currently in flight to that path the cleanup is skipped this cycle. In the worst case (sustained lock contention from rapid pushes) a dangling ref is left on disk; this is harmless because the repo is never zero-ref while the contended push is making progress, and any future push to the same path simply ignores the dangling ref.
|
||||
1. **Probe push leaves no valid refs.** The `/prs/` receive handler validates each pushed `refs/nostr/<event-id>` against the database and purgatory. If every ref fails validation, the bare repo is empty at the end of the push. The handler removes it before returning, under a brief end-of-push critical section on the per-path mutex (the mutex is *not* held across `git-receive-pack` itself — only across the init+register and decrement+cleanup windows).
|
||||
2. **Scoped placeholder fails validation against a later-arriving event.** When a PR event arrives whose `(signer, identifier, commit)` does not match the placeholder a `/prs/` push registered, the PR-event policy ([`src/nostr/policy/pr_event.rs`](../../src/nostr/policy/pr_event.rs)) takes the per-path mutex, deletes the corresponding `refs/nostr/<event-id>` ref, and discards the placeholder. It removes the bare directory only when `in_flight == 0` (no push is currently mid-receive) and `list_refs` reports empty.
|
||||
3. **Scoped placeholder expires without a matching event.** The standard purgatory sweep ([`src/purgatory/mod.rs`](../../src/purgatory/mod.rs), every 60 seconds) walks expiring `PrPurgatoryEntry` rows. For entries with a `prs_scope`, the sweep best-effort deletes the dangling `refs/nostr/<event-id>` ref and — under the same `in_flight == 0` guard — the bare repo itself. The sweep is synchronous, so it uses `try_lock` on the per-path mutex: if a push is briefly holding the mutex (during init or end-of-push), the cleanup is skipped this cycle. In the worst case a dangling ref is left on disk; this is harmless because any future push to the same path simply ignores it, and the next purgatory sweep will retry.
|
||||
|
||||
The shared lock map is wired through:
|
||||
|
||||
@@ -168,7 +168,7 @@ The shared lock map is wired through:
|
||||
- [`PolicyContext`](../../src/nostr/policy/mod.rs) holds the same `Arc<DashMap<…>>` so the PR-event policy can lock without going through HTTP.
|
||||
- [`Purgatory::set_prs_cleanup_ctx`](../../src/purgatory/mod.rs) wires the lock map (plus `git_data_path`) into purgatory at startup so the expiry sweep has everything it needs to act on a scoped placeholder.
|
||||
|
||||
Off-push deletion paths and the receive handler therefore share a single source of truth for "is a push currently in flight to this `(submitter, identifier)`".
|
||||
Off-push deletion paths and the receive handler therefore share a single source of truth for "is a push currently in flight to this `(submitter, identifier)`" — the per-path `in_flight` counter, read under the same mutex that gates its updates.
|
||||
|
||||
## Flow examples
|
||||
|
||||
|
||||
@@ -65,7 +65,7 @@ One bare repo per `(submitter, identifier)` combination under `<git_data_path>/p
|
||||
- When a PR event arrives that fails validation against a scoped `/prs/` placeholder, the corresponding `refs/nostr/<event-id>` ref is deleted; if that leaves the repo with zero refs the directory is removed in the same step.
|
||||
- When a scoped placeholder expires from purgatory without a matching PR event (default 30 minutes), the standard purgatory sweep deletes the dangling ref and, if it leaves the repo zero-ref, the bare repo itself.
|
||||
|
||||
All three sites share the same per-`(submitter, identifier)` mutex so they cannot race with an in-flight push.
|
||||
All three sites coordinate through a per-`(submitter, identifier)` mutex and an `in_flight` counter; cleanup paths only `rm -rf` the bare repo while `in_flight == 0`, so an in-flight push cannot have its repo deleted out from under it. The mutex is held only briefly at each site — pack uploads and per-ref validation run lock-free, so concurrent pushes by multiple agents using the same identity do not serialise behind each other.
|
||||
|
||||
There is no quota in this release. Disk consumption is bounded only by the rate at which contributors push valid PR refs.
|
||||
|
||||
|
||||
+153
-93
@@ -12,12 +12,14 @@
|
||||
//! exact shape `refs/nostr/<64-lowercase-hex>`. The repo on disk is not
|
||||
//! touched in this path — probe pushes that would have been rejected
|
||||
//! leave no trace.
|
||||
//! 2. Acquires the per-path lock from [`RepoInitLocks`] for the rest of
|
||||
//! the request, then creates the bare repo on demand at the GRASP-06
|
||||
//! path. Holding the lock for the whole pipeline guarantees a
|
||||
//! concurrent push to the same `(submitter, identifier)` cannot delete
|
||||
//! the bare repo out from under this one in step 5. Different paths
|
||||
//! still run in parallel.
|
||||
//! 2. Acquires the per-path coordination state from [`RepoInitLocks`]
|
||||
//! briefly: under its mutex it runs the on-demand `git init --bare`
|
||||
//! and increments the `in_flight` counter, then releases the mutex.
|
||||
//! Steps 3 and 4 run *without* the per-path lock so concurrent pushes
|
||||
//! to the same `(submitter, identifier)` proceed in parallel — git's
|
||||
//! 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.
|
||||
//! 3. Runs `git-receive-pack` against the repo, mirroring the subprocess
|
||||
//! plumbing in [`crate::git::handlers::handle_receive_pack`].
|
||||
//! 4. For each accepted `refs/nostr/<event-id>` ref, validates against the
|
||||
@@ -27,15 +29,18 @@
|
||||
//! nor purgatory knows about the event, a PR placeholder is added so the
|
||||
//! standard 30-minute purgatory sweep can clean it up if the event never
|
||||
//! arrives.
|
||||
//! 5. After validation, the repo is removed if it has zero refs left —
|
||||
//! discarding probe pushes that produced no valid state. Safe to do
|
||||
//! under the lock acquired in step 2.
|
||||
//! 5. 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.
|
||||
//! 6. 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::path::{Path, PathBuf};
|
||||
use std::sync::atomic::{AtomicUsize, Ordering};
|
||||
use std::sync::Arc;
|
||||
|
||||
use dashmap::DashMap;
|
||||
@@ -60,20 +65,46 @@ use crate::nostr::builder::{Nip34WritePolicy, SharedDatabase};
|
||||
use crate::purgatory::Purgatory;
|
||||
use crate::sync::rejected_index::RejectedEventsIndex;
|
||||
|
||||
/// Shared lock map serialising the full `/prs/<submitter>/<identifier>.git`
|
||||
/// receive-pack pipeline.
|
||||
/// Per-`(submitter, identifier)` coordination state shared between
|
||||
/// `/prs/` receive-pack pushes and the cleanup paths that may delete the
|
||||
/// bare repo.
|
||||
///
|
||||
/// The lock is held for the entire request — `git init --bare`,
|
||||
/// `git-receive-pack`, per-ref validation, and the post-validation
|
||||
/// "remove if zero refs" cleanup — so that two simultaneous pushes to the
|
||||
/// same path cannot have one push delete the bare repo while the other is
|
||||
/// mid-receive. Pushes to *different* paths still proceed in parallel.
|
||||
/// `mu` is held only briefly:
|
||||
///
|
||||
/// The same lock is taken by code paths outside the receive handler
|
||||
/// (purgatory expiry, scoped-placeholder mismatch handling) before they
|
||||
/// inspect or delete the bare repo, so all callers serialise against
|
||||
/// in-flight pushes.
|
||||
pub type RepoInitLocks = Arc<DashMap<PathBuf, Arc<Mutex<()>>>>;
|
||||
/// * 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 off-push cleanup paths (PR-event validation discard, purgatory
|
||||
/// expiry) for the duration of one `delete_ref` + optional
|
||||
/// `remove_dir_all`.
|
||||
///
|
||||
/// `git-receive-pack` itself and per-ref validation run *without* the
|
||||
/// mutex held, so two pushes to the same path proceed in parallel; git's
|
||||
/// own ref locking handles intra-push concurrency.
|
||||
///
|
||||
/// Off-push cleanup paths only `rm -rf` the bare repo when both
|
||||
/// `in_flight.load() == 0` *and* `list_refs` returns empty while they
|
||||
/// hold `mu` — the same mutex that gates `in_flight` updates — so a
|
||||
/// repo can never be deleted while a push is mid-receive.
|
||||
pub struct PrsPathState {
|
||||
pub mu: Mutex<()>,
|
||||
pub in_flight: AtomicUsize,
|
||||
}
|
||||
|
||||
impl PrsPathState {
|
||||
fn new() -> Self {
|
||||
Self {
|
||||
mu: Mutex::new(()),
|
||||
in_flight: AtomicUsize::new(0),
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/// Shared per-path state map for the GRASP-06 `/prs/` endpoint. See
|
||||
/// [`PrsPathState`] for the locking discipline.
|
||||
pub type RepoInitLocks = Arc<DashMap<PathBuf, Arc<PrsPathState>>>;
|
||||
|
||||
/// Create a fresh, empty [`RepoInitLocks`] for use as an [`HttpService`]
|
||||
/// field.
|
||||
@@ -83,6 +114,17 @@ pub fn new_repo_init_locks() -> RepoInitLocks {
|
||||
Arc::new(DashMap::new())
|
||||
}
|
||||
|
||||
/// Look up (or insert) the [`PrsPathState`] for `repo_path` in `locks`.
|
||||
/// Used by off-push cleanup paths so they take the same Arc the receive
|
||||
/// handler will see.
|
||||
pub fn path_state(locks: &RepoInitLocks, repo_path: &Path) -> Arc<PrsPathState> {
|
||||
locks
|
||||
.entry(repo_path.to_path_buf())
|
||||
.or_insert_with(|| Arc::new(PrsPathState::new()))
|
||||
.value()
|
||||
.clone()
|
||||
}
|
||||
|
||||
/// Handle `POST /prs/<npub>/<identifier>.git/git-receive-pack`.
|
||||
///
|
||||
/// See the module-level docs for the full algorithm. All application-level
|
||||
@@ -135,95 +177,113 @@ pub async fn handle_prs_receive_pack(
|
||||
}
|
||||
}
|
||||
|
||||
// 2. Acquire the per-path lock for the rest of the request. The lock
|
||||
// covers `git init --bare`, `git-receive-pack`, per-ref validation
|
||||
// and the zero-ref cleanup so a concurrent push to the same
|
||||
// `(submitter, identifier)` cannot have one push delete the bare
|
||||
// repo while the other is mid-receive. Different paths still run
|
||||
// in parallel — the DashMap entry is per `repo_path`.
|
||||
// 2. Acquire the per-path coordination state and, under its mutex,
|
||||
// initialise the bare repo on demand and register this request as
|
||||
// in-flight. The mutex is then released — `git-receive-pack` and
|
||||
// per-ref validation run WITHOUT the lock so concurrent pushes to
|
||||
// the same `(submitter, identifier)` proceed in parallel. Cleanup
|
||||
// paths consult `in_flight` (under the same mutex) before
|
||||
// deleting the bare repo, so a repo can never vanish mid-receive.
|
||||
let repo_path = prs_repo_path(
|
||||
Path::new(git_data_path),
|
||||
&prs.submitter.to_hex(),
|
||||
&prs.identifier,
|
||||
);
|
||||
let path_lock = repo_init_locks
|
||||
.entry(repo_path.clone())
|
||||
.or_insert_with(|| Arc::new(Mutex::new(())))
|
||||
.value()
|
||||
.clone();
|
||||
let _path_guard = path_lock.lock().await;
|
||||
let state = path_state(&repo_init_locks, &repo_path);
|
||||
|
||||
if let Err(e) = ensure_repo_initialised(&repo_path).await {
|
||||
error!(
|
||||
"/prs/ receive-pack: failed to initialise repo at {}: {}",
|
||||
repo_path.display(),
|
||||
e
|
||||
);
|
||||
return Err(e);
|
||||
{
|
||||
let _g = state.mu.lock().await;
|
||||
if let Err(e) = ensure_repo_initialised(&repo_path).await {
|
||||
error!(
|
||||
"/prs/ receive-pack: failed to initialise repo at {}: {}",
|
||||
repo_path.display(),
|
||||
e
|
||||
);
|
||||
return Err(e);
|
||||
}
|
||||
state.in_flight.fetch_add(1, Ordering::Relaxed);
|
||||
}
|
||||
|
||||
// 3. Run git-receive-pack against the now-existing repo.
|
||||
let response = run_receive_pack(&repo_path, &request_body, git_protocol).await?;
|
||||
// 3 + 4: run receive-pack and 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<Response<Full<Bytes>>, 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);
|
||||
// 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);
|
||||
}
|
||||
|
||||
// 4. Per-ref post-push validation. Iterate over the same parsed
|
||||
// ref list — at this point every ref name is known to be
|
||||
// `refs/nostr/<event-id>`, so the strip is infallible.
|
||||
for (_, new_oid, ref_name) in &pushed_refs {
|
||||
let event_id_hex = ref_name
|
||||
.strip_prefix("refs/nostr/")
|
||||
.expect("ref shape validated above");
|
||||
validate_pushed_nostr_ref(
|
||||
&database,
|
||||
&purgatory,
|
||||
&repo_path,
|
||||
prs,
|
||||
event_id_hex,
|
||||
new_oid,
|
||||
)
|
||||
.await;
|
||||
}
|
||||
|
||||
Ok(response)
|
||||
}
|
||||
.await;
|
||||
|
||||
// 4. Per-ref post-push validation. Iterate over the same parsed ref
|
||||
// list — at this point every ref name is known to be
|
||||
// `refs/nostr/<event-id>`, so the strip is infallible.
|
||||
for (_, new_oid, ref_name) in &pushed_refs {
|
||||
let event_id_hex = ref_name
|
||||
.strip_prefix("refs/nostr/")
|
||||
.expect("ref shape validated above");
|
||||
validate_pushed_nostr_ref(
|
||||
&database,
|
||||
&purgatory,
|
||||
&repo_path,
|
||||
prs,
|
||||
event_id_hex,
|
||||
new_oid,
|
||||
)
|
||||
.await;
|
||||
}
|
||||
|
||||
// 5. If validation removed every ref, the repo is empty: a probe push
|
||||
// that left no valid state. Delete the directory so we don't
|
||||
// accumulate empty repos.
|
||||
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 {} (probe push left no valid refs)",
|
||||
repo_path.display()
|
||||
);
|
||||
// 5. 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().await;
|
||||
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?;
|
||||
|
||||
// 6. Drive the standard purgatory-release pipeline so PR events
|
||||
// already waiting on these commits can be promoted out of
|
||||
// purgatory. We pass the `/prs/` repo path through unchanged —
|
||||
// the cross-service mirror lives elsewhere and is a future
|
||||
// addition.
|
||||
let new_oids: HashSet<String> = pushed_refs
|
||||
.iter()
|
||||
.filter(|(_, new_oid, _)| new_oid != "0000000000000000000000000000000000000000")
|
||||
.map(|(_, new_oid, _)| new_oid.clone())
|
||||
.collect();
|
||||
// purgatory. Only fires on a successful push, and only if the
|
||||
// repo still exists (it may have been removed in step 5).
|
||||
if response.status() == StatusCode::OK && repo_path.exists() {
|
||||
let new_oids: HashSet<String> = pushed_refs
|
||||
.iter()
|
||||
.filter(|(_, new_oid, _)| {
|
||||
new_oid != "0000000000000000000000000000000000000000"
|
||||
})
|
||||
.map(|(_, new_oid, _)| new_oid.clone())
|
||||
.collect();
|
||||
|
||||
if repo_path.exists() {
|
||||
if let Err(e) = process_newly_available_git_data(
|
||||
&repo_path,
|
||||
&new_oids,
|
||||
@@ -280,7 +340,7 @@ fn invalid_ref_reason(ref_name: &str) -> Option<String> {
|
||||
/// `mkdir -p` the parent and `git init --bare --initial-branch=main
|
||||
/// --quiet` into `repo_path` if it does not already exist.
|
||||
///
|
||||
/// The caller must hold the per-path lock from [`RepoInitLocks`] for
|
||||
/// The caller must hold the per-path mutex from [`PrsPathState`] for
|
||||
/// `repo_path` before invoking this function.
|
||||
async fn ensure_repo_initialised(repo_path: &Path) -> Result<(), GitError> {
|
||||
if repo_path.exists() {
|
||||
|
||||
@@ -111,11 +111,13 @@ impl PrEventPolicy {
|
||||
);
|
||||
|
||||
// Delete the ref the original /prs/ push wrote, and if
|
||||
// that leaves the bare repo with zero refs, delete the
|
||||
// directory too. Held under the same per-path lock the
|
||||
// /prs/ receive handler uses so a concurrent push to
|
||||
// this `(submitter, identifier)` cannot have its work
|
||||
// wiped mid-receive.
|
||||
// that leaves the bare repo with zero refs *and no
|
||||
// push is in flight to this path*, delete the
|
||||
// directory too. The per-path mutex is held only for
|
||||
// the brief delete-and-check window — concurrent
|
||||
// pushes increment `in_flight` under the same mutex,
|
||||
// so seeing `in_flight == 0` while holding the lock
|
||||
// is a safe signal that no push is mid-receive.
|
||||
let prs_repo = crate::grasp06::paths::prs_repo_path(
|
||||
&self.ctx.git_data_path,
|
||||
&scope.submitter.to_hex(),
|
||||
@@ -123,16 +125,11 @@ impl PrEventPolicy {
|
||||
);
|
||||
let ref_name = format!("refs/nostr/{}", event_id);
|
||||
if prs_repo.exists() {
|
||||
let lock = self
|
||||
.ctx
|
||||
.repo_init_locks
|
||||
.entry(prs_repo.clone())
|
||||
.or_insert_with(|| {
|
||||
std::sync::Arc::new(tokio::sync::Mutex::new(()))
|
||||
})
|
||||
.value()
|
||||
.clone();
|
||||
let _guard = lock.lock().await;
|
||||
let state = crate::grasp06::receive::path_state(
|
||||
&self.ctx.repo_init_locks,
|
||||
&prs_repo,
|
||||
);
|
||||
let _guard = state.mu.lock().await;
|
||||
|
||||
if let Err(e) = crate::git::delete_ref(&prs_repo, &ref_name) {
|
||||
tracing::warn!(
|
||||
@@ -141,15 +138,20 @@ impl PrEventPolicy {
|
||||
error = %e,
|
||||
"Failed to delete /prs/ ref while discarding scoped placeholder",
|
||||
);
|
||||
} else if matches!(
|
||||
crate::git::list_refs(&prs_repo),
|
||||
Ok(refs) if refs.is_empty()
|
||||
) {
|
||||
// No refs left — the repo is now an empty
|
||||
// husk. Drop the bare dir so /prs/ doesn't
|
||||
// accumulate abandoned repos. Equivalent to
|
||||
// the cleanup the receive handler already
|
||||
// does on push completion.
|
||||
} else if state
|
||||
.in_flight
|
||||
.load(std::sync::atomic::Ordering::Relaxed)
|
||||
== 0
|
||||
&& matches!(
|
||||
crate::git::list_refs(&prs_repo),
|
||||
Ok(refs) if refs.is_empty()
|
||||
)
|
||||
{
|
||||
// No refs left and no push in flight — the
|
||||
// repo is now an empty husk. Drop the bare
|
||||
// dir so /prs/ doesn't accumulate abandoned
|
||||
// repos. Equivalent to the cleanup the
|
||||
// receive handler does on push completion.
|
||||
if let Err(e) = std::fs::remove_dir_all(&prs_repo) {
|
||||
tracing::warn!(
|
||||
repo = %prs_repo.display(),
|
||||
|
||||
+29
-26
@@ -212,9 +212,11 @@ pub struct Purgatory {
|
||||
/// best-effort empty-repo cleanup the `/prs/` receive handler does, for
|
||||
/// placeholders that expire because no matching PR event ever arrived.
|
||||
///
|
||||
/// The lock map is the same one held by the receive handler, so a sync
|
||||
/// `try_lock` here is enough to know whether a push is in flight; if it
|
||||
/// is, the cleanup is deferred to the next [`Purgatory::cleanup`] cycle.
|
||||
/// The lock map is the same one held by the receive handler. The
|
||||
/// cleanup loop is synchronous, so it `try_lock`s the per-path mutex —
|
||||
/// holding it across an async point would block the runtime — and
|
||||
/// reads the `in_flight` counter under that mutex to decide whether a
|
||||
/// push is mid-receive.
|
||||
#[derive(Clone)]
|
||||
pub struct PrsCleanupCtx {
|
||||
pub git_data_path: PathBuf,
|
||||
@@ -1294,16 +1296,15 @@ impl Purgatory {
|
||||
|
||||
// For GRASP-06 `/prs/`-scoped placeholders, best-effort
|
||||
// filesystem cleanup of the dangling refs/nostr/<event-id>
|
||||
// ref and (if it leaves the bare repo with zero refs) the
|
||||
// repo directory itself. The cleanup is gated on a
|
||||
// `try_lock` of the same per-path mutex the receive handler
|
||||
// uses, so it can never race with an in-flight push: if a
|
||||
// push to this `(submitter, identifier)` is currently
|
||||
// holding the lock, the cleanup is skipped this cycle and
|
||||
// the dangling ref is simply left on disk. The receive
|
||||
// handler will re-evaluate cleanup at the end of its
|
||||
// current push, and any future push to the same path will
|
||||
// ignore the dangling ref.
|
||||
// ref and (if it leaves the bare repo with zero refs *and
|
||||
// no push is in flight*) the repo directory itself. The
|
||||
// cleanup is gated on a `try_lock` of the per-path mutex:
|
||||
// if a push is currently holding the lock the cleanup is
|
||||
// skipped this cycle and the dangling ref is simply left
|
||||
// on disk. With the lock held, the `in_flight` check is
|
||||
// what distinguishes "no push active" from "push mid-receive
|
||||
// but the mutex is briefly free between init and end-of-push"
|
||||
// — only the former is safe to follow with a `rm -rf`.
|
||||
if let (Some(scope), Some(ctx)) =
|
||||
(prs_scope.as_ref(), self.prs_cleanup_ctx.get())
|
||||
{
|
||||
@@ -1313,14 +1314,11 @@ impl Purgatory {
|
||||
&scope.identifier,
|
||||
);
|
||||
if repo_path.exists() {
|
||||
let lock = ctx
|
||||
.repo_init_locks
|
||||
.entry(repo_path.clone())
|
||||
.or_insert_with(|| Arc::new(tokio::sync::Mutex::new(())))
|
||||
.value()
|
||||
.clone();
|
||||
let guard = lock.try_lock();
|
||||
match guard {
|
||||
let state = crate::grasp06::receive::path_state(
|
||||
&ctx.repo_init_locks,
|
||||
&repo_path,
|
||||
);
|
||||
match state.mu.try_lock() {
|
||||
Ok(_guard) => {
|
||||
let ref_name = format!("refs/nostr/{}", event_id_str);
|
||||
if let Err(e) = crate::git::delete_ref(&repo_path, &ref_name) {
|
||||
@@ -1330,10 +1328,15 @@ impl Purgatory {
|
||||
error = %e,
|
||||
"Failed to delete dangling /prs/ ref during purgatory expiry",
|
||||
);
|
||||
} else if matches!(
|
||||
crate::git::list_refs(&repo_path),
|
||||
Ok(refs) if refs.is_empty()
|
||||
) {
|
||||
} else if state
|
||||
.in_flight
|
||||
.load(std::sync::atomic::Ordering::Relaxed)
|
||||
== 0
|
||||
&& matches!(
|
||||
crate::git::list_refs(&repo_path),
|
||||
Ok(refs) if refs.is_empty()
|
||||
)
|
||||
{
|
||||
if let Err(e) = std::fs::remove_dir_all(&repo_path) {
|
||||
tracing::warn!(
|
||||
repo = %repo_path.display(),
|
||||
@@ -1354,7 +1357,7 @@ impl Purgatory {
|
||||
"/prs/ repo lock contended during purgatory expiry — skipping filesystem cleanup this cycle",
|
||||
);
|
||||
}
|
||||
}
|
||||
};
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
Reference in New Issue
Block a user