Skip to content

Commit b35cb31

Browse files
committed
fix(node): address jatmn review on merged advisory-lock head
Align .env.example db pool with validate floor, move snapshot cleanup into the blocking extract task, remove fork clones on PreconditionLost, restore upload_site_reached test plumbing, and use owner DIDs in receive-pack tests after #330 owner-push default.
1 parent 4cee7e0 commit b35cb31

4 files changed

Lines changed: 70 additions & 39 deletions

File tree

‎.env.example‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -25,7 +25,7 @@ DATABASE_URL=postgresql://gitlawb:changeme@localhost:5432/gitlawb
2525
# Maximum connections in the PostgreSQL pool. A cap, not a floor —
2626
# connections open lazily. Size against the DB server's max_connections,
2727
# remembering admin tooling opens its own pool.
28-
GITLAWB_DB_MAX_CONNECTIONS=20
28+
GITLAWB_DB_MAX_CONNECTIONS=48
2929
# Maximum connections in the DEDICATED advisory-lock pool, separate from the
3030
# pool above. Every in-flight repo write pins one connection here for its whole
3131
# duration, so this is a hard ceiling on simultaneous writes node-wide: size it

‎crates/gitlawb-node/src/api/repos.rs‎

Lines changed: 13 additions & 6 deletions
Original file line numberDiff line numberDiff line change
@@ -3149,8 +3149,8 @@ pub async fn fork_repo(
31493149
// archive already sits under this key (a failed create_repo or another
31503150
// writer), and proceeding would create a DB record whose archive is shadowed
31513151
// by bytes that are not this fork. Refuse rather than accept a fork other
3152-
// nodes would fetch as unrelated content. The local clone is cleaned up on
3153-
// this path by the caller's error return dropping the handler state.
3152+
// nodes would fetch as unrelated content. Remove the local mirror clone so
3153+
// a refused fork does not leave a bare repo on disk.
31543154
state
31553155
.repo_store
31563156
.release_after_write(&forker_did, &fork_name)
@@ -3163,6 +3163,13 @@ pub async fn fork_repo(
31633163
status,
31643164
"fork refused: an archive already exists under the fork's key"
31653165
);
3166+
if let Err(clean) = std::fs::remove_dir_all(&disk_path) {
3167+
tracing::warn!(
3168+
path = %disk_path.display(),
3169+
err = %clean,
3170+
"failed to remove fork clone after precondition loss"
3171+
);
3172+
}
31663173
AppError::RepoExists(fork_name.clone())
31673174
}
31683175
other => AppError::Git(format!("fork upload failed: {other}")),
@@ -6402,7 +6409,7 @@ mod tests {
64026409
State(state.clone()),
64036410
Path((owner.to_string(), name.to_string())),
64046411
Extension(crate::auth::AuthenticatedDid(
6405-
"did:key:z6MkDisconnectWriteLockProofDidAAAAAAAA".to_string(),
6412+
"did:key:z6disc".to_string(),
64066413
)),
64076414
crate::rate_limit::PeerAddr(Some(
64086415
"203.0.113.81:5000".parse::<SocketAddr>().unwrap(),
@@ -6554,7 +6561,7 @@ mod tests {
65546561
State(state.clone()),
65556562
Path((owner.to_string(), name.to_string())),
65566563
Extension(crate::auth::AuthenticatedDid(
6557-
"did:key:z6MkPushSuccessReleaseProofDidAAAAAAAA".to_string(),
6564+
"did:key:z6succ".to_string(),
65586565
)),
65596566
crate::rate_limit::PeerAddr(Some(
65606567
"203.0.113.83:5000".parse::<SocketAddr>().unwrap(),
@@ -6630,7 +6637,7 @@ mod tests {
66306637
.await
66316638
.unwrap();
66326639

6633-
let did = "did:key:z6MkLockPoolShedProofDidAAAAAAAAAAAAAAAAAA";
6640+
let did = "did:key:z6lockpool";
66346641
let peer: SocketAddr = "203.0.113.71:5000".parse().unwrap();
66356642

66366643
// Occupy the only lock-pool connection with a write on an UNRELATED repo.
@@ -6657,7 +6664,7 @@ mod tests {
66576664

66586665
// MUST-NOT: with the pool free again, the push is not shed as capacity (it fails
66596666
// later on the nonexistent on-disk repo, which is a git error, not Overloaded).
6660-
held.release(false).await;
6667+
let _ = held.release(false).await;
66616668
let admitted = git_receive_pack(
66626669
State(state.clone()),
66636670
Path((owner.to_string(), name.to_string())),

‎crates/gitlawb-node/src/git/repo_store.rs‎

Lines changed: 26 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -47,6 +47,10 @@ pub struct RepoStore {
4747
/// `RepoWriteGuard::test_pre_unlock_gate`. Never set outside tests.
4848
#[cfg(test)]
4949
pre_unlock_gate: Option<Arc<tokio::sync::Notify>>,
50+
/// Per store rather than a process global, so parallel tests do not see each
51+
/// other's uploads. See [`RepoStore::tigris_upload_site_reached`].
52+
#[cfg(test)]
53+
upload_site_reached: Arc<std::sync::atomic::AtomicUsize>,
5054
}
5155

5256
impl RepoStore {
@@ -60,6 +64,7 @@ impl RepoStore {
6064
lock_acquire_deadline: LOCK_ACQUIRE_DEADLINE,
6165
migrated: Arc::new(tokio::sync::Mutex::new(std::collections::HashSet::new())),
6266
pre_unlock_gate: None,
67+
upload_site_reached: Arc::new(std::sync::atomic::AtomicUsize::new(0)),
6368
}
6469
}
6570

@@ -91,6 +96,14 @@ impl RepoStore {
9196
self
9297
}
9398

99+
/// Test-only: how many write guards from this store have reached the Tigris upload
100+
/// site. See [`RepoWriteGuard::release`].
101+
#[cfg(test)]
102+
pub fn tigris_upload_site_reached(&self) -> usize {
103+
self.upload_site_reached
104+
.load(std::sync::atomic::Ordering::SeqCst)
105+
}
106+
94107
pub fn new(
95108
repos_dir: PathBuf,
96109
tigris: Option<TigrisClient>,
@@ -106,6 +119,8 @@ impl RepoStore {
106119
migrated: Arc::new(Mutex::new(HashSet::new())),
107120
#[cfg(test)]
108121
pre_unlock_gate: None,
122+
#[cfg(test)]
123+
upload_site_reached: Arc::new(std::sync::atomic::AtomicUsize::new(0)),
109124
}
110125
}
111126

@@ -448,6 +463,8 @@ impl RepoStore {
448463
publish_fence: UploadPrecondition::Unconditional,
449464
#[cfg(test)]
450465
test_pre_unlock_gate: self.pre_unlock_gate.clone(),
466+
#[cfg(test)]
467+
upload_site_reached: Arc::clone(&self.upload_site_reached),
451468
};
452469

453470
// Always download the latest from Tigris before writing. Local disk may be
@@ -1040,6 +1057,8 @@ pub struct RepoWriteGuard {
10401057
/// frees the session lock. Never set outside tests.
10411058
#[cfg(test)]
10421059
test_pre_unlock_gate: Option<Arc<tokio::sync::Notify>>,
1060+
#[cfg(test)]
1061+
upload_site_reached: Arc<std::sync::atomic::AtomicUsize>,
10431062
}
10441063

10451064
impl RepoWriteGuard {
@@ -1150,6 +1169,9 @@ impl RepoWriteGuard {
11501169
let mut outcome = ReleaseOutcome::Released;
11511170
// Upload to Tigris only on success.
11521171
if success {
1172+
#[cfg(test)]
1173+
self.upload_site_reached
1174+
.fetch_add(1, std::sync::atomic::Ordering::SeqCst);
11531175
if let Some(tigris) = self.tigris.clone() {
11541176
// ONE budget for the whole publish, covering both attempts and
11551177
// the HEAD between them, for the same reason the acquire-side
@@ -2489,6 +2511,8 @@ mod tests {
24892511
publish_fence: UploadPrecondition::Unconditional,
24902512
#[cfg(test)]
24912513
test_pre_unlock_gate: None,
2514+
#[cfg(test)]
2515+
upload_site_reached: Arc::new(std::sync::atomic::AtomicUsize::new(0)),
24922516
};
24932517
// Must complete without panic and issue no unlock.
24942518
let _ = guard.release(false).await;
@@ -2851,6 +2875,8 @@ mod tests {
28512875
publish_fence: UploadPrecondition::Unconditional,
28522876
#[cfg(test)]
28532877
test_pre_unlock_gate: None,
2878+
#[cfg(test)]
2879+
upload_site_reached: Arc::new(std::sync::atomic::AtomicUsize::new(0)),
28542880
};
28552881
let _ = guard.release(true).await;
28562882

‎crates/gitlawb-node/src/git/tigris.rs‎

Lines changed: 30 additions & 32 deletions
Original file line numberDiff line numberDiff line change
@@ -271,10 +271,8 @@ impl TigrisClient {
271271
.into_bytes();
272272

273273
// The snapshot temp dir is decided HERE, in the async layer, before the
274-
// extraction runs. The extraction itself is uncancellable spawn_blocking,
275-
// so the dir is created no matter what happens to this future; an
276-
// async-layer guard that owns the path removes it when this future is
277-
// dropped mid-await (client disconnect, a bounded-transfer timeout).
274+
// extraction runs. Cleanup ownership moves into the blocking task so a
275+
// cancelled async future cannot drop it while extraction is still running.
278276
let snapshot_tmp = if publish {
279277
None
280278
} else {
@@ -289,48 +287,48 @@ impl TigrisClient {
289287
uuid::Uuid::new_v4()
290288
)))
291289
};
292-
// Armed before the extraction await; disarmed on the success return via
293-
// `mem::forget`, leaving the dir to the caller (RepoSnapshot::drop). On
294-
// any other exit, including a cancelled future, the guard removes the
295-
// dir. This is what closes the leak where a dropped read_snapshot future
296-
// abandons a completed extraction.
297-
let _cleanup = snapshot_tmp.as_ref().map(|p| SnapshotCleanup(p.clone()));
298290

299291
// Extract tar.zst to a directory.
300292
let extracted = tokio::task::spawn_blocking({
301293
let target = target.to_path_buf();
302294
let snapshot_tmp = snapshot_tmp.clone();
303295
move || -> Result<PathBuf> {
304-
if publish {
305-
decompress_repo(&data, &target)?;
306-
return Ok(target);
307-
}
308-
// Non-mutating snapshot: unpack into the temp dir decided above.
309-
// The live repo path is never touched.
310-
let tmp_dir = snapshot_tmp.expect("snapshot path was decided above");
311-
std::fs::create_dir_all(&tmp_dir).context("creating temp extract dir")?;
312-
let unpack = (|| -> Result<()> {
313-
let decoder = zstd::stream::Decoder::new(&data[..])?;
314-
let mut archive = tar::Archive::new(decoder);
315-
archive.unpack(&tmp_dir).context("unpacking tar.zst")?;
316-
Ok(())
296+
// Armed for the whole blocking extraction; disarmed on success via
297+
// `mem::forget`, leaving the dir to the caller (RepoSnapshot::drop).
298+
let cleanup = snapshot_tmp.as_ref().map(|p| SnapshotCleanup(p.clone()));
299+
let result = (|| -> Result<PathBuf> {
300+
if publish {
301+
decompress_repo(&data, &target)?;
302+
return Ok(target);
303+
}
304+
// Non-mutating snapshot: unpack into the temp dir decided above.
305+
// The live repo path is never touched.
306+
let tmp_dir = snapshot_tmp.expect("snapshot path was decided above");
307+
std::fs::create_dir_all(&tmp_dir).context("creating temp extract dir")?;
308+
let unpack = (|| -> Result<()> {
309+
let decoder = zstd::stream::Decoder::new(&data[..])?;
310+
let mut archive = tar::Archive::new(decoder);
311+
archive.unpack(&tmp_dir).context("unpacking tar.zst")?;
312+
Ok(())
313+
})();
314+
if let Err(e) = unpack {
315+
let _ = std::fs::remove_dir_all(&tmp_dir);
316+
return Err(e);
317+
}
318+
Ok(tmp_dir)
317319
})();
318-
if let Err(e) = unpack {
319-
let _ = std::fs::remove_dir_all(&tmp_dir);
320-
return Err(e);
320+
if result.is_ok() {
321+
if let Some(guard) = cleanup {
322+
std::mem::forget(guard);
323+
}
321324
}
322-
Ok(tmp_dir)
325+
result
323326
}
324327
})
325328
.await
326329
.context("extract task panicked")?
327330
.context("extracting repo")?;
328331

329-
// The dir is now the caller's to own: drop the cleanup guard without
330-
// removing anything. On a future drop before this point, `_cleanup` runs
331-
// and removes the dir even though the extraction completed.
332-
std::mem::forget(_cleanup);
333-
334332
info!(key = %key, path = %target.display(), "downloaded repo from tigris");
335333
Ok(extracted)
336334
}

0 commit comments

Comments
 (0)