use std::collections::VecDeque;
use std::fmt::Debug;
pub type Frame = u64;
#[derive(Debug, Clone)]
pub struct StateHistory<State: Clone> {
ring: VecDeque<State>,
base_frame: Frame,
capacity: usize,
resets: u64,
}
impl<State: Clone> StateHistory<State> {
pub fn new(capacity: usize) -> Self {
if capacity == 0 {
panic!("StateHistory capacity must be greater than 0");
}
Self {
ring: VecDeque::with_capacity(capacity),
base_frame: 0,
capacity,
resets: 0,
}
}
pub fn save(&mut self, frame: Frame, state: State) {
if self.ring.is_empty() {
self.base_frame = frame;
self.ring.push_back(state);
return;
}
let end = self.base_frame + self.ring.len() as u64; if frame == end {
self.ring.push_back(state);
while self.ring.len() > self.capacity {
self.ring.pop_front();
self.base_frame += 1;
}
} else if frame >= self.base_frame && frame < end {
let idx = (frame - self.base_frame) as usize;
self.ring[idx] = state;
} else {
self.resets += 1;
tracing::warn!(?frame, base = self.base_frame, len = self.ring.len(), "StateHistory: non-contiguous save, resetting the window");
self.ring.clear();
self.base_frame = frame;
self.ring.push_back(state);
}
}
pub fn restore(&self, frame: Frame) -> Option<State> {
if self.ring.is_empty() || frame < self.base_frame {
return None;
}
let idx = (frame - self.base_frame) as usize;
self.ring.get(idx).cloned()
}
pub fn oldest_frame(&self) -> Option<Frame> {
(!self.ring.is_empty()).then_some(self.base_frame)
}
pub fn latest_frame(&self) -> Option<Frame> {
(!self.ring.is_empty()).then(|| self.base_frame + self.ring.len() as u64 - 1)
}
pub fn len(&self) -> usize {
self.ring.len()
}
pub fn is_empty(&self) -> bool {
self.ring.is_empty()
}
pub fn clear(&mut self) {
self.ring.clear();
self.base_frame = 0;
}
pub fn resets(&self) -> u64 {
self.resets
}
}
#[derive(Debug, Clone)]
pub struct InputTimeline<Input: Clone + Debug> {
slots: VecDeque<Option<Input>>,
base_frame: Frame,
last_confirmed: Option<Frame>,
capacity: usize,
}
impl<Input: Clone + Debug> InputTimeline<Input> {
pub fn new(capacity: usize) -> Self {
if capacity == 0 {
panic!("InputTimeline capacity must be greater than 0");
}
Self {
slots: VecDeque::with_capacity(capacity),
base_frame: 0,
last_confirmed: None,
capacity,
}
}
pub fn confirm(&mut self, frame: Frame, input: Input) {
if self.slots.is_empty() {
self.base_frame = frame;
self.slots.push_back(Some(input));
} else if frame < self.base_frame {
return; } else {
let end = self.base_frame + self.slots.len() as u64;
if frame < end {
self.slots[(frame - self.base_frame) as usize] = Some(input);
} else {
while self.base_frame + (self.slots.len() as u64) < frame {
self.slots.push_back(None);
}
self.slots.push_back(Some(input));
}
while self.slots.len() > self.capacity {
self.slots.pop_front();
self.base_frame += 1;
}
}
self.last_confirmed = Some(self.last_confirmed.map_or(frame, |c| c.max(frame)));
}
pub fn confirmed_at(&self, frame: Frame) -> Option<&Input> {
if self.slots.is_empty() || frame < self.base_frame {
return None;
}
let idx = (frame - self.base_frame) as usize;
self.slots.get(idx).and_then(|s| s.as_ref())
}
pub fn last_confirmed_at_or_before(&self, frame: Frame) -> Option<&Input> {
if self.slots.is_empty() || frame < self.base_frame {
return None;
}
let end = self.base_frame + self.slots.len() as u64;
let mut f = frame.min(end - 1);
loop {
let idx = (f - self.base_frame) as usize;
if let Some(input) = self.slots[idx].as_ref() {
return Some(input);
}
if f == self.base_frame {
return None;
}
f -= 1;
}
}
pub fn last_confirmed_frame(&self) -> Option<Frame> {
self.last_confirmed
}
}
#[derive(Debug, Clone, Copy)]
pub struct RollbackConfig {
pub max_rollback_frames: usize,
}
impl Default for RollbackConfig {
fn default() -> Self {
Self { max_rollback_frames: 240 }
}
}
pub fn repeat_last_input<Input: Clone>(last: &Input, _frame: Frame) -> Input {
last.clone()
}
pub struct RollbackSession<State: Clone + Debug, Input: Clone + Debug + PartialEq> {
state_history: StateHistory<State>,
used: Vec<StateHistory<Input>>,
timelines: Vec<InputTimeline<Input>>,
neutral: Vec<Input>,
current_state: State,
head_frame: Frame,
advance: fn(&State, &[Input]) -> State,
predictor: fn(&Input, Frame) -> Input,
rollback_enabled: bool,
earliest_incorrect: Option<Frame>,
last_rollback_len: usize,
max_rollback_len: usize,
rollback_count: u64,
}
impl<State: Clone + Debug, Input: Clone + Debug + PartialEq> RollbackSession<State, Input> {
pub fn new(initial_state: State, neutral_inputs: Vec<Input>, config: RollbackConfig, advance: fn(&State, &[Input]) -> State) -> Self {
let cap = config.max_rollback_frames.max(1);
let players = neutral_inputs.len();
Self {
state_history: StateHistory::new(cap),
used: (0..players).map(|_| StateHistory::new(cap)).collect(),
timelines: (0..players).map(|_| InputTimeline::new(cap)).collect(),
neutral: neutral_inputs,
current_state: initial_state,
head_frame: 0,
advance,
predictor: repeat_last_input,
rollback_enabled: true,
earliest_incorrect: None,
last_rollback_len: 0,
max_rollback_len: 0,
rollback_count: 0,
}
}
pub fn with_predictor(mut self, predictor: fn(&Input, Frame) -> Input) -> Self {
self.predictor = predictor;
self
}
pub fn num_players(&self) -> usize {
self.timelines.len()
}
pub fn current_frame(&self) -> Frame {
self.head_frame
}
pub fn state(&self) -> &State {
&self.current_state
}
pub fn state_at(&self, frame: Frame) -> Option<State> {
if frame == self.head_frame {
return Some(self.current_state.clone());
}
self.state_history.restore(frame)
}
pub fn set_rollback_enabled(&mut self, enabled: bool) {
self.rollback_enabled = enabled;
}
pub fn queue_local_input(&mut self, player: usize, input: Input) {
self.timelines[player].confirm(self.head_frame, input);
}
pub fn confirm_remote_input(&mut self, player: usize, frame: Frame, input: Input) {
self.timelines[player].confirm(frame, input.clone());
if frame < self.head_frame
&& let Some(used) = self.used[player].restore(frame)
&& used != input
{
self.earliest_incorrect = Some(self.earliest_incorrect.map_or(frame, |cur| cur.min(frame)));
}
}
pub fn advance_frame(&mut self) {
self.resolve_rollback();
let f = self.head_frame;
self.simulate_frame(f);
self.head_frame += 1;
}
pub fn is_frame_confirmed(&self, frame: Frame) -> bool {
self.timelines.iter().all(|t| t.confirmed_at(frame).is_some())
}
pub fn confirmed_frame(&self, player: usize) -> Option<Frame> {
self.timelines[player].last_confirmed_frame()
}
pub fn prediction_horizon(&self) -> usize {
if self.head_frame == 0 {
return 0;
}
let last_simulated = self.head_frame - 1;
(0..self.timelines.len())
.map(|p| match self.timelines[p].last_confirmed_frame() {
Some(c) if c >= last_simulated => 0,
Some(c) => (last_simulated - c) as usize,
None => self.head_frame as usize,
})
.max()
.unwrap_or(0)
}
pub fn last_rollback_frames(&self) -> usize {
self.last_rollback_len
}
pub fn max_rollback_frames(&self) -> usize {
self.max_rollback_len
}
pub fn rollback_count(&self) -> u64 {
self.rollback_count
}
fn resolve_rollback(&mut self) {
self.last_rollback_len = 0;
let Some(mut earliest) = self.earliest_incorrect.take() else {
return;
};
if !self.rollback_enabled {
return; }
if earliest >= self.head_frame {
return; }
match self.state_history.oldest_frame() {
Some(oldest) if earliest < oldest => earliest = oldest,
None => return,
_ => {}
}
let Some(restored) = self.state_history.restore(earliest) else {
return;
};
let target = self.head_frame; self.current_state = restored;
let mut f = earliest;
while f < target {
self.simulate_frame(f);
f += 1;
}
let len = (target - earliest) as usize;
self.last_rollback_len = len;
self.max_rollback_len = self.max_rollback_len.max(len);
self.rollback_count += 1;
}
fn simulate_frame(&mut self, frame: Frame) {
self.state_history.save(frame, self.current_state.clone());
let mut inputs = Vec::with_capacity(self.timelines.len());
for p in 0..self.timelines.len() {
let input = self.input_for(p, frame);
self.used[p].save(frame, input.clone());
inputs.push(input);
}
self.current_state = (self.advance)(&self.current_state, &inputs);
}
fn input_for(&self, p: usize, frame: Frame) -> Input {
if let Some(confirmed) = self.timelines[p].confirmed_at(frame) {
return confirmed.clone();
}
let basis = self.timelines[p].last_confirmed_at_or_before(frame).cloned().unwrap_or_else(|| self.neutral[p].clone());
(self.predictor)(&basis, frame)
}
}
#[cfg(test)]
mod tests {
use super::*;
#[derive(Clone, Copy, Debug, PartialEq)]
struct World {
pos: [i64; 2],
}
#[derive(Clone, Copy, Debug, PartialEq)]
struct In(i64);
const NEUTRAL: In = In(0);
fn step(state: &World, inputs: &[In]) -> World {
let mut next = *state;
next.pos[0] += inputs[0].0;
next.pos[1] += inputs[1].0;
next
}
fn session() -> RollbackSession<World, In> {
RollbackSession::new(World { pos: [0, 0] }, vec![NEUTRAL, NEUTRAL], RollbackConfig { max_rollback_frames: 64 }, step)
}
mod the_predictor {
use super::*;
fn always_neutral(_last: &In, _frame: Frame) -> In {
NEUTRAL
}
fn rollbacks_with(predictor: Option<fn(&In, Frame) -> In>) -> u64 {
let mut s = session();
if let Some(predictor) = predictor {
s = s.with_predictor(predictor);
}
for frame in 0..20u64 {
s.queue_local_input(0, In(1));
if frame >= 2 {
s.confirm_remote_input(1, frame - 2, In(1));
}
s.advance_frame();
}
s.rollback_count()
}
#[test]
fn a_predictor_that_guesses_right_costs_fewer_rollbacks() {
let repeating = rollbacks_with(None);
let neutral = rollbacks_with(Some(always_neutral));
assert!(
neutral > repeating,
"the wrong guess re-simulates more: {neutral} against {repeating}"
);
}
#[test]
fn the_frame_counters_agree_with_what_was_driven() {
let mut s = session();
assert_eq!(s.num_players(), 2);
assert_eq!(s.current_frame(), 0);
assert_eq!(s.confirmed_frame(1), None, "nothing confirmed yet");
for frame in 0..5u64 {
s.queue_local_input(0, In(1));
s.confirm_remote_input(1, frame, In(0));
s.advance_frame();
}
assert_eq!(s.current_frame(), 5, "five frames driven, five frames on");
assert_eq!(s.confirmed_frame(1), Some(4), "and the peer is confirmed through the last");
}
}
fn ground_truth(inputs0: &[In], inputs1: &[In]) -> World {
let mut w = World { pos: [0, 0] };
for f in 0..inputs0.len() {
w = step(&w, &[inputs0[f], inputs1[f]]);
}
w
}
#[test]
fn state_history_saves_restores_and_evicts_by_frame() {
let mut h = StateHistory::new(3);
h.save(0, 10);
h.save(1, 11);
h.save(2, 12);
assert_eq!(h.restore(1), Some(11));
assert_eq!(h.oldest_frame(), Some(0));
assert_eq!(h.latest_frame(), Some(2));
h.save(3, 13); assert_eq!(h.restore(0), None);
assert_eq!(h.oldest_frame(), Some(1));
assert_eq!(h.restore(3), Some(13));
}
#[test]
fn state_history_overwrites_a_frame_in_place() {
let mut h = StateHistory::new(4);
h.save(0, 1);
h.save(1, 2);
h.save(1, 99); assert_eq!(h.restore(1), Some(99));
assert_eq!(h.latest_frame(), Some(1), "overwriting does not extend the window");
}
#[test]
fn a_non_contiguous_save_is_counted_because_it_shortens_the_reach() {
let mut h = StateHistory::new(4);
h.save(0, 1);
h.save(1, 2);
assert_eq!(h.resets(), 0);
h.save(50, 3);
assert_eq!(h.resets(), 1);
assert_eq!(h.oldest_frame(), Some(50), "and the reach is now one frame");
assert_eq!(h.restore(1), None);
}
#[test]
fn input_timeline_predicts_the_last_confirmed() {
let mut t = InputTimeline::new(8);
t.confirm(0, In(5));
t.confirm(1, In(5));
assert_eq!(t.confirmed_at(1), Some(&In(5)));
assert_eq!(t.confirmed_at(2), None, "frame 2 is unconfirmed");
assert_eq!(t.last_confirmed_at_or_before(5), Some(&In(5)));
assert_eq!(t.last_confirmed_frame(), Some(1));
}
#[test]
fn a_correct_prediction_never_rolls_back() {
let mut s = session();
let inputs = [In(2); 10];
s.confirm_remote_input(1, 0, inputs[0]);
for f in 0..10u64 {
s.queue_local_input(0, In(1));
if f >= 3 {
s.confirm_remote_input(1, f - 3, inputs[(f - 3) as usize]);
}
s.advance_frame();
}
assert_eq!(s.rollback_count(), 0, "a held remote input is predicted exactly");
assert_eq!(s.last_rollback_frames(), 0);
}
#[test]
fn a_wrong_prediction_rolls_back_and_lands_on_the_truth() {
let remote: Vec<In> = (0..10).map(|f| if f < 4 { In(1) } else { In(-3) }).collect();
let local = vec![In(1); 10];
let mut s = session();
for f in 0..10u64 {
s.queue_local_input(0, local[f as usize]);
if f >= 2 {
let past = f - 2;
s.confirm_remote_input(1, past, remote[past as usize]);
}
s.advance_frame();
}
for f in 8..10u64 {
s.confirm_remote_input(1, f, remote[f as usize]);
}
s.resolve_rollback();
assert!(s.rollback_count() > 0, "the direction change forced at least one rollback");
let truth = ground_truth(&local, &remote);
assert_eq!(*s.state(), truth, "rollback converged on the ground truth");
}
#[test]
fn two_peers_exchanging_inputs_converge_to_the_same_world() {
let p0: Vec<In> = (0..40).map(|f| In(((f * 7) % 5) as i64 - 2)).collect();
let p1: Vec<In> = (0..40).map(|f| In(((f * 3) % 4) as i64 - 1)).collect();
let mut a = session(); let mut b = session();
let delay = 2u64;
for f in 0..40u64 {
a.queue_local_input(0, p0[f as usize]);
b.queue_local_input(1, p1[f as usize]);
if f >= delay {
let past = f - delay;
a.confirm_remote_input(1, past, p1[past as usize]); b.confirm_remote_input(0, past, p0[past as usize]); }
a.advance_frame();
b.advance_frame();
}
for f in 38..40u64 {
a.confirm_remote_input(1, f, p1[f as usize]);
b.confirm_remote_input(0, f, p0[f as usize]);
}
a.resolve_rollback();
b.resolve_rollback();
let truth = ground_truth(&p0, &p1);
assert_eq!(*a.state(), truth, "peer A converged");
assert_eq!(*b.state(), truth, "peer B converged");
assert_eq!(a.state(), b.state(), "the two peers agree frame for frame");
}
#[test]
fn delay_based_policy_can_tell_when_a_frame_is_fully_known() {
let mut s = session();
s.queue_local_input(0, In(1));
assert!(!s.is_frame_confirmed(0), "the remote input for frame 0 is not in yet");
s.confirm_remote_input(1, 0, In(1));
assert!(s.is_frame_confirmed(0), "now every player's frame-0 input is known");
}
#[test]
fn a_correction_older_than_the_history_does_not_panic() {
let mut s = RollbackSession::new(World { pos: [0, 0] }, vec![NEUTRAL, NEUTRAL], RollbackConfig { max_rollback_frames: 4 }, step);
for f in 0..20u64 {
s.queue_local_input(0, In(1));
s.confirm_remote_input(1, f, In(1)); s.advance_frame();
}
s.confirm_remote_input(1, 0, In(99));
s.advance_frame(); assert!(s.state().pos[0].is_positive() || s.state().pos[0] == 0);
let _ = s.state();
}
#[test]
fn rollback_disabled_keeps_a_wrong_guess_and_diverges_from_the_truth() {
let remote: Vec<In> = (0..10).map(|f| if f < 4 { In(1) } else { In(-3) }).collect();
let local = vec![In(1); 10];
let mut s = session();
s.set_rollback_enabled(false);
for f in 0..10u64 {
s.queue_local_input(0, local[f as usize]);
if f >= 2 {
let past = f - 2;
s.confirm_remote_input(1, past, remote[past as usize]);
}
s.advance_frame();
}
for f in 8..10u64 {
s.confirm_remote_input(1, f, remote[f as usize]);
}
s.resolve_rollback();
assert_eq!(s.rollback_count(), 0, "rollback disabled never re-simulates");
let truth = ground_truth(&local, &remote);
assert_ne!(*s.state(), truth, "without rollback the trusted guess never converges");
}
#[test]
fn state_at_a_confirmed_frame_matches_across_two_peers() {
let p0: Vec<In> = (0..30).map(|f| In(((f * 5) % 3) as i64 - 1)).collect();
let p1: Vec<In> = (0..30).map(|f| In(((f * 2) % 3) as i64 - 1)).collect();
let mut a = session();
let mut b = session();
for f in 0..30u64 {
a.queue_local_input(0, p0[f as usize]);
b.queue_local_input(1, p1[f as usize]);
if f >= 3 {
a.confirm_remote_input(1, f - 3, p1[(f - 3) as usize]);
b.confirm_remote_input(0, f - 3, p0[(f - 3) as usize]);
}
a.advance_frame();
b.advance_frame();
}
let cf = 20;
let sa = a.state_at(cf).expect("A retains the frame");
let sb = b.state_at(cf).expect("B retains the frame");
assert_eq!(sa, sb, "a fully-confirmed frame is identical on both peers");
}
#[test]
fn the_prediction_horizon_reflects_how_far_ahead_of_confirmation_it_is() {
let mut s = session();
for _ in 0..6u64 {
s.queue_local_input(0, In(1));
s.advance_frame();
}
assert_eq!(s.prediction_horizon(), 6);
s.confirm_remote_input(1, 4, In(0));
assert_eq!(s.prediction_horizon(), 1, "confirmed through frame 4, one frame ahead");
}
}