Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
1 change: 1 addition & 0 deletions crates/walgit-git/src/lib.rs
Original file line number Diff line number Diff line change
Expand Up @@ -2282,6 +2282,7 @@ impl LocalRepo {
// of the history pack and made pack generation copy 241 k tree
// entries through the FUSE-linked base (prod: 23-minute clones with a
// 2 s enumeration). Blobs still resolve to the base.
let _writer = midx::lock_midx_writer(self)?;
let packs = self.packs()?;
let history: Vec<&PackInfo> = packs.iter().filter(|p| p.history_of.is_some()).collect();
let midx = self.objects_pack_dir().join("multi-pack-index");
Expand Down
122 changes: 119 additions & 3 deletions crates/walgit-git/src/midx.rs
Original file line number Diff line number Diff line change
Expand Up @@ -2,6 +2,9 @@
use std::io::{Read, Seek, SeekFrom, Write};
use std::path::{Path, PathBuf};
use std::process::{Command, Stdio};
use std::time::Duration;

use tracing::warn;

use crate::{GitError, LocalRepo, ObjectId, ge};

Expand Down Expand Up @@ -38,6 +41,72 @@ fn checked(command: &mut Command, input: Option<&[u8]>) -> Result<(), GitError>
Ok(())
}

const WRITER_LOCK_WAIT: Duration = Duration::from_secs(30);

/// Exclusive, repository-local ownership of MIDX writing. The kernel drops the
/// lock when its holder dies, so a `multi-pack-index.lock` found while holding
/// it was left by a crashed writer and is removed. Age proves nothing: a bitmap
/// write over a large repository can outlive any fixed timeout.
pub(crate) struct MidxWriterLock(std::fs::File);

impl Drop for MidxWriterLock {
fn drop(&mut self) {
let _ = self.0.unlock();
}
}
Comment thread
cursor[bot] marked this conversation as resolved.

pub(crate) fn lock_midx_writer(repo: &LocalRepo) -> Result<MidxWriterLock, GitError> {
lock_midx_writer_for(repo, WRITER_LOCK_WAIT)
}

fn lock_midx_writer_for(repo: &LocalRepo, wait: Duration) -> Result<MidxWriterLock, GitError> {
let file = std::fs::OpenOptions::new()
.read(true)
.write(true)
.create(true)
.truncate(false)
.open(repo.path().join("walgit-midx.lock"))?;
let start = std::time::Instant::now();
loop {
match file.try_lock() {
Ok(()) => break,
Err(std::fs::TryLockError::WouldBlock) if start.elapsed() < wait => {
std::thread::sleep(Duration::from_millis(25));
}
Err(std::fs::TryLockError::WouldBlock) => {
return Err(GitError::Io(std::io::Error::new(
std::io::ErrorKind::WouldBlock,
"another MIDX writer is active; retry",
)));
}
Err(std::fs::TryLockError::Error(error)) => return Err(error.into()),
}
}
match std::fs::remove_file(repo.path().join("objects/pack/multi-pack-index.lock")) {
Ok(()) => warn!("removed multi-pack-index.lock left by a crashed writer"),
Err(error) if error.kind() == std::io::ErrorKind::NotFound => {}
Err(error) => return Err(error.into()),
}
Ok(MidxWriterLock(file))
}

/// Regenerate the commit-graph before building the MIDX bitmap.
///
/// Git builds the bitmap from the commit-graph; a stale graph can leave branch
/// tips out of the bitmap, and the `rev-list --test-bitmap` check below then
/// fails for an otherwise healthy MIDX.
fn refresh_commit_graph(repo: &LocalRepo) -> Result<(), GitError> {
checked(
Command::new("git").arg("-C").arg(repo.path()).args([
"commit-graph",
"write",
"--reachable",
"--split=replace",
]),
None,
)
}

