Struct ReplicationSource 

pub struct ReplicationSource { /* private fields */ }

Bounded backlog of recent replicated mutations.

use kevy_replicate::source::ReplicationSource;

let set = kevy_resp::Argv::from(vec![b"SET".to_vec(), b"k".to_vec(), b"v".to_vec()]);
let mut backlog = ReplicationSource::new(1 << 20);
assert_eq!(backlog.push_mutation(&set), 0);
assert_eq!(backlog.push_mutation(&set), 1);
assert_eq!((backlog.oldest_offset(), backlog.newest_offset()), (Some(0), Some(1)));

Implementations

impl ReplicationSource

pub fn new(max_bytes: usize) -> Self

Create a new source with the given byte budget. max_bytes must be > 0; the source guarantees at most one over-budget frame at a time (the most recently pushed) so a single huge command does not silently disappear before its replicas even see it.

Panics

Panics if max_bytes == 0 — a zero budget could never hold even one frame, so it is a caller bug, not a runtime state.

use kevy_replicate::source::ReplicationSource;

let set = kevy_resp::Argv::from(vec![b"SET".to_vec(), b"k".to_vec(), b"v".to_vec()]);
let mut backlog = ReplicationSource::new(1); // smaller than any frame
backlog.push_mutation(&set);
assert_eq!(backlog.len(), 1); // the newest frame is kept anyway
assert!(backlog.buffered_bytes() > backlog.max_bytes());

pub fn set_next_offset(&mut self, next: u64)

Resume offset assignment at next (boot continuity from the feed sidecar). Only meaningful on an empty, freshly created source — asserts the backlog has no frames.

Panics

Panics if the backlog already holds frames: renumbering live frames would corrupt every replica’s ack bookkeeping.

use kevy_replicate::source::ReplicationSource;

let set = kevy_resp::Argv::from(vec![b"SET".to_vec(), b"k".to_vec(), b"v".to_vec()]);
let mut backlog = ReplicationSource::new(1 << 20);
backlog.set_next_offset(1_000); // offsets continue from the last session
assert_eq!(backlog.push_mutation(&set), 1_000);

pub fn max_bytes(&self) -> usize

The byte budget this source was created with.

use kevy_replicate::source::ReplicationSource;


assert_eq!(ReplicationSource::new(64 << 20).max_bytes(), 64 << 20);

pub fn next_offset(&self) -> u64

Next offset this source would assign. Equal to one past the last assigned offset; equals 0 for a fresh source.

use kevy_replicate::source::ReplicationSource;

let set = kevy_resp::Argv::from(vec![b"SET".to_vec(), b"k".to_vec(), b"v".to_vec()]);
let mut backlog = ReplicationSource::new(1 << 20);
assert_eq!(backlog.next_offset(), 0);
backlog.push_mutation(&set);
assert_eq!(backlog.next_offset(), 1);

pub fn oldest_offset(&self) -> Option<u64>

Lowest offset still in the backlog, or None if empty.

use kevy_replicate::source::ReplicationSource;

let set = kevy_resp::Argv::from(vec![b"SET".to_vec(), b"k".to_vec(), b"v".to_vec()]);
let mut backlog = ReplicationSource::new(1 << 20);
assert_eq!(backlog.oldest_offset(), None);
for _ in 0..3 {
    backlog.push_mutation(&set);
}
backlog.drop_up_to(2);
assert_eq!(backlog.oldest_offset(), Some(2));

pub fn newest_offset(&self) -> Option<u64>

Highest offset still in the backlog, or None if empty.

use kevy_replicate::source::ReplicationSource;

let set = kevy_resp::Argv::from(vec![b"SET".to_vec(), b"k".to_vec(), b"v".to_vec()]);
let mut backlog = ReplicationSource::new(1 << 20);
assert_eq!(backlog.newest_offset(), None);
backlog.push_mutation(&set);
backlog.push_mutation(&set);
assert_eq!(backlog.newest_offset(), Some(1));

pub fn buffered_bytes(&self) -> usize

Total bytes occupied by frames currently in the backlog.

use kevy_replicate::source::ReplicationSource;

