#[cfg(test)]
use std::num::NonZeroUsize;
use std::sync::atomic::{AtomicU64, Ordering};
const STATE_SHIFT: u32 = 56;
const COUNT_MASK: u64 = (1u64 << STATE_SHIFT) - 1;
const STATE_LIVE_TAG: u64 = 0;
const STATE_DYING_TAG: u64 = 1;
const STATE_DEAD_TAG: u64 = 2;
#[cfg(test)]
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub(crate) enum SessionLifecycleSnapshot {
Live(NonZeroUsize),
Dying,
Dead,
}
pub(crate) struct SessionLifecycle {
inner: AtomicU64,
}
impl SessionLifecycle {
pub(crate) fn new() -> Self {
Self {
inner: AtomicU64::new(encode_live(1)),
}
}
#[cfg(test)]
pub(crate) fn snapshot(&self) -> SessionLifecycleSnapshot {
decode(self.inner.load(Ordering::SeqCst))
}
pub(crate) fn is_torn_down(&self) -> bool {
self.inner.load(Ordering::Acquire) >> STATE_SHIFT != STATE_LIVE_TAG
}
pub(crate) fn live_count(&self) -> usize {
let v = self.inner.load(Ordering::SeqCst);
if v >> STATE_SHIFT == STATE_LIVE_TAG {
(v & COUNT_MASK) as usize
} else {
0
}
}
pub(crate) fn try_bump_live(&self) -> bool {
let mut v = self.inner.load(Ordering::SeqCst);
loop {
if v >> STATE_SHIFT != STATE_LIVE_TAG {
return false;
}
let new_v = v + 1;
debug_assert!(
new_v >> STATE_SHIFT == STATE_LIVE_TAG,
"SessionLifecycle live-count overflow"
);
match self
.inner
.compare_exchange_weak(v, new_v, Ordering::SeqCst, Ordering::SeqCst)
{
Ok(_) => return true,
Err(actual) => v = actual,
}
}
}
pub(crate) fn drop_connection(&self) -> bool {
let mut v = self.inner.load(Ordering::SeqCst);
loop {
debug_assert_eq!(
v >> STATE_SHIFT,
STATE_LIVE_TAG,
"drop_connection called from non-Live state"
);
let count = v & COUNT_MASK;
debug_assert!(count >= 1, "Live state with count == 0");
let (new_v, was_last) = if count == 1 {
(encode_dying(), true)
} else {
(v - 1, false)
};
match self
.inner
.compare_exchange_weak(v, new_v, Ordering::SeqCst, Ordering::SeqCst)
{
Ok(_) => return was_last,
Err(actual) => v = actual,
}
}
}
pub(crate) fn mark_dead(&self) {
let prev = self.inner.swap(encode_dead(), Ordering::SeqCst);
debug_assert_eq!(
prev >> STATE_SHIFT,
STATE_DYING_TAG,
"mark_dead called from non-Dying state"
);
}
}
fn encode_live(n: usize) -> u64 {
debug_assert!(n >= 1 && (n as u64) <= COUNT_MASK);
n as u64
}
fn encode_dying() -> u64 {
STATE_DYING_TAG << STATE_SHIFT
}
fn encode_dead() -> u64 {
STATE_DEAD_TAG << STATE_SHIFT
}
#[cfg(test)] fn decode(v: u64) -> SessionLifecycleSnapshot {
match v >> STATE_SHIFT {
STATE_LIVE_TAG => {
let count = (v & COUNT_MASK) as usize;
SessionLifecycleSnapshot::Live(
NonZeroUsize::new(count).expect("Live state must have count >= 1"),
)
}
STATE_DYING_TAG => SessionLifecycleSnapshot::Dying,
STATE_DEAD_TAG => SessionLifecycleSnapshot::Dead,
other => unreachable!("invalid SessionLifecycle state tag: {other}"),
}
}
#[cfg(test)]
mod tests {
use super::*;
use std::sync::Arc;
use std::thread;
fn nz(n: usize) -> NonZeroUsize {
NonZeroUsize::new(n).unwrap()
}
#[test]
fn new_session_is_live_one() {
let lc = SessionLifecycle::new();
assert_eq!(lc.snapshot(), SessionLifecycleSnapshot::Live(nz(1)));
assert_eq!(lc.live_count(), 1);
assert!(!lc.is_torn_down());
}
#[test]
fn bump_then_decrement_back_to_one_stays_live() {
let lc = SessionLifecycle::new();
assert!(lc.try_bump_live()); assert_eq!(lc.live_count(), 2);
assert!(!lc.drop_connection()); assert_eq!(lc.live_count(), 1);
assert!(!lc.is_torn_down());
}
#[test]
fn last_drop_transitions_through_dying_to_dead() {
let lc = SessionLifecycle::new();
assert!(lc.drop_connection()); assert_eq!(lc.snapshot(), SessionLifecycleSnapshot::Dying);
assert_eq!(lc.live_count(), 0);
assert!(lc.is_torn_down());
lc.mark_dead();
assert_eq!(lc.snapshot(), SessionLifecycleSnapshot::Dead);
assert!(lc.is_torn_down());
}
#[test]
fn try_bump_after_dying_or_dead_refuses() {
let lc = SessionLifecycle::new();
assert!(lc.drop_connection()); assert!(!lc.try_bump_live(), "Dying must refuse new attach");
lc.mark_dead();
assert!(!lc.try_bump_live(), "Dead must refuse new attach");
}
#[test]
fn concurrent_bumpers_against_drop_never_attach_to_dying() {
for _ in 0..32 {
let lc = Arc::new(SessionLifecycle::new());
assert!(lc.try_bump_live());
assert_eq!(lc.live_count(), 2);
let dropper = {
let lc = Arc::clone(&lc);
thread::spawn(move || {
let _ = lc.drop_connection();
let was_last = lc.drop_connection();
if was_last {
lc.mark_dead();
}
})
};
let attackers: Vec<_> = (0..8)
.map(|_| {
let lc = Arc::clone(&lc);
thread::spawn(move || {
let ok = lc.try_bump_live();
if ok {
assert!(matches!(lc.snapshot(), SessionLifecycleSnapshot::Live(_)));
if lc.drop_connection() {
lc.mark_dead();
}
}
ok
})
})
.collect();
dropper.join().unwrap();
for a in attackers {
let _ = a.join().unwrap();
}
match lc.snapshot() {
SessionLifecycleSnapshot::Dead => {}
SessionLifecycleSnapshot::Live(_) => {
}
SessionLifecycleSnapshot::Dying => {
panic!("Dying is transient — should have moved to Dead via mark_dead");
}
}
}
}
#[test]
fn dying_window_is_observable_to_hot_path_checks() {
let lc = SessionLifecycle::new();
assert!(!lc.is_torn_down());
assert!(lc.drop_connection());
assert!(
lc.is_torn_down(),
"Dying must surface as is_torn_down() == true \
before mark_dead, so drop-path reapers skip"
);
lc.mark_dead();
assert!(lc.is_torn_down());
}
}