pub const SUBPROTOCOL_CAPABILITY_ANN: u16 = 0x0C00;
pub const SUBPROTOCOL_ROUTE_WITHDRAW: u16 = 0x0C01;
pub const SUBPROTOCOL_SCOPED_CAPABILITY_ANN: u16 = 0x0C04;
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub struct RouteWithdrawal {
pub dest: u64,
pub seq: u64,
}
impl RouteWithdrawal {
pub const SIZE: usize = 16;
pub fn to_bytes(&self) -> [u8; Self::SIZE] {
let mut buf = [0u8; Self::SIZE];
buf[..8].copy_from_slice(&self.dest.to_le_bytes());
buf[8..].copy_from_slice(&self.seq.to_le_bytes());
buf
}
pub fn from_bytes(data: &[u8]) -> Option<Self> {
if data.len() != Self::SIZE {
return None;
}
Some(Self {
dest: u64::from_le_bytes(data[..8].try_into().ok()?),
seq: u64::from_le_bytes(data[8..].try_into().ok()?),
})
}
}
#[derive(Debug, Default)]
pub struct WithdrawalSeqGate {
seen: dashmap::DashMap<(u64, u64), SeqEntry>,
tick: std::sync::atomic::AtomicU64,
}
#[derive(Debug)]
struct SeqEntry {
seq: u64,
touch: u64,
}
impl WithdrawalSeqGate {
const MAX_ENTRIES: usize = 8192;
const LOW_WATER: usize = 6144;
pub fn new() -> Self {
Self::default()
}
pub fn admit(&self, sender: u64, dest: u64, seq: u64) -> bool {
let touch = self.tick.fetch_add(1, std::sync::atomic::Ordering::Relaxed);
let mut admitted = false;
self.seen
.entry((sender, dest))
.and_modify(|e| {
e.touch = touch;
if seq > e.seq {
e.seq = seq;
admitted = true;
}
})
.or_insert_with(|| {
admitted = true;
SeqEntry { seq, touch }
});
self.evict_if_over_capacity();
admitted
}
fn evict_if_over_capacity(&self) {
if self.seen.len() <= Self::MAX_ENTRIES {
return;
}
let mut touches: Vec<u64> = self.seen.iter().map(|e| e.value().touch).collect();
if touches.len() <= Self::LOW_WATER {
return;
}
let cutoff_idx = touches.len() - Self::LOW_WATER;
touches.select_nth_unstable(cutoff_idx);
let cutoff = touches[cutoff_idx];
self.seen.retain(|_, e| e.touch >= cutoff);
}
pub fn forget_sender(&self, sender: u64) {
self.seen.retain(|(s, _), _| *s != sender);
}
}
#[cfg(test)]
mod tests {
use super::*;
#[test]
fn seq_gate_admits_strictly_newer_only() {
let gate = WithdrawalSeqGate::new();
assert!(gate.admit(1, 9, 5), "first sighting admits");
assert!(!gate.admit(1, 9, 5), "duplicate seq rejected");
assert!(!gate.admit(1, 9, 3), "older seq rejected");
assert!(gate.admit(1, 9, 6), "newer seq admits");
assert!(gate.admit(1, 8, 0), "different dest is a fresh pair");
assert!(gate.admit(2, 9, 0), "different sender is a fresh pair");
}
#[test]
fn seq_gate_forget_sender_resets_only_that_sender() {
let gate = WithdrawalSeqGate::new();
assert!(gate.admit(1, 9, 10));
assert!(gate.admit(2, 9, 10));
gate.forget_sender(1);
assert!(
gate.admit(1, 9, 0),
"forgotten sender's reset counter admits again"
);
assert!(
!gate.admit(2, 9, 0),
"other senders' history must survive the purge"
);
}
#[test]
fn seq_gate_survives_overflow_eviction() {
let gate = WithdrawalSeqGate::new();
for dest in 0..=(WithdrawalSeqGate::MAX_ENTRIES as u64) {
assert!(gate.admit(1, dest, 1));
}
assert!(gate.admit(1, 0, 1), "evicted pair's sighting admits");
assert!(!gate.admit(1, 0, 1), "gating resumes after the eviction");
}
#[test]
fn seq_gate_overflow_preserves_recently_active_ordering() {
let gate = WithdrawalSeqGate::new();
for dest in 0..(WithdrawalSeqGate::MAX_ENTRIES as u64) {
gate.admit(2, dest, 1);
}
assert!(gate.admit(1, 7, 100));
assert!(gate.admit(2, WithdrawalSeqGate::MAX_ENTRIES as u64, 1));
assert!(
!gate.admit(1, 7, 50),
"recently-active pair's ordering must survive overflow eviction",
);
assert!(gate.admit(1, 7, 101), "a genuinely newer seq still admits");
}
#[test]
fn seq_gate_admit_survives_the_overflow_it_triggers() {
let gate = WithdrawalSeqGate::new();
for dest in 0..(WithdrawalSeqGate::MAX_ENTRIES as u64) {
gate.admit(2, dest, 1);
}
assert!(gate.admit(1, 7, 100), "overflow-triggering pair admitted");
assert!(
!gate.admit(1, 7, 50),
"the pair that triggered the overflow kept its ordering",
);
}
#[test]
fn route_withdrawal_roundtrip() {
let w = RouteWithdrawal {
dest: 0xDEAD_BEEF_CAFE_F00D,
seq: 42,
};
assert_eq!(RouteWithdrawal::from_bytes(&w.to_bytes()), Some(w));
}
#[test]
fn route_withdrawal_rejects_wrong_length() {
assert_eq!(RouteWithdrawal::from_bytes(&[0u8; 15]), None);
assert_eq!(RouteWithdrawal::from_bytes(&[0u8; 17]), None);
assert_eq!(RouteWithdrawal::from_bytes(&[]), None);
}
}