use std::net::SocketAddr;
use std::time::Instant;
use super::frame::SessionId;
use super::migration::MigrationState;
use super::pacing::{FramePacer, RetransmitController};
use super::timing::{RttEstimator, TimestampTracker};
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum ConnectionPhase {
Handshaking,
Established,
Closing,
Closed,
Failed,
}
#[derive(Debug, Clone)]
pub struct NonceWindow {
highest: u64,
window: [u64; 32], }
impl Default for NonceWindow {
fn default() -> Self {
Self::new()
}
}
impl NonceWindow {
pub const WINDOW_SIZE: usize = 2048;
pub fn new() -> Self {
Self {
highest: 0,
window: [0; 32],
}
}
pub fn check_and_mark(&mut self, nonce: u64) -> bool {
if self.highest == 0 && nonce > 0 {
self.highest = nonce;
return true;
}
if nonce > self.highest {
let shift = (nonce - self.highest) as usize;
self.shift_window(shift);
self.highest = nonce;
true
} else if nonce == self.highest {
false
} else {
let offset = (self.highest - nonce) as usize;
if offset > Self::WINDOW_SIZE {
return false;
}
let offset = offset - 1; let word_idx = offset / 64;
let bit_idx = offset % 64;
let mask = 1u64 << bit_idx;
if self.window[word_idx] & mask != 0 {
false
} else {
self.window[word_idx] |= mask;
true
}
}
}
fn shift_window(&mut self, shift: usize) {
if shift >= Self::WINDOW_SIZE {
self.window = [0; 32];
return;
}
let word_shift = shift / 64;
let bit_shift = shift % 64;
if word_shift > 0 {
for i in (word_shift..32).rev() {
self.window[i] = self.window[i - word_shift];
}
for i in 0..word_shift {
self.window[i] = 0;
}
}
if bit_shift > 0 {
let mut carry = 0u64;
for i in (0..32).rev() {
let new_carry = self.window[i] << (64 - bit_shift);
self.window[i] = (self.window[i] >> bit_shift) | carry;
carry = new_carry;
}
}
if shift > 0 {
let offset = shift - 1;
if offset < Self::WINDOW_SIZE {
let word_idx = offset / 64;
let bit_idx = offset % 64;
self.window[word_idx] |= 1u64 << bit_idx;
}
}
}
}
#[derive(Debug)]
pub struct ConnectionState {
pub session_id: SessionId,
pub phase: ConnectionPhase,
pub remote_endpoint: SocketAddr,
pub last_received: Instant,
pub epoch: u32,
pub send_nonce: u64,
pub recv_nonce_window: NonceWindow,
pub rtt: RttEstimator,
pub timestamps: TimestampTracker,
pub pacer: FramePacer,
pub retransmit: RetransmitController,
pub migration: MigrationState,
pub local_state_version: u64,
pub remote_state_version: u64,
pub acked_state_version: u64,
}
impl ConnectionState {
pub fn new(session_id: SessionId, remote_endpoint: SocketAddr) -> Self {
let now = Instant::now();
Self {
session_id,
phase: ConnectionPhase::Established,
remote_endpoint,
last_received: now,
epoch: 0,
send_nonce: 0,
recv_nonce_window: NonceWindow::new(),
rtt: RttEstimator::new(),
timestamps: TimestampTracker::new(),
pacer: FramePacer::new(),
retransmit: RetransmitController::new(super::timing::constants::INITIAL_RTO),
migration: MigrationState::new(remote_endpoint),
local_state_version: 0,
remote_state_version: 0,
acked_state_version: 0,
}
}
pub fn handshaking(remote_endpoint: SocketAddr) -> Self {
let now = Instant::now();
Self {
session_id: SessionId::zero(),
phase: ConnectionPhase::Handshaking,
remote_endpoint,
last_received: now,
epoch: 0,
send_nonce: 0,
recv_nonce_window: NonceWindow::new(),
rtt: RttEstimator::new(),
timestamps: TimestampTracker::new(),
pacer: FramePacer::new(),
retransmit: RetransmitController::new(super::timing::constants::INITIAL_RTO),
migration: MigrationState::new(remote_endpoint),
local_state_version: 0,
remote_state_version: 0,
acked_state_version: 0,
}
}
pub fn next_send_nonce(&mut self) -> u64 {
let nonce = self.send_nonce;
self.send_nonce = self.send_nonce.saturating_add(1);
nonce
}
pub fn check_recv_nonce(&mut self, nonce: u64) -> bool {
self.recv_nonce_window.check_and_mark(nonce)
}
pub fn on_authenticated_frame(&mut self, from: SocketAddr) {
self.last_received = Instant::now();
if from != self.remote_endpoint && self.migration.validate_address(from) {
self.remote_endpoint = from;
}
}
pub fn is_alive(&self) -> bool {
!self.pacer.is_connection_dead(self.last_received) && !self.retransmit.is_failed()
}
pub fn is_failed(&self) -> bool {
self.phase == ConnectionPhase::Failed
|| self.pacer.is_connection_dead(self.last_received)
|| self.retransmit.is_failed()
}
pub fn has_unacked_data(&self) -> bool {
self.local_state_version > self.acked_state_version
}
pub fn on_ack(&mut self, acked_version: u64) {
if acked_version > self.acked_state_version {
self.acked_state_version = acked_version;
self.retransmit.on_ack();
}
}
pub fn close(&mut self) {
self.phase = ConnectionPhase::Closing;
}
pub fn mark_closed(&mut self) {
self.phase = ConnectionPhase::Closed;
}
pub fn mark_failed(&mut self) {
self.phase = ConnectionPhase::Failed;
}
pub fn complete_handshake(&mut self, session_id: SessionId) {
self.session_id = session_id;
self.phase = ConnectionPhase::Established;
self.timestamps = TimestampTracker::new(); }
pub fn on_rekey(&mut self) {
self.epoch = self.epoch.saturating_add(1);
}
}
#[cfg(test)]
mod tests {
use super::*;
use std::net::{IpAddr, Ipv4Addr};
fn test_addr(port: u16) -> SocketAddr {
SocketAddr::new(IpAddr::V4(Ipv4Addr::new(127, 0, 0, 1)), port)
}
#[test]
fn test_nonce_window_new() {
let mut window = NonceWindow::new();
assert!(window.check_and_mark(1));
assert!(!window.check_and_mark(1));
assert!(window.check_and_mark(2));
}
#[test]
fn test_nonce_window_gap() {
let mut window = NonceWindow::new();
assert!(window.check_and_mark(1));
assert!(window.check_and_mark(100));
assert!(window.check_and_mark(50));
assert!(window.check_and_mark(75));
assert!(!window.check_and_mark(50));
assert!(!window.check_and_mark(100));
}
#[test]
fn test_nonce_window_too_old() {
let mut window = NonceWindow::new();
assert!(window.check_and_mark(3000));
assert!(!window.check_and_mark(1));
assert!(!window.check_and_mark(500)); }
#[test]
fn test_connection_state_nonces() {
let mut conn = ConnectionState::new(SessionId::zero(), test_addr(8080));
assert_eq!(conn.next_send_nonce(), 0);
assert_eq!(conn.next_send_nonce(), 1);
assert_eq!(conn.next_send_nonce(), 2);
assert_eq!(conn.send_nonce, 3);
}
#[test]
fn test_connection_state_lifecycle() {
let addr = test_addr(8080);
let mut conn = ConnectionState::handshaking(addr);
assert_eq!(conn.phase, ConnectionPhase::Handshaking);
let session_id = SessionId::from_bytes([1, 2, 3, 4, 5, 6]);
conn.complete_handshake(session_id);
assert_eq!(conn.phase, ConnectionPhase::Established);
assert_eq!(conn.session_id, session_id);
conn.close();
assert_eq!(conn.phase, ConnectionPhase::Closing);
conn.mark_closed();
assert_eq!(conn.phase, ConnectionPhase::Closed);
}
#[test]
fn test_connection_state_ack() {
let mut conn = ConnectionState::new(SessionId::zero(), test_addr(8080));
conn.local_state_version = 10;
assert!(conn.has_unacked_data());
conn.on_ack(5);
assert_eq!(conn.acked_state_version, 5);
assert!(conn.has_unacked_data());
conn.on_ack(10);
assert_eq!(conn.acked_state_version, 10);
assert!(!conn.has_unacked_data());
}
#[test]
fn test_connection_alive_check() {
let conn = ConnectionState::new(SessionId::zero(), test_addr(8080));
assert!(conn.is_alive());
assert!(!conn.is_failed());
}
}