194c44c
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::{Instrument as _, debug, error, info, 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);
info!(
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 queue = self.queue;
if queue.consumer_conns().is_empty() {
queue = SyncQueue::with_consumer_conns(queue.redis().clone(), concurrency).await?;
}
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 = 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, id) => 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,
payload_len = payload.len(),
"invalid sync task payload; dropping"
);
continue;
}
};
let started = std::time::Instant::now();
debug!(
consumer = id,
task_id = %task.id,
repo_id = %task.repo_id,
refs = task.refs.len(),
"sync task received"
);
let span = tracing::info_span!(
"git.sync",
task_id = %task.id,
repo_id = %task.repo_id,
refs = task.refs.len(),
request_id = tracing::field::Empty
);
if let Some(request_id) = task.request_id.as_deref() {
span.record("request_id", request_id);
}
export::telemetry::set_parent_from_traceparent(&span, task.trace.as_deref());
let repo = match repos::Entity::find_by_id(task.repo_id).one(db.read()).await {
Ok(Some(r)) => r,
Ok(None) => {
warn!(
consumer = id,
task_id = %task.id,
repo_id = %task.repo_id,
"repo not found; skipping"
);
continue;
}
Err(e) => {
error!(
consumer = id,
task_id = %task.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 export::middleware::with_request_id(
task.request_id.as_deref(),
rsync.sync().instrument(span.clone()),
)
.await
{
Ok(()) => {
export::metrics::sync_job("ok");
info!(
consumer = id,
task_id = %task.id,
repo_id = %task.repo_id,
elapsed_ms = started.elapsed().as_millis() as u64,
"sync task completed"
);
if let Err(e) =
crate::sync::pulls::refresh_pulls_after_sync(db, rsync.repo_model(), &task)
.await
{
warn!(
consumer = id,
task_id = %task.id,
repo_id = %task.repo_id,
error = %e,
"pull request refresh failed"
);
}
if let Err(e) = export::middleware::with_request_id(
task.request_id.as_deref(),
webhook.dispatch(&task).instrument(span.clone()),
)
.await
{
warn!(
consumer = id,
task_id = %task.id,
repo_id = %task.repo_id,
error = %e,
"webhook dispatch failed"
);
}
}
Err(e) if e.is_locked() => {
export::metrics::sync_job("locked");
warn!(
consumer = id,
task_id = %task.id,
repo_id = %task.repo_id,
"sync locked; requeueing"
);
if let Err(e2) = requeue_or_dlq(queue, settings, &task, &payload).await {
error!(
consumer = id,
task_id = %task.id,
repo_id = %task.repo_id,
error = %e2,
"requeue failed"
);
}
}
Err(e) => {
export::metrics::sync_job("error");
error!(
consumer = id,
task_id = %task.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,
task_id = %task.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!(
task_id = %task.id,
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!(
task_id = %task.id,
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
}