let set = kevy_resp::Argv::from(vec![b"SET".to_vec(), b"k".to_vec(), b"v".to_vec()]);
let mut backlog = ReplicationSource::new(1 << 20);
backlog.push_mutation(&set);
assert_eq!(backlog.buffered_bytes(), kevy_replicate::wire::encode_frame(0, &set).len());

pub fn len(&self) -> usize

Number of frames currently in the backlog.

use kevy_replicate::source::ReplicationSource;

let set = kevy_resp::Argv::from(vec![b"SET".to_vec(), b"k".to_vec(), b"v".to_vec()]);
let mut backlog = ReplicationSource::new(1 << 20);
backlog.push_mutation(&set);
backlog.push_mutation(&set);
assert_eq!(backlog.len(), 2);

pub fn is_empty(&self) -> bool

Whether the backlog has no frames.

use kevy_replicate::source::ReplicationSource;

let set = kevy_resp::Argv::from(vec![b"SET".to_vec(), b"k".to_vec(), b"v".to_vec()]);
let mut backlog = ReplicationSource::new(1 << 20);
assert!(backlog.is_empty());
backlog.push_mutation(&set);
assert!(!backlog.is_empty());

pub fn push_mutation<A: ArgvView + ?Sized>(&mut self, argv: &A) -> u64

Append one applied mutation. Returns the offset assigned to it. Generic over ArgvView so the dispatcher’s borrowed argv can flow straight in — no Argv materialisation on the write path.

May evict older frames if the new frame would exceed the byte budget; the new frame is always retained (even if it is larger than max_bytes on its own — losing the most recent applied write before any replica has had a chance to ack it would be a worse failure than briefly running over budget).

use kevy_replicate::source::ReplicationSource;


// the reactor's borrowed argv goes straight in
let request = b"*3\r\n$3\r\nSET\r\n$1\r\nk\r\n$1\r\nv\r\n";
let (argv, _) = kevy_resp::parse_command_borrowed(request)?.expect("a whole command");
let mut backlog = ReplicationSource::new(1 << 20);
assert_eq!(backlog.push_mutation(&argv), 0);
assert_eq!(backlog.len(), 1);

pub fn drop_up_to(&mut self, watermark: u64)

Drop every buffered frame whose offset is < watermark — i.e. every replica has consumed past it. Used by the per- shard tick to enforce a retention floor tighter than the raw byte budget; lets the backlog reclaim space for live frames once all consumers have advanced.

No-op when watermark <= oldest_offset() (nothing to drop) or when the buffer is empty. Updates the internal byte accounting to stay consistent with the live buffer length.

use kevy_replicate::source::ReplicationSource;

let set = kevy_resp::Argv::from(vec![b"SET".to_vec(), b"k".to_vec(), b"v".to_vec()]);
let mut backlog = ReplicationSource::new(1 << 20);
for _ in 0..5 {
    backlog.push_mutation(&set);
}
backlog.drop_up_to(3); // every replica has acked past offset 2
assert_eq!((backlog.len(), backlog.oldest_offset()), (2, Some(3)));

pub fn frames_from(&self, from: u64) -> Result<FramesIter<'_>, FromOffset>

Borrow the slice of frames with offset ≥ from. Suitable for the streaming loop to write each frame’s bytes to a replica socket. Returns:

• Ok(iter) — zero or more frames in offset order (empty iter means the replica is caught up).
• Err(FromOffset::TooOld) — from is older than the oldest buffered frame; the streaming loop must snapshot-ship.
• Err(FromOffset::Future) — from > next_offset(); peer is ahead of us, drop the link.

use kevy_replicate::source::ReplicationSource;

let set = kevy_resp::Argv::from(vec![b"SET".to_vec(), b"k".to_vec(), b"v".to_vec()]);
let mut backlog = ReplicationSource::new(1 << 20);
for _ in 0..4 {
    backlog.push_mutation(&set);
}
let offsets: Vec<u64> = backlog.frames_from(2)?.map(|f| f.offset).collect();
assert_eq!(offsets, [2, 3]);
assert_eq!(backlog.frames_from(4)?.count(), 0); // caught up

Trait Implementations

impl Debug for ReplicationSource

fn fmt(&self, f: &mut Formatter<'_>) -> Result

Formats the value using the given formatter.