pub
/
waymaker-client
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

Compare commits

base: pub:main
Branches Tags
pub:main
pub:v0.1.33
pub:v0.1.32
pub:v0.1.31
pub:v0.1.29
pub:v0.1.28
pub:v0.1.27
..
compare: pub:v0.1.32
Branches Tags
pub:main
pub:v0.1.33
pub:v0.1.32
pub:v0.1.31
pub:v0.1.29
pub:v0.1.28
pub:v0.1.27

No commits in common. "main" and "v0.1.32" have entirely different histories.
main ... v0.1.32

4 changed files with 214 additions and 123 deletions
Show Stats Expand all files Collapse all files

2
VERSION
View File

@ -1 +1 @@
0.1.33
0.1.32

2
rust/Cargo.toml
View File

@ -1,6 +1,6 @@
[package]
name = "waymaker-client"
version = "0.1.33"
version = "0.1.32"
edition = "2021"
description = "Official Rust client for waymaker — locks, streams, KV, collections, sketches, cache, object store"
repository = "https://git.awesomike.com/pub/waymaker-client"

287
rust/src/lock/mod.rs
View File

@ -7,15 +7,15 @@
//! - `client.lease_status(&lock_id)`
//! - `client.multi_lock(...)`
//!
//! The returned [`Lock`] keeps a background task that drains the
//! server event stream and, once that stream ends (the server closes
//! it at grant; the lease lives on its TTL plus renewal), re-proves
//! ownership through `lease_status` on a cadence of a third of the
//! lease. It never re-acquires on its own: a lease it can no longer
//! prove is published as `lost`, and re-acquiring is the caller's
//! decision. The lock's live state (fence token, lease expiry, and a
//! `lost` flag) is published through [`Lock::watch`]; re-read
//! [`Lock::fence_token`] before every fenced side effect.
//! The returned [`Lock`] keeps a background task that holds the
//! server event stream open and, if that stream drops (e.g. the
//! key's primary bounces), transparently re-binds it and re-confirms
//! ownership — reusing the original `request_id` so a still-held
//! lease is recovered rather than re-contended. The lock's live
//! state (fence token, lease expiry, and a `lost` flag) is published
//! through [`Lock::watch`]; re-read [`Lock::fence_token`] before every
//! fenced side effect, because a lost-then-re-won lock carries a new,
//! higher token.
//!
//! ## Leader-election pattern
//!
@ -57,9 +57,9 @@ use uuid::Uuid;
/// already applies to every non-`Ephemeral` lock regardless of
/// scope. A stronger scope does not make a lock survive; a
/// replicated lease does.
/// - **Client-side ownership checks.** [`Lock`] re-proves ownership via
/// `lease_status` after its event stream ends, but that is a
/// client-lib behaviour — no scope value changes it.
/// - **Client-side transparency.** [`Lock`] transparently re-binds
/// its event stream after a disconnect, but that is a client-lib
/// behaviour — no scope value changes it.
/// - **Mutual exclusion.** Holding the lock is not, by itself, a
/// guarantee that no one else acts. You MUST validate
/// [`Lock::fence_token`] at your side effect (the DB write / object
@ -202,7 +202,7 @@ pub struct LockState {
/// a lost-then-re-won lock carries a higher token, and acting on a
/// stale copy defeats fencing.
pub fence_token: u64,
/// Lease expiry (epoch ms), refreshed from renewal and lease_status.
/// Lease expiry (epoch ms), refreshed from heartbeats / re-acquire.
pub lease_expires_at_ms: i64,
/// Monotonic local ownership-generation number. A same-generation event
/// stream re-bind preserves this value; a transparent fresh acquire with
@ -243,12 +243,14 @@ pub enum RenewalFailure {
/// on drop would silently swallow errors and surprise callers
/// who expect the server-side state to outlive the local handle.
///
/// A background task drains the server event stream, then re-proves
/// ownership via `lease_status` once it ends — which the server does
/// at grant. The lease itself is kept alive by the server's TTL plus
/// [`Lock::spawn_renewal`], independent of the stream: a closed stream
/// loses event visibility, not the lock. Live state is exposed through
/// the accessors below and [`Lock::watch`].
/// A background task keeps the server event stream open. If that
/// stream drops — typically because the key's primary bounced — the
/// task transparently re-binds it, reusing the original `request_id`
/// so a still-held lease is recovered rather than re-contended. The
/// lease itself is kept alive by the server's TTL plus
/// [`Lock::spawn_renewal`], independent of the stream: dropping the
/// stream loses event visibility, not the lock. Live state is exposed
/// through the accessors below and [`Lock::watch`].
pub struct Lock {
client: Client,
pub key: String,
@ -575,19 +577,27 @@ impl Lock {
/// Release the lock. After this returns, the server-side state
/// is gone and subsequent operations on this `Lock` will fail.
pub async fn unlock(mut self) -> Result<()> {
// Stop the hold task FIRST and wait for it to wind down, so nothing
// it publishes (or marks lost) races the UnLock. The task never
// acquires; at most it has one lease_status in flight, bounded by
// `HOLD_RPC_TIMEOUT`.
// Stop the hold task FIRST so it cannot re-acquire the lock we
// are about to release — a re-acquire racing the UnLock would
// resurrect a lock the caller believes is gone.
self.stopped.store(true, Ordering::SeqCst);
self.stop.notify_one();
// ...and WAIT for it to wind down before sending UnLock. Signalling
// alone is not enough: if the task has an idempotent re-acquire RPC
// in flight, cancelling that RPC client-side does not cancel it
// server-side — the server can still grant a fresh lease to a stream
// nobody reads, a ghost writer that blocks every later reader on the
// key until its lease is reaped (and a lease whose stream vanished
// mid-grant has been observed to never be reaped). So the task lets
// any in-flight acquire finish (bounded by `HOLD_RPC_TIMEOUT`),
// releases anything it won while stopping, and only then exits.
let wind_down = HOLD_RPC_TIMEOUT + Duration::from_secs(2);
if tokio::time::timeout(wind_down, &mut self._hold).await.is_err() {
// Backstop only — the task is wedged past every bound it has.
self._hold.abort();
}
// Read the id AFTER the task exited, so the release names the
// last lease the task published.
// Read the id AFTER the task exited: a transparent re-bind that
// completed while we were stopping may have changed it.
let id = self.id();
let mut c = self.client.locks_client();
let r = c
@ -722,9 +732,11 @@ impl Lock {
impl Drop for Lock {
fn drop(&mut self) {
// Stop the background hold task so a dropped (but not explicitly
// unlocked) handle does not keep polling in the background. The
// task honours `stopped` at every step, so it exits on its own
// within one bounded RPC.
// unlocked) handle does not keep re-acquiring in the background.
// Deliberately NOT `abort()`: aborting mid-acquire strands a lease
// the server may still grant (see `unlock`). The task honours
// `stopped` at every step and releases anything it won after the
// flag was set, so it exits on its own within one bounded RPC.
// (`unlock` already joined it; this only matters for a drop
// without unlock, where the lease is left to TTL by contract.)
self.stopped.store(true, Ordering::SeqCst);
@ -732,13 +744,13 @@ impl Drop for Lock {
}
}
// ---- Background hold / loss-detection loop -----------------------------
// ---- Background hold / re-bind / loss-detection loop -----------------
//
// The lease lives on the server's TTL + renewal, NOT on this event
// stream (granting a lock sets its `sender_stream` to None, which ends
// the stream). So an ended stream means "lost event visibility", not
// "lost lock". This task then polls lease_status, and only declares the
// lock `lost` when it can no longer prove ownership.
// stream (granting a lock sets its `sender_stream` to None). So a
// dropped stream means "lost event visibility", not "lost lock". This
// task re-binds the stream when it can, and only declares the lock
// `lost` when it can no longer prove ownership via lease_status.
/// Outcome of draining the current event stream.
enum Drained {
@ -749,6 +761,17 @@ enum Drained {
Disconnected,
}
/// Outcome of one idempotent re-acquire attempt.
enum Rebound {
/// Got a live stream back (transparent re-bind, or a clean re-win
/// after a real loss). Carries the fresh `(stream, id, fence, exp)`.
Bound(Box<tonic::Streaming<LockEvent>>, ValidatedGrant),
/// The acquire did not grant (key held — possibly by our own
/// still-adopted lease on a new primary, possibly by someone else)
/// or the RPC failed. Disambiguate via lease_status.
NotBound,
}
/// Whether we still demonstrably own the lease.
enum Ownership {
Held(ValidatedGrant),
@ -758,41 +781,65 @@ enum Ownership {
const HOLD_BASE_BACKOFF: Duration = Duration::from_millis(200);
const HOLD_MAX_BACKOFF: Duration = Duration::from_secs(10);
const HOLD_MAX_POLL: Duration = Duration::from_secs(5);
const HOLD_RPC_TIMEOUT: Duration = Duration::from_secs(10);
/// Background task body: drain the event stream, then poll ownership,
/// and surface lease updates / loss through `state_tx`.
/// Background task body: hold the event stream, re-bind it on drop, and
/// surface fence-token changes / loss through `state_tx`.
#[allow(clippy::too_many_arguments)]
async fn hold_loop(
client: Client,
key: String,
lease_ms: u32,
read: bool,
reacquire_req: LockRequest,
mut stream: tonic::Streaming<LockEvent>,
state_tx: watch::Sender<LockState>,
scope: Scope,
stop: Arc<Notify>,
stopped: Arc<AtomicBool>,
) {
if let Drained::Stopped = drain_stream(&mut stream, &state_tx, &key, scope, &stop).await {
return;
loop {
match drain_stream(&mut stream, &state_tx, &key, scope, &stop).await {
Drained::Stopped => return,
Drained::Disconnected => {}
}
// The server ends the stream as soon as it grants (the entry's
// `sender_stream` is taken at grant), so reaching here is the normal
// state of a held lock, not a loss. Watch ownership through
// lease_status, and never re-acquire a lease we may still hold:
// re-binding first spun at hundreds of `Lock` RPCs per second per held
// lock (each re-bind got a stream that ended at once). The flood starved
// renewals, leases lapsed, and the spinning re-acquire re-won them under
// fresh ids — the holder's generation changed under it and a second
// client could be granted the key meanwhile (leptin prod, 2026-10-04).
let poll = hold_poll_interval(lease_ms);
// Re-establish: re-bind the stream; if we cannot, decide whether
// we still own the lease or have truly lost it.
let mut backoff = HOLD_BASE_BACKOFF;
loop {
if stopped.load(Ordering::SeqCst) {
return;
}
// The acquire RPC is deliberately NOT raced against `stop`.
// Dropping an in-flight `Lock`/`ReadLock` request only resets the
// client side of the stream; the server keeps processing it and,
// once `unlock` has removed the original lease, the idempotent
// re-acquire re-WINS the key with a fresh id that nobody holds —
// a ghost writer that parks every later reader. Let the RPC run
// to its own bound (`HOLD_RPC_TIMEOUT`) and then honour `stop`,
// releasing whatever it won in the meantime.
let rebound = reacquire(&client, read, &reacquire_req, &key, scope).await;
if stopped.load(Ordering::SeqCst) {
if let Rebound::Bound(_stream, grant) = rebound {
let known = state_tx.borrow().id.clone();
if grant.id != known {
// A fresh lease won while stopping: the holder never
// learns this id, so it must be released here.
release_orphan(&client, &key, &grant.id).await;
}
// Same id = a transparent re-bind of the lease the holder
// is about to `UnLock` itself; nothing to do.
}
return;
}
match rebound {
Rebound::Bound(s, grant) => {
publish_grant(&state_tx, grant);
stream = *s;
break; // resume draining the fresh stream
}
Rebound::NotBound => {
let current = state_tx.borrow().clone();
let owned = match race_stop(
confirm_ownership(&client, &key, &current.id, current.fence_token, scope),
@ -803,37 +850,15 @@ async fn hold_loop(
Some(o) => o,
None => return,
};
let wait = match owned {
match owned {
Ownership::Held(grant) => {
// Still ours: renewal keeps the lease alive.
// Still ours (renewal keeps the lease alive);
// we just could not get an event stream — keep
// monitoring with backoff.
publish_grant(&state_tx, grant);
backoff = HOLD_BASE_BACKOFF;
poll
}
// The server reports a missing lease as `Unavailable`, the same
// as a transport failure, so "unproven" is all we can know. Once
// the last proven expiry has passed, unproven means gone.
Ownership::Unknown if current.lease_expires_at_ms <= unix_now_ms() => {
mark_lost(&state_tx);
return;
}
Ownership::Unknown => {
let wait = backoff;
backoff = (backoff * 2).min(HOLD_MAX_BACKOFF);
wait
}
Ownership::Unknown => { /* transient — back off, retry */ }
Ownership::Lost => {
mark_lost(&state_tx);
return;
}
};
if race_stop(tokio::time::sleep(wait), &stop).await.is_none() {
return;
}
}
}
fn mark_lost(state_tx: &watch::Sender<LockState>) {
state_tx.send_if_modified(|current| {
if current.lost {
false
@ -842,14 +867,20 @@ fn mark_lost(state_tx: &watch::Sender<LockState>) {
true
}
});
return;
}
}
if race_stop(tokio::time::sleep(backoff), &stop)
.await
.is_none()
{
return;
}
backoff = (backoff * 2).min(HOLD_MAX_BACKOFF);
}
}
}
}
/// How often a held lock re-proves ownership once its event stream has
/// ended: a third of the lease, so a loss is seen well inside one lease,
/// bounded so short leases do not hammer the server and long ones still
/// notice within seconds.
fn hold_poll_interval(lease_ms: u32) -> Duration {
Duration::from_millis(u64::from(lease_ms) / 3).clamp(HOLD_BASE_BACKOFF, HOLD_MAX_POLL)
}
/// Drain events from the current stream until it ends, the lease is
@ -915,6 +946,57 @@ async fn drain_stream(
}
}
/// One idempotent re-acquire attempt (reusing the original
/// `request_id`, `max_wait = 0`). On the original primary with the
/// lease still held this re-binds the stream and re-emits the held
/// entry; if the lease was lost and the key is free it re-wins with a
/// fresh, higher token.
async fn reacquire(
client: &Client,
read: bool,
req: &LockRequest,
expected_key: &str,
scope: Scope,
) -> Rebound {
let req = req.clone();
let mut c = client.locks_client();
let attempt = async move {
let mut stream = if read {
c.read_lock(Request::new(req)).await?.into_inner()
} else {
c.lock(Request::new(req)).await?.into_inner()
};
loop {
match stream.message().await? {
Some(ev) if ev.event_type == LockEventType::Acquired as i32 => {
let grant = match validate_grant_event(&ev, expected_key, scope, unix_now_ms())
{
Ok(grant) => grant,
Err(_) => return Ok(None),
};
return Ok::<
Option<(tonic::Streaming<LockEvent>, ValidatedGrant)>,
tonic::Status,
>(Some((stream, grant)));
}
Some(ev)
if ev.event_type == LockEventType::Failed as i32
|| ev.event_type == LockEventType::Expired as i32 =>
{
return Ok(None); // not granted (contended)
}
Some(_) => continue, // Waiting / Heartbeat
None => return Ok(None), // closed before grant
}
}
};
match tokio::time::timeout(HOLD_RPC_TIMEOUT, attempt).await {
Ok(Ok(Some((stream, grant)))) => Rebound::Bound(Box::new(stream), grant),
Ok(Ok(None)) => Rebound::NotBound,
Ok(Err(_)) | Err(_) => Rebound::NotBound,
}
}
/// Query whether `id` still holds `key`. `success = false` (or a
/// missing lease) means we lost it; a transport / timeout error is
/// inconclusive (`Unknown`) so we never falsely declare loss.
@ -956,6 +1038,18 @@ async fn confirm_ownership(
}
}
/// Best-effort release of a lease the hold task won after it was told to
/// stop (see [`hold_loop`]). Errors are swallowed: the lease is already an
/// orphan, and the server's TTL is the only remaining backstop.
async fn release_orphan(client: &Client, key: &str, id: &str) {
let mut c = client.locks_client();
let call = c.un_lock(Request::new(UnLockRequest {
key: key.to_string(),
id: id.to_string(),
}));
let _ = tokio::time::timeout(HOLD_RPC_TIMEOUT, call).await;
}
/// Run `fut`, returning `None` if `stop` fires first.
async fn race_stop<F: std::future::Future>(fut: F, stop: &Notify) -> Option<F::Output> {
tokio::select! {
@ -1014,7 +1108,11 @@ impl Client {
async fn acquire_lock_inner(&self, key: String, config: Config, read: bool) -> Result<Lock> {
let scope = config.scope;
let req = config.into_request(key.clone());
let lease_ms = req.max_lease_period;
// Template for transparent re-acquire after a stream drop: same
// request_id (so a still-held lease is recovered idempotently
// rather than re-contended) but never block.
let mut reacquire_req = req.clone();
reacquire_req.max_wait_period = 0;
let mut c = self.locks_client();
let mut stream = if read {
@ -1023,8 +1121,9 @@ impl Client {
c.lock(Request::new(req)).await?.into_inner()
};
// Consume events until we see Acquired / Failed / Expired. Whatever
// remains of the stream after Acquired goes to the hold task.
// Consume events until we see Acquired / Failed / Expired. The
// stream stays open after Acquired — the hold task keeps it bound
// (and re-binds it across primary bounces).
loop {
let event = stream.message().await?;
let event = match event {
@ -1050,13 +1149,14 @@ impl Client {
let (state_tx, state_rx) = watch::channel(init.clone());
let stop = Arc::new(Notify::new());
let stopped = Arc::new(AtomicBool::new(false));
// Hand the stream to the hold task, which drains it,
// then polls ownership and publishes lease/loss via
// Hand the open stream to the hold task, which parks
// it, re-binds on drop, and publishes fence/loss via
// `state_tx`.
let hold = tokio::spawn(hold_loop(
self.clone(),
key.clone(),
lease_ms,
read,
reacquire_req,
stream,
state_tx.clone(),
scope,
@ -1277,15 +1377,6 @@ mod tests {
);
}
#[test]
fn hold_poll_is_a_third_of_the_lease_within_bounds() {
assert_eq!(hold_poll_interval(30_000), Duration::from_secs(5));
assert_eq!(hold_poll_interval(9_000), Duration::from_secs(3));
assert_eq!(hold_poll_interval(60_000), HOLD_MAX_POLL);
assert_eq!(hold_poll_interval(10), HOLD_BASE_BACKOFF);
assert_eq!(hold_poll_interval(0), HOLD_BASE_BACKOFF);
}
#[test]
fn rebind_preserves_generation_but_rewin_increments_it() {
let (state_tx, _state_rx) = watch::channel(state("lease-1", 9, 4));

2
ts/package.json
View File

@ -1,6 +1,6 @@
{
"name": "@waymaker/client",
"version": "0.1.33",
"version": "0.1.32",
"description": "Official TypeScript client for waymaker — locks, streams, KV, collections, sketches, cache, object store",
"repository": "https://git.awesomike.com/pub/waymaker-client",
"license": "MIT OR Apache-2.0",