194c44c
use std::process;
use export::{AppRedis, RedisConnection};
use uuid::Uuid;
use crate::sync::error::SyncResult;
const LOCK_KEY_PREFIX: &str = "git:sync:lock:";
const REFRESH_SCRIPT: &str = "local v = redis.call('GET', KEYS[1])\nif v == ARGV[1] then\n redis.call('PEXPIRE', KEYS[1], ARGV[2])\n return 1\nend\nreturn 0";
const RELEASE_SCRIPT: &str = "local v = redis.call('GET', KEYS[1])\nif v == ARGV[1] then\n return redis.call('DEL', KEYS[1])\nend\nreturn 0";
macro_rules! dispatch {
($conn:expr, |$c:ident| $body:expr) => {
match &mut $conn {
RedisConnection::Standalone($c) => $body,
RedisConnection::Cluster($c) => $body,
}
};
}
pub struct SyncLock {
redis: AppRedis,
key: String,
value: String,
}
impl SyncLock {
#[must_use]
pub fn value(&self) -> &str {
&self.value
}
#[must_use]
pub fn key(&self) -> &str {
&self.key
}
pub async fn refresh(&self, ttl_secs: u64) -> SyncResult<bool> {
let mut conn = self.redis.connection_clone();
let n: i64 = dispatch!(conn, |c| {
redis::cmd("EVAL")
.arg(REFRESH_SCRIPT)
.arg(1)
.arg(&self.key)
.arg(&self.value)
.arg(ttl_secs.saturating_mul(1000))
.query_async(c)
.await
})?;
if n != 1 {
tracing::warn!(key = %self.key, "sync lock heartbeat refresh lost ownership");
}
Ok(n == 1)
}
pub async fn release(&self) -> SyncResult<()> {
let mut conn = self.redis.connection_clone();
let _: i64 = dispatch!(conn, |c| {
redis::cmd("EVAL")
.arg(RELEASE_SCRIPT)
.arg(1)
.arg(&self.key)
.arg(&self.value)
.query_async(c)
.await
})?;
tracing::debug!(key = %self.key, "sync lock released");
Ok(())
}
}
pub async fn acquire_lock(
redis: &AppRedis,
repo_id: Uuid,
ttl_secs: u64,
) -> SyncResult<Option<SyncLock>> {
let key = format!("{LOCK_KEY_PREFIX}{repo_id}");
let value = format!("{}:{}", Uuid::now_v7(), process::id());
let mut conn = redis.connection_clone();
let acquired: bool = dispatch!(conn, |c| {
redis::cmd("SET")
.arg(&key)
.arg(&value)
.arg("NX")
.arg("EX")
.arg(ttl_secs)
.query_async(c)
.await
})?;
if !acquired {
tracing::debug!(key = %key, repo_id = %repo_id, "sync lock busy; another sync in progress");
return Ok(None);
}
tracing::debug!(key = %key, repo_id = %repo_id, ttl_secs, "sync lock acquired");
Ok(Some(SyncLock {
redis: redis.clone(),
key,
value,
}))
}