2ae9196
ZhenYi
feat(lib/git): add HTTP handler tree with routes and errors across analysis, archives, blobs, branches, commits, conflicts, content, diffs, hooks, lfs, logs, maintenance, packs, patches, pipelines, refs, remotes, repos, signatures, submodules, tags, trees
2ae9196
use std::path::{Path, PathBuf};
use std::sync::Arc;
use std::time::Duration;
use cache::CacheClient;
use config::git::SyncSettings;
use export::{AppDatabase, AppRedis};
use models::repo::repos;
use sea_orm::EntityTrait;
use tokio::task::JoinHandle;
use tokio_util::sync::CancellationToken;
use tracing::{debug, error, warn};
use crate::sync::error::{SyncError, SyncResult};
use crate::sync::event::PostReceiveTask;
use crate::sync::queue::SyncQueue;
use crate::sync::sync::RepoSync;
use crate::sync::webhook::WebhookTx;
const ATTEMPT_KEY_PREFIX: &str = "git:sync:attempts:";
const ATTEMPT_TTL_SECS: u64 = 3600;
const REQUEUE_BACKOFF_SECS: u64 = 5;
#[derive(Clone)]
pub struct SyncWorker {
db: Arc<AppDatabase>,
redis: AppRedis,
cache: Arc<CacheClient>,
repo_root: PathBuf,
queue: SyncQueue,
webhook: Arc<dyn WebhookTx>,
settings: SyncSettings,
cancel: CancellationToken,
}
pub struct RunningWorker {
handles: Vec<JoinHandle<()>>,
}
impl RunningWorker {
pub async fn shutdown(self) -> SyncResult<()> {
for h in self.handles {
if !h.is_finished() {
h.abort();
let _ = h.await;
}
}
Ok(())
}
}
impl SyncWorker {
#[allow(clippy::too_many_arguments)]
pub fn new(
db: Arc<AppDatabase>,
redis: AppRedis,
cache: Arc<CacheClient>,
repo_root: PathBuf,
queue: SyncQueue,
webhook: Arc<dyn WebhookTx>,
settings: SyncSettings,
cancel: CancellationToken,
) -> Self {
let concurrency = settings.concurrency.max(1);
debug!(
concurrency,
topic = %settings.queue_topic,
"starting sync worker"
);
Self {
db,
redis,
cache,
repo_root,
queue,
webhook,
settings,
cancel,
}
}
pub async fn run(self) -> SyncResult<RunningWorker> {
let concurrency = self.settings.concurrency.max(1);
let topic = self.settings.queue_topic.clone();
let brpop = self.settings.brpop_timeout_secs;
let mut handles = Vec::with_capacity(concurrency);
for i in 0..concurrency {
let db = self.db.clone();
let redis = self.redis.clone();
let cache = self.cache.clone();
let repo_root = self.repo_root.clone();
let queue = self.queue.clone();
let webhook = self.webhook.clone();
let settings = self.settings.clone();
let cancel = self.cancel.clone();
let topic = topic.clone();
handles.push(tokio::spawn(async move {
run_consumer(
i, &db, &redis, &cache, &repo_root, &queue, webhook, &settings, &cancel,
&topic, brpop,
)
.await;
}));
}
Ok(RunningWorker { handles })
}
}
#[allow(clippy::too_many_arguments)]
async fn run_consumer(
id: usize,
db: &Arc<AppDatabase>,
redis: &AppRedis,
cache: &Arc<CacheClient>,
repo_root: &Path,
queue: &SyncQueue,
webhook: Arc<dyn WebhookTx>,
settings: &SyncSettings,
cancel: &CancellationToken,
topic: &str,
brpop_timeout: u64,
) {
debug!(consumer = id, "consumer started");
loop {
if cancel.is_cancelled() {
break;
}
let payload = tokio::select! {
biased;
_ = cancel.cancelled() => break,
res = queue.consume(topic, brpop_timeout) => match res {
Ok(Some(p)) => p,
Ok(None) => continue,
Err(e) => {
warn!(consumer = id, error = %e, "queue consume error; backing off");
tokio::time::sleep(Duration::from_secs(REQUEUE_BACKOFF_SECS)).await;
continue;
}
}
};
let task: PostReceiveTask = match serde_json::from_slice(&payload) {
Ok(t) => t,
Err(e) => {
error!(consumer = id, error = %e, "invalid sync task payload; dropping");
continue;
}
};
let repo = match repos::Entity::find_by_id(task.repo_id).one(db.read()).await {
Ok(Some(r)) => r,
Ok(None) => {
warn!(consumer = id, repo_id = %task.repo_id, "repo not found; skipping");
continue;
}
Err(e) => {
error!(consumer = id, repo_id = %task.repo_id, error = %e, "repo lookup failed; skipping");
continue;
}
};
let rsync = RepoSync::new(
db.clone(),
redis.clone(),
cache.clone(),
repo,
repo_root.to_path_buf(),
settings.clone(),
);
match rsync.sync().await {
Ok(()) => {
if let Err(e) = webhook.dispatch(&task).await {
warn!(consumer = id, repo_id = %task.repo_id, error = %e, "webhook dispatch failed");
}
}
Err(e) if e.is_locked() => {
warn!(consumer = id, repo_id = %task.repo_id, "sync locked; requeueing");
if let Err(e2) = requeue_or_dlq(queue, settings, &task, &payload).await {
error!(consumer = id, repo_id = %task.repo_id, error = %e2, "requeue failed");
}
}
Err(e) => {
error!(consumer = id, repo_id = %task.repo_id, error = %e, "sync task failed");
if let Err(e2) = requeue_or_dlq(queue, settings, &task, &payload).await {
error!(consumer = id, repo_id = %task.repo_id, error = %e2, "requeue failed");
}
}
}
}
debug!(consumer = id, "consumer stopped");
}
async fn requeue_or_dlq(
queue: &SyncQueue,
settings: &SyncSettings,
task: &PostReceiveTask,
payload: &[u8],
) -> SyncResult<()> {
let key = format!("{ATTEMPT_KEY_PREFIX}{}", task.id);
let attempts = queue.incr_with_ttl(&key, ATTEMPT_TTL_SECS).await?;
if attempts as u64 > settings.max_lock_retries {
let _ = queue.delete(&key).await;
queue
.publish_raw(&settings.dlq_topic, Some(task.repo_id.as_bytes()), payload)
.await?;
warn!(
repo_id = %task.repo_id,
attempts,
"task exceeded retry limit; sent to DLQ"
);
return Ok(());
}
tokio::time::sleep(Duration::from_secs(REQUEUE_BACKOFF_SECS)).await;
queue
.publish_raw(
&settings.queue_topic,
Some(task.repo_id.as_bytes()),
payload,
)
.await?;
debug!(
repo_id = %task.repo_id,
attempts,
"re-queued retry task"
);
Ok(())
}
pub async fn run_once(
worker: &SyncWorker,
task: PostReceiveTask,
payload: Vec<u8>,
) -> SyncResult<()> {
let repo = repos::Entity::find_by_id(task.repo_id)
.one(worker.db.read())
.await
.map_err(SyncError::Db)?
.ok_or_else(|| SyncError::NotFound(format!("repo {}", task.repo_id)))?;
let rsync = RepoSync::new(
worker.db.clone(),
worker.redis.clone(),
worker.cache.clone(),
repo,
worker.repo_root.clone(),
worker.settings.clone(),
);
let result = rsync.sync().await;
if result.is_ok() {
let _ = worker.webhook.dispatch(&task).await;
} else if let Err(e) = requeue_or_dlq(&worker.queue, &worker.settings, &task, &payload).await {
warn!(error = %e, "requeue failed in run_once");
}
result
}