fix(lock): never cancel an in-flight re-acquire; unlock joins the hold task

A released lock could come back as a ghost. `Lock::unlock` signalled the
background hold task and sent UnLock immediately; if the task had an
idempotent re-acquire RPC in flight, `race_stop` dropped the future
client-side but the server kept processing it — once UnLock had removed
the original lease, the same request_id re-WON the key with a fresh id
that no client held. On graylin prod that ghost writer sat on
`shutdown:pod:<id>:barrier` (created ~1 ms after "released :barrier
writer", then "lease_ttl pq has a removed node" and never reaped) and
parked every proxied request on the pod for the full 120 s restart wait.

- hold_loop no longer races the acquire RPC against stop; it lets the
  RPC finish (bounded by HOLD_RPC_TIMEOUT), then honours a level
  `stopped` flag and releases any lease it won while stopping.
- unlock() sets `stopped`, notifies, JOINS the hold task (abort only as
  a backstop past every bound), then UnLocks the *current* id.
- Drop sets the flag and notifies but no longer aborts mid-RPC.

Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01MCwe5Sf5NU8Ei13oqTcesj

VERSION

@@ -1 +1 @@
-0.1.31
+0.1.32

rust/Cargo.toml

@@ -1,6 +1,6 @@
 [package]
 name = "waymaker-client"
-version = "0.1.31"
+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"

rust/src/lock/mod.rs

@@ -39,6 +39,7 @@ use crate::server::{
     LeaseStatusRequest, LockEvent, LockEventType, LockRequest, MultiLockKey, MultiLockRequest,
     UnLockRequest,
 };
+use std::sync::atomic::{AtomicBool, Ordering};
 use std::sync::Arc;
 use std::time::{Duration, SystemTime, UNIX_EPOCH};
 use tokio::sync::{watch, Notify};
@@ -261,6 +262,11 @@ pub struct Lock {
     /// `Drop` so the task never re-acquires a deliberately-released
     /// lock.
     stop: Arc<Notify>,
+    /// Level-triggered twin of `stop`: a `Notify` permit can be consumed by
+    /// whichever `select!` observes it first, but the hold task also needs
+    /// to re-check "was I told to stop?" AFTER an in-flight acquire RPC
+    /// completes. See [`hold_loop`] for why that RPC is never cancelled.
+    stopped: Arc<AtomicBool>,
     _hold: tokio::task::JoinHandle<()>,
 }
 
@@ -570,11 +576,28 @@ 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(self) -> Result<()> {
+    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.
+        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.
         let id = self.id();
         let mut c = self.client.locks_client();
         let r = c
@@ -710,10 +733,14 @@ 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.
-        // `notify_one` leaves a permit if the task is mid-RPC; `abort`
+        // Deliberately NOT `abort()`: aborting mid-acquire strands a lease
-        // is the backstop.
+        // 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);
         self.stop.notify_one();
-        self._hold.abort();
     }
 }
 
@@ -768,6 +795,7 @@ async fn hold_loop(
     state_tx: watch::Sender<LockState>,
     scope: Scope,
     stop: Arc<Notify>,
+    stopped: Arc<AtomicBool>,
 ) {
     loop {
         match drain_stream(&mut stream, &state_tx, &key, scope, &stop).await {
@@ -779,15 +807,31 @@ async fn hold_loop(
         // we still own the lease or have truly lost it.
         let mut backoff = HOLD_BASE_BACKOFF;
         loop {
-            // Race every step against `stop` so unlock/drop halts us
+            if stopped.load(Ordering::SeqCst) {
-            // promptly — and, crucially, before any re-acquire could
+                return;
-            // resurrect a released lock.
+            }
-            let rebound =
+            // The acquire RPC is deliberately NOT raced against `stop`.
-                match race_stop(reacquire(&client, read, &reacquire_req, &key, scope), &stop).await
+            // Dropping an in-flight `Lock`/`ReadLock` request only resets the
-                {
+            // client side of the stream; the server keeps processing it and,
-                    Some(r) => r,
+            // once `unlock` has removed the original lease, the idempotent
-                    None => return,
+            // 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) => {
@@ -994,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! {
@@ -1092,6 +1148,7 @@ 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
                     // `state_tx`.
@@ -1104,6 +1161,7 @@ impl Client {
                         state_tx.clone(),
                         scope,
                         stop.clone(),
+                        stopped.clone(),
                     ));
                     return Ok(Lock {
                         client: self.clone(),
@@ -1112,6 +1170,7 @@ impl Client {
                         state_tx,
                         scope,
                         stop,
+                        stopped,
                         _hold: hold,
                     });
                 }

ts/package.json

@@ -1,6 +1,6 @@
 {
   "name": "@waymaker/client",
-  "version": "0.1.31",
+  "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",