fix(lock): a held lock polls ownership instead of re-acquiring in a spin (v0.1.33)

The server ends a Lock stream at grant (`sender_stream` is taken), so the
hold task always saw "disconnected" right after acquiring and re-acquired
with the original request_id. Each re-bind got a stream that ended at once,
and the Bound path had no backoff: every held lock spun at hundreds of Lock
RPCs per second (~350/s measured direct, ~10k in 3s from one leptin pod at
boot). On leptin prod (2026-10-04) the flood starved ExtendLease, leases
lapsed by TTL (430 expiries on one node in 5h), and the spinning re-acquire
re-won them under fresh ids: the holder's generation changed under it, a
second pod could be granted the key meanwhile, and leptin-server — which
treats a generation change as loss — crash-looped for 13h.

- After the stream ends, the hold task never re-acquires. It re-proves
  ownership via lease_status every lease/3 (200ms..5s) and publishes `lost`
  when it cannot.
- The server reports a missing lease as Unavailable (indistinguishable from
  a transport error), so `Lost` was unreachable and a lapsed lease was never
  reported. An unproven lease past its last proven expiry is now `lost`.
- Removes the now-unused re-acquire / orphan-release path.

Verified live against the leptin prod cluster through the istio mesh: two
pods contending on one key for 3 min — 36 polls, 0 generation changes,
mutual exclusion held; a 3s lease without renewal is reported lost at 3.0s.

The Go and TS clients still use the re-bind design and are NOT fixed here.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01MCJdDXt4eEwJc7HHiUrM8d

VERSION

@@ -1 +1 @@
-0.1.32
+0.1.33

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"

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
+//! The returned [`Lock`] keeps a background task that drains the
-//! server event stream open and, if that stream drops (e.g. the
+//! server event stream and, once that stream ends (the server closes
-//! key's primary bounces), transparently re-binds it and re-confirms
+//! it at grant; the lease lives on its TTL plus renewal), re-proves
-//! ownership — reusing the original `request_id` so a still-held
+//! ownership through `lease_status` on a cadence of a third of the
-//! lease is recovered rather than re-contended. The lock's live
+//! lease. It never re-acquires on its own: a lease it can no longer
-//! state (fence token, lease expiry, and a `lost` flag) is published
+//! prove is published as `lost`, and re-acquiring is the caller's
-//! through [`Lock::watch`]; re-read [`Lock::fence_token`] before every
+//! decision. The lock's live state (fence token, lease expiry, and a
-//! fenced side effect, because a lost-then-re-won lock carries a new,
+//! `lost` flag) is published through [`Lock::watch`]; re-read
-//! higher token.
+//! [`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
+/// - **Client-side ownership checks.** [`Lock`] re-proves ownership via
-///   its event stream after a disconnect, but that is a client-lib
+///   `lease_status` after its event stream ends, but that is a
-///   behaviour — no scope value changes it.
+///   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
+/// A background task drains the server event stream, then re-proves
-/// stream drops — typically because the key's primary bounced — the
+/// ownership via `lease_status` once it ends — which the server does
-/// task transparently re-binds it, reusing the original `request_id`
+/// at grant. The lease itself is kept alive by the server's TTL plus
-/// so a still-held lease is recovered rather than re-contended. The
+/// [`Lock::spawn_renewal`], independent of the stream: a closed stream
-/// lease itself is kept alive by the server's TTL plus
+/// loses event visibility, not the lock. Live state is exposed through
-/// [`Lock::spawn_renewal`], independent of the stream: dropping the
+/// the accessors below and [`Lock::watch`].
-/// 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
+        // Stop the hold task FIRST and wait for it to wind down, so nothing
-        // are about to release — a re-acquire racing the UnLock would
+        // it publishes (or marks lost) races the UnLock. The task never
-        // resurrect a lock the caller believes is gone.
+        // 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
+        // Read the id AFTER the task exited, so the release names the
-        // completed while we were stopping may have changed it.
+        // 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.
+        // unlocked) handle does not keep polling in the background. The
-        // Deliberately NOT `abort()`: aborting mid-acquire strands a lease
+        // task honours `stopped` at every step, so it exits on its own
-        // the server may still grant (see `unlock`). The task honours
+        // within one bounded RPC.
-        // `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);
@@ -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
+// stream (granting a lock sets its `sender_stream` to None, which ends
-// dropped stream means "lost event visibility", not "lost lock". This
+// the stream). So an ended stream means "lost event visibility", not
-// task re-binds the stream when it can, and only declares the lock
+// "lost lock". This task then polls lease_status, and only declares the
-// `lost` when it can no longer prove ownership via lease_status.
+// 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<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),
@@ -781,65 +758,41 @@ 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
+/// Background task body: drain the event stream, then poll ownership,
-/// surface fence-token changes / loss through `state_tx`.
+/// and surface lease updates / loss through `state_tx`.
 #[allow(clippy::too_many_arguments)]
 async fn hold_loop(
     client: Client,
     key: String,
-    read: bool,
+    lease_ms: u32,
-    reacquire_req: LockRequest,
     mut stream: tonic::Streaming<LockEvent>,
     state_tx: watch::Sender<LockState>,
     scope: Scope,
     stop: Arc<Notify>,
     stopped: Arc<AtomicBool>,
 ) {
-    loop {
+    if let Drained::Stopped = drain_stream(&mut stream, &state_tx, &key, scope, &stop).await {
-        match drain_stream(&mut stream, &state_tx, &key, scope, &stop).await {
+        return;
-            Drained::Stopped => return,
-            Drained::Disconnected => {}
     }
 
-        // Re-establish: re-bind the stream; if we cannot, decide whether
+    // The server ends the stream as soon as it grants (the entry's
-        // we still own the lease or have truly lost it.
+    // `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 {
         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),
@@ -850,15 +803,37 @@ async fn hold_loop(
             Some(o) => o,
             None => return,
         };
-                    match owned {
+        let wait = 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
@@ -867,20 +842,14 @@ async fn hold_loop(
             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
@@ -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<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.
@@ -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<F: std::future::Future>(fut: F, stop: &Notify) -> Option<F::Output> {
     tokio::select! {
@@ -1108,11 +1014,7 @@ 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());
-        // Template for transparent re-acquire after a stream drop: same
+        let lease_ms = req.max_lease_period;
-        // 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 {
@@ -1121,9 +1023,8 @@ impl Client {
             c.lock(Request::new(req)).await?.into_inner()
         };
 
-        // Consume events until we see Acquired / Failed / Expired. The
+        // Consume events until we see Acquired / Failed / Expired. Whatever
-        // stream stays open after Acquired — the hold task keeps it bound
+        // remains of the stream after Acquired goes to the hold task.
-        // (and re-binds it across primary bounces).
         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
+                    // Hand the stream to the hold task, which drains it,
-                    // it, re-binds on drop, and publishes fence/loss via
+                    // then polls ownership and publishes lease/loss via
                     // `state_tx`.
                     let hold = tokio::spawn(hold_loop(
                         self.clone(),
                         key.clone(),
-                        read,
+                        lease_ms,
-                        reacquire_req,
                         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));

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",