1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
//! Per-session concurrency gate.
//!
//! Prevents concurrent runs from interleaving writes to the same session.
//! Each session is protected by its own [`tokio::sync::Mutex`]; a run
//! acquires the lock at entry and holds it until the run completes or
//! is cancelled.
use std::collections::HashMap;
use std::sync::Arc;
use tokio::sync::{Mutex, RwLock};
use uuid::Uuid;
/// RAII guard that holds a per-session lock for the duration of a run.
///
/// Returned by [`SessionGate::acquire`]. Dropping this guard releases the
/// session lock, allowing the next queued run to proceed.
pub struct SessionGuard {
_inner: tokio::sync::OwnedMutexGuard<()>,
}
impl SessionGuard {
/// Creates a new guard wrapping an acquired mutex lock.
fn new(inner: tokio::sync::OwnedMutexGuard<()>) -> Self {
Self { _inner: inner }
}
}
impl std::fmt::Debug for SessionGuard {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
f.debug_struct("SessionGuard").finish_non_exhaustive()
}
}
/// Registry of per-session mutexes.
///
/// Uses a `RwLock<HashMap>` to lazily create mutex entries. The outer
/// `RwLock` is held briefly (read or write) to look up / insert a
/// mutex *handle*; the actual session serialization happens via the
/// inner `Mutex`.
#[derive(Debug, Clone, Default)]
pub struct SessionGate {
locks: Arc<RwLock<HashMap<Uuid, Arc<Mutex<()>>>>>,
}
impl SessionGate {
/// Creates a new empty gate.
#[must_use]
pub fn new() -> Self {
Self::default()
}
/// Acquires a session lock, returning an RAII guard.
///
/// If the session is already held by another run, this returns
/// immediately with an error rather than waiting — concurrent
/// writes to the same session are a programming mistake, not a
/// normal contention scenario.
///
/// # Errors
///
/// Returns [`SessionBusy`] if the lock is already held.
pub async fn acquire(&self, session_id: Uuid) -> Result<SessionGuard, SessionBusy> {
let mutex = {
let read = self.locks.read().await;
read.get(&session_id).cloned()
};
let mutex = if let Some(m) = mutex {
m
} else {
let mut write = self.locks.write().await;
write
.entry(session_id)
.or_insert_with(|| Arc::new(Mutex::new(())))
.clone()
};
// Try to acquire the lock without blocking. If another run
// already holds it, return an error immediately.
let guard = mutex
.try_lock_owned()
.map_err(|_| SessionBusy { session_id })?;
Ok(SessionGuard::new(guard))
}
}
/// Error returned when a session is already being processed by another run.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct SessionBusy {
/// The session that was already locked.
pub session_id: Uuid,
}
impl std::fmt::Display for SessionBusy {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
write!(
f,
"session {} is busy — another run is active",
self.session_id
)
}
}
impl std::error::Error for SessionBusy {}
#[cfg(test)]
#[allow(clippy::unwrap_used, clippy::expect_used)]
mod tests {
use super::*;
#[tokio::test]
async fn acquire_and_release() {
let gate = SessionGate::new();
let sid = Uuid::new_v4();
let guard = gate.acquire(sid).await.expect("should acquire");
// Guard is alive — session is locked.
drop(guard);
// After drop, the same session can be acquired again.
let _guard2 = gate
.acquire(sid)
.await
.expect("should re-acquire after drop");
}
#[tokio::test]
async fn concurrent_acquire_fails() {
let gate = SessionGate::new();
let sid = Uuid::new_v4();
let guard = gate.acquire(sid).await.expect("first acquire");
let result = gate.acquire(sid).await;
assert!(result.is_err(), "second acquire should fail");
assert_eq!(result.unwrap_err().session_id, sid);
drop(guard);
// After release, acquire succeeds.
let _guard2 = gate
.acquire(sid)
.await
.expect("should succeed after release");
}
#[tokio::test]
async fn independent_sessions_no_contention() {
let gate = SessionGate::new();
let sid1 = Uuid::new_v4();
let sid2 = Uuid::new_v4();
let guard1 = gate.acquire(sid1).await.expect("acquire sid1");
let guard2 = gate.acquire(sid2).await.expect("acquire sid2");
// Both acquired — no contention.
drop(guard1);
drop(guard2);
}
#[tokio::test]
async fn sequential_runs_serialize() {
let gate = SessionGate::new();
let sid = Uuid::new_v4();
let g1 = gate.acquire(sid).await.expect("first");
// Simulate first run finishing.
drop(g1);
let g2 = gate.acquire(sid).await.expect("second");
drop(g2);
}
}