ash_lock/
builder.rs

1//! The fluent [`LockBuilder`] returned by [`LockClient::lock`](crate::LockClient::lock).
2
3use std::sync::Arc;
4use std::time::Duration;
5
6use tokio::time::{Instant as TokioInstant, sleep};
7use tracing::{Instrument, info_span};
8
9use crate::backend::LockBackend;
10use crate::error::{Error, Result};
11use crate::guard::LockGuard;
12use crate::watchdog::{self, LeaseState};
13use crate::{ClientInner, telemetry};
14
15/// Fluent configuration for a single acquire attempt.
16///
17/// Created by [`LockClient::lock`](crate::LockClient::lock). Configure it, then
18/// call [`acquire`](LockBuilder::acquire).
19pub struct LockBuilder<'a> {
20    pub(crate) inner: &'a ClientInner,
21    pub(crate) resource: String,
22    pub(crate) ttl: Duration,
23    pub(crate) fair: bool,
24    /// How long to keep retrying a contended lock before giving up. Derived
25    /// from the TTL; kept here so it is overridable in future.
26    pub(crate) wait_budget: Duration,
27}
28
29impl<'a> LockBuilder<'a> {
30    /// Override the lease TTL for this lock.
31    pub fn with_ttl(mut self, ttl: Duration) -> Self {
32        self.ttl = ttl;
33        self
34    }
35
36    /// Enable FIFO-fair queueing.
37    ///
38    /// MOCK: the in-memory / consensus(mock) backends do not yet implement a
39    /// server-side wait queue, so `fair` currently only influences the client
40    /// retry pacing (fixed interval instead of jittered backoff) to approximate
41    /// arrival-order fairness. Recorded on the builder for the real backend.
42    pub fn with_fairness(mut self, fair: bool) -> Self {
43        self.fair = fair;
44        self
45    }
46
47    /// Attempt to acquire the lock, retrying on contention until the wait
48    /// budget is exhausted.
49    pub async fn acquire(self) -> Result<LockGuard> {
50        let backend = Arc::clone(&self.inner.backend);
51        let node_id = self.inner.node_id;
52        let span = info_span!(
53            "ash_lock.acquire",
54            resource_name = %self.resource,
55            node_id = node_id,
56            fair = self.fair,
57        );
58        self.acquire_inner(backend, node_id).instrument(span).await
59    }
60
61    async fn acquire_inner(self, backend: Arc<dyn LockBackend>, node_id: u64) -> Result<LockGuard> {
62        if self.ttl.is_zero() {
63            return Err(Error::Config("lock ttl must be non-zero".into()));
64        }
65        let deadline = TokioInstant::now() + self.wait_budget;
66        let mut attempt: u32 = 0;
67
68        loop {
69            match backend.try_acquire(&self.resource, self.ttl).await {
70                Ok((token, _lease_deadline)) => {
71                    telemetry::inc_acquisitions();
72                    let stamp = self.inner.clock.stamp()?;
73
74                    let state = Arc::new(LeaseState::new(
75                        self.resource.clone(),
76                        token,
77                        node_id,
78                        self.ttl,
79                    ));
80                    let handle = watchdog::spawn(
81                        Arc::clone(&backend),
82                        Arc::clone(&state),
83                        self.inner.watchdog_interval,
84                    );
85                    return Ok(LockGuard::new(backend, state, handle, stamp));
86                }
87                Err(Error::Contended { .. }) => {
88                    // First contention on this call counts as a contention event.
89                    if attempt == 0 {
90                        telemetry::inc_contention();
91                    }
92                    attempt += 1;
93
94                    if TokioInstant::now() >= deadline {
95                        return Err(Error::Timeout {
96                            resource: self.resource.clone(),
97                            waited: self.wait_budget,
98                        });
99                    }
100                    sleep(self.retry_delay(attempt)).await;
101                }
102                // Any non-contention error is terminal.
103                Err(e) => return Err(e),
104            }
105        }
106    }
107
108    /// Retry pacing. Fair mode uses a steady interval to preserve arrival order;
109    /// unfair mode uses capped exponential backoff with simple deterministic
110    /// jitter (derived from node_id + attempt, so tests stay reproducible).
111    fn retry_delay(&self, attempt: u32) -> Duration {
112        let base = self.inner.watchdog_interval.min(Duration::from_millis(50));
113        if self.fair {
114            return base;
115        }
116        let backoff = base.saturating_mul(1u32 << attempt.min(5));
117        // Deterministic pseudo-jitter in [0, base): avoids thundering herd
118        // without pulling in an RNG dependency for the mock.
119        let jitter_ns = self
120            .inner
121            .node_id
122            .wrapping_mul(2654435761)
123            .wrapping_add(attempt as u64)
124            % (base.as_nanos() as u64).max(1);
125        backoff.saturating_add(Duration::from_nanos(jitter_ns))
126    }
127}
ash_lock/builder.rs

ash_lock/
builder.rs
