9301161
use std::path::PathBuf;
use std::sync::Arc;
use cache::CacheClient;
use config::git::SyncSettings;
use export::AppDatabase;
use gitbaby::GitBaby;
use models::repo::repos;
use sea_orm::{DatabaseTransaction, TransactionTrait};
use tokio::sync::watch;
use uuid::Uuid;
use crate::context::facade::GitContext;
use crate::sync::branches;
use crate::sync::commits::{CommitData, collect_new_commits_for_sync, sync_commits};
use crate::sync::error::{SyncError, SyncResult};
use crate::sync::fsck;
use crate::sync::languages;
use crate::sync::license;
use crate::sync::lock::{self, SyncLock};
use crate::sync::stats;
use crate::sync::tags;
struct HeartbeatGuard {
handle: tokio::task::JoinHandle<()>,
}
impl Drop for HeartbeatGuard {
fn drop(&mut self) {
if !self.handle.is_finished() {
self.handle.abort();
}
}
}
#[derive(Clone)]
pub struct RepoSync {
db: Arc<AppDatabase>,
repo: repos::Model,
baby: GitBaby,
redis: export::AppRedis,
repo_root: PathBuf,
settings: SyncSettings,
}
impl RepoSync {
pub fn new(
db: Arc<AppDatabase>,
redis: export::AppRedis,
cache: Arc<CacheClient>,
repo: repos::Model,
repo_root: PathBuf,
settings: SyncSettings,
) -> Self {
let ctx = GitContext {
model: repo.clone(),
cache: cache.clone(),
database: db.clone(),
repo_root: repo_root.clone(),
objects_root: repo_root.clone(),
};
Self {
db,
repo,
baby: GitBaby::new(Arc::new(ctx)),
redis,
repo_root,
settings,
}
}
#[must_use]
pub fn repo_model(&self) -> &repos::Model {
&self.repo
}
#[must_use]
pub fn repo_id(&self) -> Uuid {
self.repo.id
}
#[must_use]
pub fn repo_dir(&self) -> PathBuf {
crate::context::facade::repo_dir_for(&self.repo_root, self.repo.id)
}
pub async fn sync(&self) -> SyncResult<()> {
let ttl = self.settings.lock_ttl_secs;
let lock = match lock::acquire_lock(&self.redis, self.repo.id, ttl).await? {
Some(l) => Arc::new(l),
None => return Err(SyncError::Locked),
};
let (tx, mut rx) = watch::channel(false);
let hb_lock = lock.clone();
let heartbeat = tokio::spawn(async move {
let interval = (ttl / 3).max(1);
loop {
tokio::time::sleep(std::time::Duration::from_secs(interval)).await;
match hb_lock.refresh(ttl).await {
Ok(true) => {}
Ok(false) | Err(_) => {
let _ = tx.send(true);
break;
}
}
}
});
let heartbeat = HeartbeatGuard { handle: heartbeat };
let result = tokio::select! {
r = self.sync_work() => r,
_ = rx.changed() => Err(SyncError::Locked),
};
let is_locked_failure = matches!(result.as_ref(), Err(SyncError::Locked));
if !is_locked_failure {
let _ = lock.release().await;
}
drop(heartbeat);
result
}
pub async fn sync_work(&self) -> SyncResult<()> {
let baby = self.baby.clone();
let new_commits: Vec<CommitData> = collect_new_commits_for_sync(
&baby,
&self.db,
self.repo.id,
self.settings.max_commits_per_sync,
)
.await?;
tracing::debug!(
repo_id = %self.repo.id,
new_commits = new_commits.len(),
"collected new commits for sync"
);
let txn: DatabaseTransaction = self.db.write().begin().await.map_err(SyncError::Db)?;
let (branch_count, _synced_commits, tag_count) =
self.sync_refs_and_data(&txn, &new_commits).await?;
tracing::debug!(
repo_id = %self.repo.id,
branches = branch_count,
tags = tag_count,
"refs synced"
);
languages::sync_repo_languages(&baby, &txn, &self.repo, &self.settings).await?;
tracing::debug!(repo_id = %self.repo.id, "languages synced");
stats::sync_repo_stats(
&txn,
&self.repo,
&self.repo_dir(),
new_commits.len(),
branch_count,
tag_count,
self.settings.git_command_timeout_secs,
)
.await?;
tracing::debug!(repo_id = %self.repo.id, "stats synced");
license::sync_repo_license(&txn, &self.repo_dir(), &self.repo).await?;
tracing::debug!(repo_id = %self.repo.id, "license synced");
search::outbox::enqueue_repo(&txn, self.repo.id)
.await
.map_err(SyncError::Db)?;
txn.commit().await.map_err(SyncError::Db)?;
tracing::debug!(repo_id = %self.repo.id, "sync transaction committed");
let gc_timeout = self.settings.git_command_timeout_secs.max(1800);
tracing::debug!(repo_id = %self.repo.id, "starting fsck/gc");
fsck::run_gc(&self.repo_dir(), gc_timeout).await;
fsck::run_fsck(&self.repo_dir(), gc_timeout).await;
Ok(())
}
async fn sync_refs_and_data(
&self,
txn: &DatabaseTransaction,
new_commits: &[CommitData],
) -> SyncResult<(usize, usize, usize)> {
let baby = self.baby.clone();
let branch_count = branches::sync_branches(&baby, txn, &self.repo).await?;
let synced = sync_commits(txn, self.repo.id, new_commits).await?;
let tag_count = tags::sync_tags(&baby, txn, &self.repo).await?;
Ok((branch_count, synced, tag_count))
}
pub async fn set_default_branch(&self, name: &str) -> SyncResult<()> {
self.baby
.set_default_branch(name)
.await
.map_err(SyncError::from)
}
pub async fn force_release_lock(lock: &SyncLock) -> SyncResult<()> {
lock.release().await
}
}