diff --git a/VERSION b/VERSION index db7a480..28d0075 100644 --- a/VERSION +++ b/VERSION @@ -1 +1 @@ -0.1.31 +0.1.32 diff --git a/rust/Cargo.toml b/rust/Cargo.toml index a05a298..6f06179 100644 --- a/rust/Cargo.toml +++ b/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" diff --git a/rust/src/lock/mod.rs b/rust/src/lock/mod.rs index e972866..7ee1bdd 100644 --- a/rust/src/lock/mod.rs +++ b/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, + /// 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, _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` - // is the backstop. + // 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); self.stop.notify_one(); - self._hold.abort(); } } @@ -768,6 +795,7 @@ async fn hold_loop( state_tx: watch::Sender, scope: Scope, stop: Arc, + stopped: Arc, ) { 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 - // promptly — and, crucially, before any re-acquire could - // resurrect a released lock. - let rebound = - match race_stop(reacquire(&client, read, &reacquire_req, &key, scope), &stop).await - { - Some(r) => r, - None => return, - }; + 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) => { @@ -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(fut: F, stop: &Notify) -> Option { 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, }); } diff --git a/ts/package.json b/ts/package.json index 705a8ae..43a741c 100644 --- a/ts/package.json +++ b/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",