use std::sync::Arc;
use std::sync::atomic::{AtomicBool, AtomicU64, Ordering};
use std::time::Duration;
use tokio::task::JoinHandle;
use tracing::{Instrument, info_span, warn};
use crate::backend::LockBackend;
use crate::telemetry;
pub(crate) struct LeaseState {
pub resource: String,
pub token: u64,
pub node_id: u64,
poisoned: AtomicBool,
released: AtomicBool,
pub renewals: AtomicU64,
ttl_nanos: AtomicU64,
}
impl LeaseState {
pub(crate) fn new(resource: String, token: u64, node_id: u64, ttl: Duration) -> Self {
Self {
resource,
token,
node_id,
poisoned: AtomicBool::new(false),
released: AtomicBool::new(false),
renewals: AtomicU64::new(0),
ttl_nanos: AtomicU64::new(ttl.as_nanos().min(u64::MAX as u128) as u64),
}
}
pub(crate) fn ttl(&self) -> Duration {
Duration::from_nanos(self.ttl_nanos.load(Ordering::Acquire))
}
pub(crate) fn set_ttl(&self, ttl: Duration) {
self.ttl_nanos.store(
ttl.as_nanos().min(u64::MAX as u128) as u64,
Ordering::Release,
);
}
pub(crate) fn is_poisoned(&self) -> bool {
self.poisoned.load(Ordering::Acquire)
}
pub(crate) fn poison(&self) {
self.poisoned.store(true, Ordering::Release);
}
pub(crate) fn mark_released(&self) {
self.released.store(true, Ordering::Release);
}
fn is_released(&self) -> bool {
self.released.load(Ordering::Acquire)
}
}
pub(crate) struct WatchdogHandle {
handle: JoinHandle<()>,
}
impl WatchdogHandle {
pub(crate) fn abort(&self) {
self.handle.abort();
}
}
pub(crate) fn spawn(
backend: Arc<dyn LockBackend>,
state: Arc<LeaseState>,
interval: Duration,
) -> WatchdogHandle {
let span = info_span!(
"ash_lock.watchdog",
resource_name = %state.resource,
fencing_token = state.token,
node_id = state.node_id,
);
let handle = tokio::spawn(
async move {
loop {
let effective = interval.min(state.ttl() / 2).max(Duration::from_millis(1));
tokio::time::sleep(effective).await;
if state.is_released() {
return;
}
match backend
.try_renew(&state.resource, state.token, state.ttl())
.await
{
Ok(_deadline) => {
state.renewals.fetch_add(1, Ordering::Relaxed);
}
Err(err) => {
telemetry::inc_renewal_failures();
warn!(
error = %err,
"lease renewal failed; poisoning guard (crash-fast fencing)"
);
state.poison();
return;
}
}
}
}
.instrument(span),
);
WatchdogHandle { handle }
}