fn file_digest(path: &Path, kind: gix_hash::Kind) -> Result<ObjectId, GitError> {
let mut file = std::fs::File::open(path)?;
let size = file.metadata()?.len();
Expand Down Expand Up @@ -173,6 +242,8 @@ impl LocalRepo {
));
}
}
let _writer = lock_midx_writer(&repo)?;
refresh_commit_graph(&repo)?;
let input = format!("{}\n", names.join("\n"));
checked(
Command::new("git").arg("-C").arg(repo.path()).args([
Expand All @@ -191,19 +262,24 @@ impl LocalRepo {
.args(["multi-pack-index", "verify"]),
None,
)?;
checked(
// Git's bitmap writer selects a subset of commits; a branch tip is
// not guaranteed to have a bitmap. That leaves ordinary traversal
// available and must not fail the caller's pack lifecycle.
if let Err(error) = checked(
Command::new("git").arg("-C").arg(repo.path()).args([
"rev-list",
"--test-bitmap",
&commit.to_string(),
]),
None,
)?;
) {
warn!(%error, "MIDX bitmap graph walk unavailable; ordinary traversal remains active");
}
checked(
Command::new("git").arg("-C").arg(repo.path()).args([
"config",
"pack.allowPackReuse",
"multi",
"single",
]),
None,
)?;
Expand All @@ -214,3 +290,43 @@ impl LocalRepo {
.map_err(|e| GitError::Protocol(format!("bitmap task: {e}")))?
}
}

#[cfg(test)]
#[allow(clippy::unwrap_used)]
mod tests {
use super::*;
use crate::{ObjectFormat, RepoId};

fn repo() -> (tempfile::TempDir, LocalRepo) {
let root = tempfile::tempdir().unwrap();
let repo = LocalRepo::init(
root.path(),
&RepoId::new("test", "midx-lock").unwrap(),
ObjectFormat::Sha1,
)
.unwrap();
(root, repo)
}

#[test]
fn lock_left_by_a_dead_writer_is_removed_whatever_its_age() {
let (_root, repo) = repo();
let stale = repo.path().join("objects/pack/multi-pack-index.lock");
std::fs::write(&stale, b"").unwrap();
let _held = lock_midx_writer_for(&repo, Duration::from_millis(50)).unwrap();
assert!(!stale.exists());
}

#[test]
fn live_writer_keeps_its_lock_and_excludes_a_second_writer() {
let (_root, repo) = repo();
let live = repo.path().join("objects/pack/multi-pack-index.lock");
let held = lock_midx_writer_for(&repo, Duration::from_millis(50)).unwrap();
std::fs::write(&live, b"").unwrap();
assert!(lock_midx_writer_for(&repo, Duration::from_millis(50)).is_err());
assert!(live.exists());
drop(held);
let _next = lock_midx_writer_for(&repo, Duration::from_millis(50)).unwrap();
assert!(!live.exists());
}
}
4 changes: 3 additions & 1 deletion crates/walgit-git/tests/midx.rs
Original file line number Diff line number Diff line change
Expand Up @@ -47,9 +47,11 @@ async fn verifies_real_multi_pack_bitmap_and_rejects_missing_or_corrupt_bytes()
.await
.unwrap();
assert!(verified.bitmap_path.is_file());
// `multi` lets pack-objects reuse deltas across packs; a clone then
// gets a thin pack whose bases are absent. Keep reuse to one pack.
assert_eq!(
common::run_git(repo.path(), &["config", "pack.allowPackReuse"]).trim(),
"multi"
"single"
);
common::run_git(repo.path(), &["fsck", "--strict"]);
let bytes = std::fs::read(&verified.bitmap_path).unwrap();
Expand Down
65 changes: 63 additions & 2 deletions crates/walgit-server/src/maintain.rs
Original file line number Diff line number Diff line change
@@ -1,9 +1,10 @@
//! Bounded maintenance from committed state: checkpoints, repair, compaction and audits.

use std::collections::HashMap;
use std::sync::Arc;
use std::time::{Instant, SystemTime};
use std::sync::{Arc, LazyLock};
use std::time::{Duration, Instant, SystemTime};

use parking_lot::Mutex;
use prost::Message;
use tracing::{Instrument, info, warn};
use walgit_git::RepoId;
Expand Down Expand Up @@ -125,6 +126,47 @@ pub fn host_name(state: &AppState) -> String {
.unwrap_or_else(|| walgit_store::coord::instance_id().to_string())
}

/// Consecutive failures per repository and the time its next unit may run.
///
/// A failing unit would otherwise be retried every pass; the exponential
/// backoff keeps a broken repository from rewriting packs in a tight loop.
static FAILURES: LazyLock<Mutex<HashMap<String, (u32, Instant)>>> =
LazyLock::new(|| Mutex::new(HashMap::new()));

fn backoff_delay(failures: u32) -> Duration {
let minutes = 1u64 << failures.min(6);
Duration::from_secs(minutes.saturating_mul(60).min(3_600))
}

fn failure_ready(id: &RepoId) -> bool {
let now = Instant::now();
FAILURES
.lock()
.get(&id.to_string())
.is_none_or(|(_, until)| *until <= now)
}

fn record_failure(id: &RepoId) {
let mut failures = FAILURES.lock();
let entry = failures
.entry(id.to_string())
.or_insert((0, Instant::now()));
entry.0 = entry.0.saturating_add(1);
entry.1 = Instant::now() + backoff_delay(entry.0);
if entry.0.is_power_of_two() {
warn!(
repo = %id,
failures = entry.0,
retry_in = ?backoff_delay(entry.0),
"maintenance: backing off after repeated failures"
);
}
}

fn clear_failure(id: &RepoId) {
FAILURES.lock().remove(&id.to_string());
}

pub async fn next_unit(state: &Arc<AppState>, id: &RepoId) -> anyhow::Result<Unit> {
if !state.cfg.placement.maintains(id.owner(), id.name()) {
return Ok(Unit::NotAssigned);
Expand Down Expand Up @@ -235,6 +277,10 @@ pub async fn run_pass(state: &Arc<AppState>) -> anyhow::Result<PassReport> {
break;
}
report.repos += 1;
if !failure_ready(&id) {
report.skipped += 1;
continue;
}
let unit = match next_unit(state, &id).await {
Ok(u) => u,
Err(e) => {
Expand Down Expand Up @@ -300,8 +346,10 @@ pub async fn run_pass(state: &Arc<AppState>) -> anyhow::Result<PassReport> {
if done {
report.units += 1;
report.last_unit = Some(format!("{id} {unit:?}"));
clear_failure(&id);
} else {
report.skipped += 1;
record_failure(&id);
}
}
Ok(report)
Expand Down Expand Up @@ -428,3 +476,16 @@ async fn run_op_value(
None => None,
}
}

#[cfg(test)]
mod tests {
use super::backoff_delay;
use std::time::Duration;

#[test]
fn backoff_grows_and_is_capped() {
assert_eq!(backoff_delay(1), Duration::from_mins(2));
assert_eq!(backoff_delay(2), Duration::from_mins(4));
assert_eq!(backoff_delay(10), Duration::from_hours(1));
}
}
Loading