corium_transactor/
lease.rs

1//! Write-lease records with CAS fencing (see `docs/design/log-and-transactor.md`).
2//!
3//! The lease is a record in the root store: `{owner, version, expiry}`,
4//! renewed by CAS. The fencing rule pairs it with the database root: every
5//! db-root publication carries the lease version the writer believes it
6//! holds, and acquiring a lease with a new version immediately re-publishes
7//! the db root under that version (the *fence bump*). A deposed transactor
8//! that wakes later always fails its root CAS — its expected bytes are
9//! stale — and observes the newer lease version instead of publishing.
10
11use corium_store::{RootStore, StoreError};
12use thiserror::Error;
13
14/// Root-store key for a database's lease record.
15#[must_use]
16pub fn lease_root(db: &str) -> String {
17    format!("lease:{db}")
18}
19
20/// A write-lease record.
21#[derive(Clone, Debug, Eq, PartialEq)]
22pub struct Lease {
23    /// Owning transactor id.
24    pub owner: String,
25    /// Fencing version; increments on every change of ownership.
26    pub version: u64,
27    /// Expiry as Unix milliseconds.
28    pub expires_unix_ms: i64,
29}
30
31impl Lease {
32    /// Encodes the record for the root store.
33    #[must_use]
34    pub fn encode(&self) -> Vec<u8> {
35        format!(
36            "{}\n{}\n{}\n",
37            self.owner, self.version, self.expires_unix_ms
38        )
39        .into_bytes()
40    }
41
42    /// Decodes a stored record.
43    #[must_use]
44    pub fn decode(bytes: &[u8]) -> Option<Self> {
45        let text = std::str::from_utf8(bytes).ok()?;
46        let mut lines = text.lines();
47        Some(Self {
48            owner: lines.next()?.to_owned(),
49            version: lines.next()?.parse().ok()?,
50            expires_unix_ms: lines.next()?.parse().ok()?,
51        })
52    }
53}
54
55/// Lease acquisition/renewal failure.
56#[derive(Debug, Error)]
57pub enum LeaseError {
58    /// Another owner holds an unexpired lease.
59    #[error("lease held by {owner} until {expires_unix_ms}")]
60    Held {
61        /// Current owner id.
62        owner: String,
63        /// Expiry of the conflicting lease.
64        expires_unix_ms: i64,
65    },
66    /// This owner no longer holds the lease.
67    #[error("lease lost to another owner")]
68    Lost,
69    /// Root store failure.
70    #[error(transparent)]
71    Store(#[from] StoreError),
72}
73
74/// Acquires (or re-acquires) the write lease for `db`.
75///
76/// Returns the held lease. The version increments only when ownership
77/// changes hands; a same-owner acquisition keeps its version.
78///
79/// # Errors
80/// Returns [`LeaseError::Held`] while another owner's lease is unexpired.
81pub fn acquire(
82    store: &dyn RootStore,
83    db: &str,
84    owner: &str,
85    ttl_ms: i64,
86    now_unix_ms: i64,
87) -> Result<Lease, LeaseError> {
88    let name = lease_root(db);
89    loop {
90        let current = store.get_root(&name)?;
91        let decoded = current.as_deref().and_then(Lease::decode);
92        let version = match &decoded {
93            Some(lease) if lease.owner == owner => lease.version,
94            Some(lease) if lease.expires_unix_ms > now_unix_ms => {
95                return Err(LeaseError::Held {
96                    owner: lease.owner.clone(),
97                    expires_unix_ms: lease.expires_unix_ms,
98                });
99            }
100            Some(lease) => lease.version + 1,
101            None => 1,
102        };
103        let next = Lease {
104            owner: owner.to_owned(),
105            version,
106            expires_unix_ms: now_unix_ms + ttl_ms,
107        };
108        match store.cas_root(&name, current.as_deref(), &next.encode()) {
109            Ok(()) => return Ok(next),
110            Err(StoreError::CasFailed { .. }) => {}
111            Err(error) => return Err(error.into()),
112        }
113    }
114}
115
116/// Renews a held lease; fails with [`LeaseError::Lost`] when another owner
117/// has taken it (or the stored record no longer matches).
118///
119/// # Errors
120/// Returns [`LeaseError::Lost`] on any ownership change.
121pub fn renew(
122    store: &dyn RootStore,
123    db: &str,
124    held: &Lease,
125    ttl_ms: i64,
126    now_unix_ms: i64,
127) -> Result<Lease, LeaseError> {
128    let name = lease_root(db);
129    let current = store.get_root(&name)?;
130    if current.as_deref() != Some(held.encode().as_slice()) {
131        return Err(LeaseError::Lost);
132    }
133    let next = Lease {
134        owner: held.owner.clone(),
135        version: held.version,
136        expires_unix_ms: now_unix_ms + ttl_ms,
137    };
138    match store.cas_root(&name, current.as_deref(), &next.encode()) {
139        Ok(()) => Ok(next),
140        Err(StoreError::CasFailed { .. }) => Err(LeaseError::Lost),
141        Err(error) => Err(error.into()),
142    }
143}
144
145/// Releases a held lease by expiring it immediately. Ownership changes
146/// after release still increment the version (the record remains).
147///
148/// # Errors
149/// Returns [`LeaseError::Lost`] when the lease is no longer held.
150pub fn release(store: &dyn RootStore, db: &str, held: &Lease) -> Result<(), LeaseError> {
151    let name = lease_root(db);
152    let current = store.get_root(&name)?;
153    if current.as_deref() != Some(held.encode().as_slice()) {
154        return Err(LeaseError::Lost);
155    }
156    let expired = Lease {
157        expires_unix_ms: 0,
158        ..held.clone()
159    };
160    match store.cas_root(&name, current.as_deref(), &expired.encode()) {
161        Ok(()) => Ok(()),
162        Err(StoreError::CasFailed { .. }) => Err(LeaseError::Lost),
163        Err(error) => Err(error.into()),
164    }
165}
corium_transactor/lease.rs

corium_transactor/
lease.rs
