diff --git a/VERSION b/VERSION index 28d0075..50140e3 100644 --- a/VERSION +++ b/VERSION @@ -1 +1 @@ -0.1.32 +0.1.33 diff --git a/rust/Cargo.toml b/rust/Cargo.toml index 6f06179..6309e2c 100644 --- a/rust/Cargo.toml +++ b/rust/Cargo.toml @@ -1,6 +1,6 @@ [package] name = "waymaker-client" -version = "0.1.32" +version = "0.1.33" edition = "2021" description = "Official Rust client for waymaker — locks, streams, KV, collections, sketches, cache, object store" repository = "https://git.awesomike.com/pub/waymaker-client" diff --git a/rust/src/lock/mod.rs b/rust/src/lock/mod.rs index 7ee1bdd..c9f8fad 100644 --- a/rust/src/lock/mod.rs +++ b/rust/src/lock/mod.rs @@ -7,15 +7,15 @@ //! - `client.lease_status(&lock_id)` //! - `client.multi_lock(...)` //! -//! 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. +//! 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. //! //! ## 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 transparency.** [`Lock`] transparently re-binds -/// its event stream after a disconnect, but that is a client-lib -/// behaviour — no scope value changes it. +/// - **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. /// - **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 heartbeats / re-acquire. + /// Lease expiry (epoch ms), refreshed from renewal and lease_status. 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,14 +243,12 @@ 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 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`]. +/// 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`]. pub struct Lock { client: Client, pub key: String, @@ -577,27 +575,19 @@ 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 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. + // 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`. 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: a transparent re-bind that - // completed while we were stopping may have changed it. + // Read the id AFTER the task exited, so the release names the + // last lease the task published. let id = self.id(); let mut c = self.client.locks_client(); let r = c @@ -732,11 +722,9 @@ 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 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. + // 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. // (`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); @@ -744,13 +732,13 @@ impl Drop for Lock { } } -// ---- Background hold / re-bind / loss-detection loop ----------------- +// ---- Background hold / 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). 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. +// 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. /// Outcome of draining the current event stream. enum Drained { @@ -761,17 +749,6 @@ 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>, 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), @@ -781,108 +758,100 @@ 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: hold the event stream, re-bind it on drop, and -/// surface fence-token changes / loss through `state_tx`. +/// Background task body: drain the event stream, then poll ownership, +/// and surface lease updates / loss through `state_tx`. #[allow(clippy::too_many_arguments)] async fn hold_loop( client: Client, key: String, - read: bool, - reacquire_req: LockRequest, + lease_ms: u32, mut stream: tonic::Streaming, state_tx: watch::Sender, scope: Scope, stop: Arc, stopped: Arc, ) { + if let Drained::Stopped = drain_stream(&mut stream, &state_tx, &key, scope, &stop).await { + return; + } + + // 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); + let mut backoff = HOLD_BASE_BACKOFF; loop { - match drain_stream(&mut stream, &state_tx, &key, scope, &stop).await { - Drained::Stopped => return, - Drained::Disconnected => {} + if stopped.load(Ordering::SeqCst) { + return; } - - // 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) { + let current = state_tx.borrow().clone(); + let owned = match race_stop( + confirm_ownership(&client, &key, ¤t.id, current.fence_token, scope), + &stop, + ) + .await + { + Some(o) => o, + None => return, + }; + let wait = match owned { + Ownership::Held(grant) => { + // Still ours: renewal keeps the lease alive. + 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; } - // 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. - } + Ownership::Unknown => { + let wait = backoff; + backoff = (backoff * 2).min(HOLD_MAX_BACKOFF); + wait + } + Ownership::Lost => { + mark_lost(&state_tx); 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, ¤t.id, current.fence_token, scope), - &stop, - ) - .await - { - Some(o) => o, - None => return, - }; - match owned { - Ownership::Held(grant) => { - // Still ours (renewal keeps the lease alive); - // we just could not get an event stream — keep - // monitoring with backoff. - publish_grant(&state_tx, grant); - } - Ownership::Unknown => { /* transient — back off, retry */ } - Ownership::Lost => { - state_tx.send_if_modified(|current| { - if current.lost { - false - } else { - current.lost = true; - true - } - }); - return; - } - } - if race_stop(tokio::time::sleep(backoff), &stop) - .await - .is_none() - { - return; - } - backoff = (backoff * 2).min(HOLD_MAX_BACKOFF); - } - } + }; + if race_stop(tokio::time::sleep(wait), &stop).await.is_none() { + return; } } } +fn mark_lost(state_tx: &watch::Sender) { + state_tx.send_if_modified(|current| { + if current.lost { + false + } else { + current.lost = true; + true + } + }); +} + +/// 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 /// reported gone, or `stop` fires. Heartbeat / Acquired updates flow /// into `state_tx`. @@ -946,57 +915,6 @@ 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, 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. @@ -1038,18 +956,6 @@ 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(fut: F, stop: &Notify) -> Option { tokio::select! { @@ -1108,11 +1014,7 @@ impl Client { async fn acquire_lock_inner(&self, key: String, config: Config, read: bool) -> Result { let scope = config.scope; let req = config.into_request(key.clone()); - // 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 lease_ms = req.max_lease_period; let mut c = self.locks_client(); let mut stream = if read { @@ -1121,9 +1023,8 @@ impl Client { c.lock(Request::new(req)).await?.into_inner() }; - // 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). + // Consume events until we see Acquired / Failed / Expired. Whatever + // remains of the stream after Acquired goes to the hold task. loop { let event = stream.message().await?; let event = match event { @@ -1149,14 +1050,13 @@ 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 open stream to the hold task, which parks - // it, re-binds on drop, and publishes fence/loss via + // Hand the stream to the hold task, which drains it, + // then polls ownership and publishes lease/loss via // `state_tx`. let hold = tokio::spawn(hold_loop( self.clone(), key.clone(), - read, - reacquire_req, + lease_ms, stream, state_tx.clone(), scope, @@ -1377,6 +1277,15 @@ 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)); diff --git a/ts/package.json b/ts/package.json index 43a741c..34ac570 100644 --- a/ts/package.json +++ b/ts/package.json @@ -1,6 +1,6 @@ { "name": "@waymaker/client", - "version": "0.1.32", + "version": "0.1.33", "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",