NASP/rpki
84
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

优化rtr

Browse Source
This commit is contained in:
xiuting.xu
2026-04-08 14:03:23 +08:00
parent b483c2af8b
commit 23cdad095d
16 changed files with 1835 additions and 1606 deletions
Download Patch File Download Diff File Expand all files Collapse all files

290
src/bin/ccr_fixture_gen.rs Normal file
View File

@@ -0,0 +1,290 @@
use std::fs;
use std::net::{IpAddr, Ipv4Addr, Ipv6Addr};
use std::path::{Path, PathBuf};
use anyhow::{Context, Result};
use sha2::{Digest, Sha256};
const CONTENT_TYPE_OID: &str = "1.2.840.113549.1.9.16.1.54";
const SHA256_OID: &str = "2.16.840.1.101.3.4.2.1";
#[derive(Clone)]
struct Vrp {
addr: IpAddr,
prefix_len: u8,
max_len: u8,
asn: u32,
}
#[derive(Clone)]
struct Vap {
customer_asn: u32,
providers: Vec<u32>,
}
fn main() -> Result<()> {
let out_dir = parse_out_dir_arg();
fs::create_dir_all(&out_dir)
.with_context(|| format!("failed to create output dir {}", out_dir.display()))?;
write_snapshot(
&out_dir.join("20260403T000001Z-mini-a.ccr"),
"20260403000001Z",
vec![
Vrp {
addr: IpAddr::V4(Ipv4Addr::new(10, 0, 0, 0)),
prefix_len: 24,
max_len: 24,
asn: 65001,
},
Vrp {
addr: IpAddr::V6(Ipv6Addr::new(0x2001, 0xdb8, 1, 0, 0, 0, 0, 0)),
prefix_len: 48,
max_len: 48,
asn: 65002,
},
],
vec![Vap {
customer_asn: 65010,
providers: vec![65011, 65012],
}],
)?;
write_snapshot(
&out_dir.join("20260403T000101Z-mini-b.ccr"),
"20260403000101Z",
vec![
Vrp {
addr: IpAddr::V4(Ipv4Addr::new(10, 0, 0, 0)),
prefix_len: 24,
max_len: 24,
asn: 65001,
},
Vrp {
addr: IpAddr::V4(Ipv4Addr::new(10, 0, 1, 0)),
prefix_len: 24,
max_len: 24,
asn: 65003,
},
Vrp {
addr: IpAddr::V6(Ipv6Addr::new(0x2001, 0xdb8, 1, 0, 0, 0, 0, 0)),
prefix_len: 48,
max_len: 48,
asn: 65002,
},
],
vec![Vap {
customer_asn: 65010,
providers: vec![65011, 65012, 65013],
}],
)?;
write_snapshot(
&out_dir.join("20260403T000201Z-mini-c.ccr"),
"20260403000201Z",
vec![
Vrp {
addr: IpAddr::V4(Ipv4Addr::new(10, 0, 1, 0)),
prefix_len: 24,
max_len: 24,
asn: 65003,
},
Vrp {
addr: IpAddr::V6(Ipv6Addr::new(0x2001, 0xdb8, 2, 0, 0, 0, 0, 0)),
prefix_len: 48,
max_len: 48,
asn: 65004,
},
],
vec![
Vap {
customer_asn: 65010,
providers: vec![65012, 65013],
},
Vap {
customer_asn: 65020,
providers: vec![65021],
},
],
)?;
println!("generated CCR fixtures under {}", out_dir.display());
Ok(())
}
fn parse_out_dir_arg() -> PathBuf {
let mut args = std::env::args().skip(1);
let mut out_dir = PathBuf::from("data");
while let Some(arg) = args.next() {
if arg == "--out-dir" {
if let Some(v) = args.next() {
out_dir = PathBuf::from(v);
}
}
}
out_dir
}
fn write_snapshot(path: &Path, produced_at: &str, vrps: Vec<Vrp>, vaps: Vec<Vap>) -> Result<()> {
let bytes = encode_ccr_snapshot(produced_at, vrps, vaps);
fs::write(path, bytes).with_context(|| format!("failed to write {}", path.display()))?;
println!("wrote {}", path.display());
Ok(())
}
fn encode_ccr_snapshot(produced_at: &str, vrps: Vec<Vrp>, vaps: Vec<Vap>) -> Vec<u8> {
let vrp_sets = vrps
.into_iter()
.map(encode_roa_payload_set)
.collect::<Vec<_>>();
let vap_sets = vaps
.into_iter()
.map(encode_aspa_payload_set)
.collect::<Vec<_>>();
let vrp_set_seq = der_sequence(vrp_sets);
let vap_set_seq = der_sequence(vap_sets);
let vrp_hash = Sha256::digest(&vrp_set_seq).to_vec();
let vap_hash = Sha256::digest(&vap_set_seq).to_vec();
// draft-ietf-sidrops-rpki-ccr-02:
// ROAPayloadState/ASPAPayloadState include payload-set sequence + hash.
let vrp_state = der_sequence(vec![vrp_set_seq, der_octet_string(vrp_hash)]);
let vap_state = der_sequence(vec![vap_set_seq, der_octet_string(vap_hash)]);
// AlgorithmIdentifier for SHA-256.
let hash_alg = der_sequence(vec![der_oid(SHA256_OID), der_null()]);
let payload = der_sequence(vec![
der_integer(0),
hash_alg,
der_generalized_time(produced_at),
der_ctx(2, vrp_state),
der_ctx(3, vap_state),
]);
der_sequence(vec![der_oid(CONTENT_TYPE_OID), der_ctx(0, payload)])
}
fn encode_roa_payload_set(v: Vrp) -> Vec<u8> {
let (afi, addr_bytes) = match v.addr {
IpAddr::V4(ip) => ([0u8, 1u8].to_vec(), ip.octets().to_vec()),
IpAddr::V6(ip) => ([0u8, 2u8].to_vec(), ip.octets().to_vec()),
};
let bit_string = prefix_to_bit_string(&addr_bytes, v.prefix_len);
let roa_ip = der_sequence(vec![
der_bit_string(0, bit_string),
der_integer(u32::from(v.max_len)),
]);
let family = der_sequence(vec![der_octet_string(afi), der_sequence(vec![roa_ip])]);
der_sequence(vec![der_integer(v.asn), der_sequence(vec![family])])
}
fn encode_aspa_payload_set(v: Vap) -> Vec<u8> {
let providers = v.providers.into_iter().map(der_integer).collect::<Vec<_>>();
der_sequence(vec![der_integer(v.customer_asn), der_sequence(providers)])
}
fn prefix_to_bit_string(addr: &[u8], prefix_len: u8) -> Vec<u8> {
let byte_len = usize::from(prefix_len).div_ceil(8);
let mut out = addr[..byte_len].to_vec();
let rem = prefix_len % 8;
if rem != 0 {
let mask = 0xFFu8 << (8 - rem);
let last = out.len() - 1;
out[last] &= mask;
}
out
}
fn der_sequence(items: Vec<Vec<u8>>) -> Vec<u8> {
let content = items.concat();
der_tlv(0x30, content)
}
fn der_integer(v: u32) -> Vec<u8> {
if v == 0 {
return der_tlv(0x02, vec![0]);
}
let mut bytes = v.to_be_bytes().to_vec();
while bytes.len() > 1 && bytes[0] == 0 {
bytes.remove(0);
}
if bytes[0] & 0x80 != 0 {
bytes.insert(0, 0);
}
der_tlv(0x02, bytes)
}
fn der_oid(oid: &str) -> Vec<u8> {
let parts = oid
.split('.')
.map(|s| s.parse::<u32>().unwrap())
.collect::<Vec<_>>();
assert!(parts.len() >= 2);
let mut out = Vec::new();
out.push((parts[0] * 40 + parts[1]) as u8);
for &part in &parts[2..] {
out.extend(base128(part));
}
der_tlv(0x06, out)
}
fn base128(mut n: u32) -> Vec<u8> {
let mut buf = vec![(n & 0x7F) as u8];
n >>= 7;
while n > 0 {
buf.push(((n & 0x7F) as u8) | 0x80);
n >>= 7;
}
buf.reverse();
buf
}
fn der_octet_string(bytes: Vec<u8>) -> Vec<u8> {
der_tlv(0x04, bytes)
}
fn der_null() -> Vec<u8> {
der_tlv(0x05, Vec::new())
}
fn der_bit_string(unused_bits: u8, bytes: Vec<u8>) -> Vec<u8> {
let mut content = Vec::with_capacity(1 + bytes.len());
content.push(unused_bits);
content.extend(bytes);
der_tlv(0x03, content)
}
fn der_generalized_time(v: &str) -> Vec<u8> {
der_tlv(0x18, v.as_bytes().to_vec())
}
fn der_ctx(tag_no: u8, encoded_inner_der: Vec<u8>) -> Vec<u8> {
der_tlv(0xA0 + tag_no, encoded_inner_der)
}
fn der_tlv(tag: u8, content: Vec<u8>) -> Vec<u8> {
let mut out = Vec::with_capacity(2 + content.len());
out.push(tag);
out.extend(der_len(content.len()));
out.extend(content);
out
}
fn der_len(len: usize) -> Vec<u8> {
if len < 128 {
return vec![len as u8];
}
let mut bytes = Vec::new();
let mut n = len;
while n > 0 {
bytes.push((n & 0xFF) as u8);
n >>= 8;
}
bytes.reverse();
let mut out = vec![0x80 | (bytes.len() as u8)];
out.extend(bytes);
out
}

128
src/bin/rtr_debug_client/main.rs
View File

@@ -94,7 +94,7 @@ async fn main() -> io::Result<()> {
Ok(Some(line)) => {
match handle_console_command(
&line,
&mut writer,
Some(&mut writer),
&mut state,
).await {
Ok(should_quit) => {
@@ -169,7 +169,47 @@ async fn main() -> io::Result<()> {
let delay = state.reconnect_delay_secs();
state.current_session_id = None;
println!("[reconnect] transport disconnected, retry after {}s", delay);
tokio::time::sleep(Duration::from_secs(delay)).await;
let reconnect_sleep = tokio::time::sleep(Duration::from_secs(delay));
tokio::pin!(reconnect_sleep);
let mut reconnect_now = false;
loop {
tokio::select! {
_ = &mut reconnect_sleep => break,
line = stdin_lines.next_line() => {
match line {
Ok(Some(line)) => {
match handle_console_command(&line, None, &mut state).await {
Ok(should_quit) => {
if should_quit {
println!("quit requested, closing client.");
return Ok(());
}
}
Ok(false) => {
if state.take_reconnect_now() {
reconnect_now = true;
break;
}
}
Err(err) => return Err(err),
}
}
Ok(None) => {
println!("stdin closed, continue reconnect loop.");
}
Err(err) => {
eprintln!("read stdin failed: {}", err);
}
}
}
}
}
if reconnect_now {
println!("[reconnect] user requested immediate reconnect");
}
}
}
}
@@ -179,6 +219,16 @@ async fn send_resume_query(
state: &mut ClientState,
mode: &QueryMode,
) -> io::Result<()> {
if state.force_reset_on_reconnect {
state.force_reset_on_reconnect = false;
state.session_id = None;
state.serial = None;
state.current_session_id = None;
send_reset_query(writer, state.version).await?;
println!("reconnected, send Reset Query (forced)");
return Ok(());
}
match (state.session_id, state.serial) {
(Some(session_id), Some(serial)) => {
println!(
@@ -380,7 +430,7 @@ async fn handle_poll_tick(writer: &mut ClientWriter, state: &mut ClientState) ->
async fn handle_console_command(
line: &str,
writer: &mut ClientWriter,
mut writer: Option<&mut ClientWriter>,
state: &mut ClientState,
) -> io::Result<bool> {
let line = line.trim();
@@ -400,10 +450,35 @@ async fn handle_console_command(
print_state(state);
}
["version"] => {
println!("current RTR version: {}", state.version);
}
["version", version] => {
let version = match version.parse::<u8>() {
Ok(v) => v,
Err(err) => {
println!("invalid version: {}", err);
return Ok(false);
}
};
state.version = version;
println!("updated RTR version to {}", state.version);
}
["reset"] => {
println!("manual command: send Reset Query");
send_reset_query(writer, state.version).await?;
state.schedule_next_poll();
if let Some(writer) = writer.as_mut() {
send_reset_query(writer, state.version).await?;
state.schedule_next_poll();
} else {
state.force_reset_on_reconnect = true;
state.request_reconnect_now();
state.session_id = None;
state.serial = None;
state.current_session_id = None;
println!("not connected, queued Reset Query for next reconnect");
}
}
["serial"] => match (state.session_id, state.serial) {
@@ -412,8 +487,12 @@ async fn handle_console_command(
"manual command: send Serial Query with current state: session_id={}, serial={}",
session_id, serial
);
send_serial_query(writer, state.version, session_id, serial).await?;
state.schedule_next_poll();
if let Some(writer) = writer.as_mut() {
send_serial_query(writer, state.version, session_id, serial).await?;
state.schedule_next_poll();
} else {
println!("not connected, will send Serial Query on reconnect");
}
}
_ => {
println!(
@@ -445,8 +524,13 @@ async fn handle_console_command(
);
state.session_id = Some(session_id);
state.serial = Some(serial);
send_serial_query(writer, state.version, session_id, serial).await?;
state.schedule_next_poll();
if let Some(writer) = writer.as_mut() {
send_serial_query(writer, state.version, session_id, serial).await?;
state.schedule_next_poll();
} else {
state.force_reset_on_reconnect = false;
println!("not connected, queued Serial Query for next reconnect");
}
}
["timeout"] => {
@@ -533,6 +617,8 @@ fn print_help() {
println!("available commands:");
println!(" help show this help");
println!(" state print current client state");
println!(" version show current RTR version");
println!(" version <n> update RTR version");
println!(" reset send Reset Query");
println!(" serial send Serial Query with current session_id/serial");
println!(" serial <sid> <serial> send Serial Query with explicit values");
@@ -583,6 +669,8 @@ struct ClientState {
default_poll_secs: u64,
next_poll_deadline: Instant,
poll_paused: bool,
force_reset_on_reconnect: bool,
reconnect_now: bool,
}
impl ClientState {
@@ -606,6 +694,8 @@ impl ClientState {
default_poll_secs,
next_poll_deadline: Instant::now() + Duration::from_secs(default_poll_secs),
poll_paused: false,
force_reset_on_reconnect: false,
reconnect_now: false,
}
}
@@ -667,6 +757,19 @@ impl ClientState {
self.default_poll_secs
}
}
fn request_reconnect_now(&mut self) {
self.reconnect_now = true;
}
fn take_reconnect_now(&mut self) -> bool {
if self.reconnect_now {
self.reconnect_now = false;
true
} else {
false
}
}
}
#[derive(Debug)]
@@ -767,12 +870,7 @@ impl Config {
.transpose()?
.unwrap_or(1);
if version > 2 {
return Err(io::Error::new(
io::ErrorKind::InvalidInput,
format!("unsupported RTR version {}, expected 0..=2", version),
));
}
// Allow any version here; server will validate and respond.
let mode = match positional.next().as_deref() {
None | Some("reset") => QueryMode::Reset,

20
src/main.rs
View File

@@ -68,6 +68,7 @@ impl AppConfig {
fn from_env() -> Result<Self> {
let mut config = Self::default();
// TLS and TCP
if let Some(value) = env_var("RPKI_RTR_ENABLE_TLS")? {
config.enable_tls = parse_bool(&value, "RPKI_RTR_ENABLE_TLS")?;
}
@@ -81,6 +82,8 @@ impl AppConfig {
.parse()
.map_err(|err| anyhow!("invalid RPKI_RTR_TLS_ADDR '{}': {}", value, err))?;
}
// data
if let Some(value) = env_var("RPKI_RTR_DB_PATH")? {
config.db_path = value;
}
@@ -105,9 +108,16 @@ impl AppConfig {
config.tls_client_ca_path = value;
}
if let Some(value) = env_var("RPKI_RTR_MAX_DELTA")? {
config.max_delta = value
let parsed: u8 = value
.parse()
.map_err(|err| anyhow!("invalid RPKI_RTR_MAX_DELTA '{}': {}", value, err))?;
if parsed == 0 {
return Err(anyhow!(
"invalid RPKI_RTR_MAX_DELTA '{}': must be >= 1",
value
));
}
config.max_delta = parsed;
}
if let Some(value) = env_var("RPKI_RTR_PRUNE_DELTA_BY_SNAPSHOT_SIZE")? {
config.prune_delta_by_snapshot_size =
@@ -214,9 +224,9 @@ fn init_shared_cache(config: &AppConfig, store: &RtrStore) -> Result<SharedRtrCa
.map_err(|_| anyhow!("cache read lock poisoned during startup"))?;
info!(
"cache initialized: session_ids={:?}, serial={}",
"cache initialized: session_ids={:?}, serials={:?}",
cache.session_ids(),
cache.serial()
cache.serials()
);
}
@@ -270,11 +280,11 @@ fn spawn_refresh_task(
}
};
let old_serial = cache.serial();
let old_serial = cache.serial_for_version(2);
match cache.update(payloads, &store) {
Ok(()) => {
let new_serial = cache.serial();
let new_serial = cache.serial_for_version(2);
if new_serial != old_serial {
info!(
"RTR cache refresh applied: ccr_dir={}, payload_count={}, old_serial={}, new_serial={}",

609
src/rtr/cache/core.rs vendored
View File

@@ -11,6 +11,7 @@ use super::model::{Delta, DualTime, Snapshot};
use super::ordering::{ChangeKey, change_key};
const SERIAL_HALF_RANGE: u32 = 1 << 31;
const VERSION_COUNT: usize = 3;
#[derive(Debug, Clone, Copy, Serialize, Deserialize, Eq, PartialEq)]
pub enum CacheAvailability {
@@ -20,16 +21,16 @@ pub enum CacheAvailability {
#[derive(Debug, Clone, Serialize, Deserialize, Eq, PartialEq)]
pub struct SessionIds {
ids: [u16; 3],
ids: [u16; VERSION_COUNT],
}
impl SessionIds {
pub fn from_array(ids: [u16; 3]) -> Self {
pub fn from_array(ids: [u16; VERSION_COUNT]) -> Self {
Self { ids }
}
pub fn random_distinct() -> Self {
let mut ids = [0u16; 3];
let mut ids = [0u16; VERSION_COUNT];
for idx in 0..ids.len() {
loop {
let candidate: u16 = rand::random();
@@ -45,15 +46,36 @@ impl SessionIds {
pub fn get(&self, version: u8) -> u16 {
self.ids[version_index(version)]
}
pub fn as_array(&self) -> [u16; VERSION_COUNT] {
self.ids
}
}
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct VersionState {
session_id: u16,
serial: u32,
snapshot: Snapshot,
#[serde(skip)]
deltas: VecDeque<Arc<Delta>>,
}
impl VersionState {
fn new(session_id: u16, serial: u32, snapshot: Snapshot, max_delta: u8) -> Self {
Self {
session_id,
serial,
snapshot,
deltas: VecDeque::with_capacity(max_delta as usize),
}
}
}
#[derive(Debug)]
pub struct RtrCache {
availability: CacheAvailability,
session_ids: SessionIds,
serial: u32,
snapshot: Snapshot,
deltas: VecDeque<Arc<Delta>>,
versions: [VersionState; VERSION_COUNT],
max_delta: u8,
prune_delta_by_snapshot_size: bool,
timing: Timing,
@@ -65,12 +87,13 @@ pub struct RtrCache {
impl Default for RtrCache {
fn default() -> Self {
let now = DualTime::now();
let session_ids = SessionIds::random_distinct();
let versions = std::array::from_fn(|idx| {
VersionState::new(session_ids.as_array()[idx], 0, Snapshot::empty(), 100)
});
Self {
availability: CacheAvailability::Ready,
session_ids: SessionIds::random_distinct(),
serial: 0,
snapshot: Snapshot::empty(),
deltas: VecDeque::with_capacity(100),
versions,
max_delta: 100,
prune_delta_by_snapshot_size: false,
timing: Timing::default(),
@@ -87,9 +110,9 @@ pub struct RtrCacheBuilder {
max_delta: Option<u8>,
prune_delta_by_snapshot_size: Option<bool>,
timing: Option<Timing>,
serial: Option<u32>,
snapshot: Option<Snapshot>,
deltas: Option<VecDeque<Arc<Delta>>>,
serials: Option<[u32; VERSION_COUNT]>,
snapshots: Option<[Snapshot; VERSION_COUNT]>,
deltas: Option<[VecDeque<Arc<Delta>>; VERSION_COUNT]>,
created_at: Option<DualTime>,
}
@@ -101,8 +124,8 @@ impl RtrCacheBuilder {
max_delta: None,
prune_delta_by_snapshot_size: None,
timing: None,
serial: None,
snapshot: None,
serials: None,
snapshots: None,
deltas: None,
created_at: None,
}
@@ -133,17 +156,17 @@ impl RtrCacheBuilder {
self
}
pub fn serial(mut self, v: u32) -> Self {
self.serial = Some(v);
pub fn serials(mut self, v: [u32; VERSION_COUNT]) -> Self {
self.serials = Some(v);
self
}
pub fn snapshot(mut self, v: Snapshot) -> Self {
self.snapshot = Some(v);
pub fn snapshots(mut self, v: [Snapshot; VERSION_COUNT]) -> Self {
self.snapshots = Some(v);
self
}
pub fn deltas(mut self, v: VecDeque<Arc<Delta>>) -> Self {
pub fn deltas_by_version(mut self, v: [VecDeque<Arc<Delta>>; VERSION_COUNT]) -> Self {
self.deltas = Some(v);
self
}
@@ -158,22 +181,28 @@ impl RtrCacheBuilder {
let max_delta = self.max_delta.unwrap_or(100);
let prune_delta_by_snapshot_size = self.prune_delta_by_snapshot_size.unwrap_or(false);
let timing = self.timing.unwrap_or_default();
let snapshot = self.snapshot.unwrap_or_else(Snapshot::empty);
let deltas = self
.deltas
.unwrap_or_else(|| VecDeque::with_capacity(max_delta.into()));
let session_ids = self.session_ids.unwrap_or_else(SessionIds::random_distinct);
let serials = self.serials.unwrap_or([0; VERSION_COUNT]);
let snapshots = self
.snapshots
.unwrap_or_else(|| std::array::from_fn(|_| Snapshot::empty()));
let deltas = self.deltas.unwrap_or_else(|| {
std::array::from_fn(|_| VecDeque::with_capacity(max_delta as usize))
});
let versions = std::array::from_fn(|idx| VersionState {
session_id: session_ids.as_array()[idx],
serial: serials[idx],
snapshot: snapshots[idx].clone(),
deltas: deltas[idx].clone(),
});
let serial = self.serial.unwrap_or(0);
let created_at = self.created_at.unwrap_or_else(|| now.clone());
let availability = self.availability.unwrap_or(CacheAvailability::Ready);
let session_ids = self.session_ids.unwrap_or_else(SessionIds::random_distinct);
RtrCache {
availability,
session_ids,
serial,
snapshot,
deltas,
versions,
max_delta,
prune_delta_by_snapshot_size,
timing,
@@ -187,72 +216,73 @@ impl RtrCacheBuilder {
impl RtrCache {
fn set_unavailable(&mut self) {
warn!(
"RTR cache entering NoDataAvailable: old_serial={}, snapshot_empty={}, delta_count={}",
self.serial,
self.snapshot.is_empty(),
self.deltas.len()
"RTR cache entering NoDataAvailable: serials={:?}",
self.serials()
);
self.availability = CacheAvailability::NoDataAvailable;
self.snapshot = Snapshot::empty();
self.deltas.clear();
for version_state in &mut self.versions {
version_state.snapshot = Snapshot::empty();
version_state.deltas.clear();
}
}
fn reinitialize_from_snapshot(&mut self, snapshot: Snapshot) -> AppliedUpdate {
let old_serial = self.serial;
let old_session_ids = self.session_ids.clone();
fn reinitialize_from_snapshot(&mut self, source_snapshot: &Snapshot) -> AppliedUpdate {
let old_serials = self.serials();
let old_session_ids = self.session_ids();
let new_session_ids = SessionIds::random_distinct();
self.availability = CacheAvailability::Ready;
self.session_ids = SessionIds::random_distinct();
self.serial = 1;
self.snapshot = snapshot.clone();
self.deltas.clear();
for version in 0..VERSION_COUNT {
let v = version as u8;
let state = &mut self.versions[version];
state.session_id = new_session_ids.get(v);
state.serial = 1;
state.snapshot = project_snapshot_for_version(source_snapshot, v);
state.deltas.clear();
}
self.last_update_end = DualTime::now();
info!(
"RTR cache reinitialized from usable snapshot: old_serial={}, new_serial={}, old_session_ids={:?}, new_session_ids={:?}, payloads(route_origins={}, router_keys={}, aspas={})",
old_serial,
self.serial,
"RTR cache reinitialized from usable snapshot: old_serials={:?}, new_serials={:?}, old_session_ids={:?}, new_session_ids={:?}",
old_serials,
self.serials(),
old_session_ids,
self.session_ids,
snapshot.origins().len(),
snapshot.router_keys().len(),
snapshot.aspas().len()
new_session_ids
);
AppliedUpdate {
availability: self.availability,
snapshot,
serial: self.serial,
session_ids: self.session_ids.clone(),
delta: None,
delta_window: None,
clear_delta_window: true,
}
self.applied_update_with_clear()
}
fn next_serial(&mut self) -> u32 {
let old = self.serial;
self.serial = self.serial.wrapping_add(1);
fn next_serial(state: &mut VersionState) -> u32 {
let old = state.serial;
state.serial = state.serial.wrapping_add(1);
debug!(
"RTR cache advanced serial: old_serial={}, new_serial={}",
old, self.serial
"RTR cache advanced serial for version state: old_serial={}, new_serial={}",
old, state.serial
);
self.serial
state.serial
}
fn push_delta(&mut self, delta: Arc<Delta>) {
if self.deltas.len() >= self.max_delta as usize {
self.deltas.pop_front();
fn push_delta(
state: &mut VersionState,
max_delta: u8,
prune_delta_by_snapshot_size: bool,
delta: Arc<Delta>,
) {
let max_keep = usize::from(max_delta.max(1));
while state.deltas.len() >= max_keep {
state.deltas.pop_front();
}
self.deltas.push_back(delta);
state.deltas.push_back(delta);
let mut dropped_serials = Vec::new();
if self.prune_delta_by_snapshot_size {
let snapshot_wire_size = estimate_snapshot_payload_wire_size(&self.snapshot);
if prune_delta_by_snapshot_size {
let snapshot_wire_size = estimate_snapshot_payload_wire_size(&state.snapshot);
let mut cumulative_delta_wire_size =
estimate_delta_window_payload_wire_size(&self.deltas);
while !self.deltas.is_empty() && cumulative_delta_wire_size >= snapshot_wire_size {
if let Some(oldest) = self.deltas.pop_front() {
estimate_delta_window_payload_wire_size(&state.deltas);
while !state.deltas.is_empty() && cumulative_delta_wire_size >= snapshot_wire_size {
if let Some(oldest) = state.deltas.pop_front() {
dropped_serials.push(oldest.serial());
cumulative_delta_wire_size =
estimate_delta_window_payload_wire_size(&self.deltas);
estimate_delta_window_payload_wire_size(&state.deltas);
}
}
debug!(
@@ -260,25 +290,11 @@ impl RtrCache {
snapshot_wire_size, cumulative_delta_wire_size, dropped_serials
);
}
debug!(
"RTR cache pushing delta into window: delta_serial={}, announced={}, withdrawn={}, dropped_oldest_serials={:?}, window_size_after={}, max_delta={}, prune_delta_by_snapshot_size={}",
self.deltas.back().map(|d| d.serial()).unwrap_or(0),
self.deltas.back().map(|d| d.announced().len()).unwrap_or(0),
self.deltas.back().map(|d| d.withdrawn().len()).unwrap_or(0),
dropped_serials,
self.deltas.len(),
self.max_delta,
self.prune_delta_by_snapshot_size
);
}
fn replace_snapshot(&mut self, snapshot: Snapshot) {
self.snapshot = snapshot;
}
fn delta_window(&self) -> Option<(u32, u32)> {
let min = self.deltas.front().map(|d| d.serial());
let max = self.deltas.back().map(|d| d.serial());
fn delta_window(state: &VersionState) -> Option<(u32, u32)> {
let min = state.deltas.front().map(|d| d.serial());
let max = state.deltas.back().map(|d| d.serial());
match (min, max) {
(Some(min), Some(max)) => Some((min, max)),
_ => None,
@@ -291,116 +307,104 @@ impl RtrCache {
) -> Result<Option<AppliedUpdate>> {
self.last_update_begin = DualTime::now();
info!(
"RTR cache applying update: availability={:?}, current_serial={}, incoming_payloads={}",
"RTR cache applying update: availability={:?}, current_serials={:?}, incoming_payloads={}",
self.availability,
self.serial,
self.serials(),
new_payloads.len()
);
let new_snapshot = Snapshot::from_payloads(new_payloads);
debug!(
"RTR cache built new snapshot from update: route_origins={}, router_keys={}, aspas={}, snapshot_empty={}",
new_snapshot.origins().len(),
new_snapshot.router_keys().len(),
new_snapshot.aspas().len(),
new_snapshot.is_empty()
);
if new_snapshot.is_empty() {
let source_snapshot = Snapshot::from_payloads(new_payloads);
if source_snapshot.is_empty() {
let changed = self.availability != CacheAvailability::NoDataAvailable
|| !self.snapshot.is_empty()
|| !self.deltas.is_empty();
|| self.versions.iter().any(|state| !state.snapshot.is_empty())
|| self.versions.iter().any(|state| !state.deltas.is_empty());
self.set_unavailable();
self.last_update_end = DualTime::now();
if !changed {
debug!(
"RTR cache update produced empty snapshot but cache was already unavailable; no state change"
);
return Ok(None);
}
info!(
"RTR cache update cleared usable data and marked cache unavailable: serial={}, session_ids={:?}",
self.serial, self.session_ids
);
return Ok(Some(AppliedUpdate {
availability: self.availability,
snapshot: Snapshot::empty(),
serial: self.serial,
session_ids: self.session_ids.clone(),
delta: None,
delta_window: None,
clear_delta_window: true,
}));
return Ok(Some(self.applied_update_with_clear()));
}
if self.availability == CacheAvailability::NoDataAvailable {
info!("RTR cache recovered from NoDataAvailable with non-empty snapshot");
return Ok(Some(self.reinitialize_from_snapshot(new_snapshot)));
return Ok(Some(self.reinitialize_from_snapshot(&source_snapshot)));
}
if self.snapshot.same_content(&new_snapshot) {
self.last_update_end = DualTime::now();
debug!(
"RTR cache update detected identical snapshot content: serial={}, session_ids={:?}",
self.serial, self.session_ids
let mut changed_any = false;
for version in 0..VERSION_COUNT {
let v = version as u8;
let projected = project_snapshot_for_version(&source_snapshot, v);
let state = &mut self.versions[version];
if state.snapshot.same_content(&projected) {
continue;
}
let (announced, withdrawn) = state.snapshot.diff(&projected);
if announced.is_empty() && withdrawn.is_empty() {
continue;
}
let new_serial = Self::next_serial(state);
let delta = Arc::new(Delta::new(new_serial, announced, withdrawn));
if delta.is_empty() {
continue;
}
state.snapshot = projected;
Self::push_delta(
state,
self.max_delta,
self.prune_delta_by_snapshot_size,
delta,
);
return Ok(None);
changed_any = true;
}
let (announced, withdrawn) = self.snapshot.diff(&new_snapshot);
debug!(
"RTR cache diff computed: announced={}, withdrawn={}, current_serial={}",
announced.len(),
withdrawn.len(),
self.serial
);
if announced.is_empty() && withdrawn.is_empty() {
self.last_update_end = DualTime::now();
debug!("RTR cache diff was empty after normalization; no update applied");
return Ok(None);
}
let new_serial = self.next_serial();
let delta = Arc::new(Delta::new(new_serial, announced, withdrawn));
if delta.is_empty() {
self.last_update_end = DualTime::now();
debug!(
"RTR cache delta collapsed to empty after dedup/order normalization: serial={}",
new_serial
);
return Ok(None);
}
self.replace_snapshot(new_snapshot.clone());
self.push_delta(delta.clone());
self.last_update_end = DualTime::now();
let delta_window = self.delta_window();
info!(
"RTR cache applied update: serial={}, announced={}, withdrawn={}, delta_window={:?}, snapshot(route_origins={}, router_keys={}, aspas={})",
new_serial,
delta.announced().len(),
delta.withdrawn().len(),
delta_window,
new_snapshot.origins().len(),
new_snapshot.router_keys().len(),
new_snapshot.aspas().len()
);
if !changed_any {
return Ok(None);
}
Ok(Some(AppliedUpdate {
info!(
"RTR cache applied update: serials={:?}, session_ids={:?}, delta_lengths={:?}",
self.serials(),
self.session_ids(),
self.delta_lengths()
);
Ok(Some(self.applied_update_with_windows()))
}
fn applied_update_with_clear(&self) -> AppliedUpdate {
let snapshots = std::array::from_fn(|idx| self.versions[idx].snapshot.clone());
let serials = std::array::from_fn(|idx| self.versions[idx].serial);
let session_ids = std::array::from_fn(|idx| self.versions[idx].session_id);
AppliedUpdate {
availability: self.availability,
snapshot: new_snapshot,
serial: new_serial,
session_ids: self.session_ids.clone(),
delta: Some(delta),
delta_window,
clear_delta_window: false,
}))
snapshots,
serials,
session_ids,
deltas: [None, None, None],
delta_windows: [None, None, None],
clear_delta_windows: [true, true, true],
}
}
fn applied_update_with_windows(&self) -> AppliedUpdate {
let snapshots = std::array::from_fn(|idx| self.versions[idx].snapshot.clone());
let serials = std::array::from_fn(|idx| self.versions[idx].serial);
let session_ids = std::array::from_fn(|idx| self.versions[idx].session_id);
let deltas = std::array::from_fn(|idx| self.versions[idx].deltas.back().cloned());
let delta_windows = std::array::from_fn(|idx| Self::delta_window(&self.versions[idx]));
AppliedUpdate {
availability: self.availability,
snapshots,
serials,
session_ids,
deltas,
delta_windows,
clear_delta_windows: [false, false, false],
}
}
pub fn is_data_available(&self) -> bool {
@@ -412,29 +416,33 @@ impl RtrCache {
}
pub fn session_id_for_version(&self, version: u8) -> u16 {
self.session_ids.get(version)
self.versions[version_index(version)].session_id
}
pub fn session_ids(&self) -> SessionIds {
self.session_ids.clone()
SessionIds::from_array(std::array::from_fn(|idx| self.versions[idx].session_id))
}
pub fn snapshot(&self) -> Snapshot {
self.snapshot.clone()
pub fn snapshot_for_version(&self, version: u8) -> Snapshot {
self.versions[version_index(version)].snapshot.clone()
}
pub fn serial(&self) -> u32 {
self.serial
pub fn serial_for_version(&self, version: u8) -> u32 {
self.versions[version_index(version)].serial
}
pub fn serials(&self) -> [u32; VERSION_COUNT] {
std::array::from_fn(|idx| self.versions[idx].serial)
}
pub fn delta_lengths(&self) -> [usize; VERSION_COUNT] {
std::array::from_fn(|idx| self.versions[idx].deltas.len())
}
pub fn timing(&self) -> Timing {
self.timing
}
pub fn current_snapshot_with_session_ids(&self) -> (&Snapshot, u32, SessionIds) {
(&self.snapshot, self.serial, self.session_ids.clone())
}
pub fn last_update_begin(&self) -> DualTime {
self.last_update_begin.clone()
}
@@ -447,151 +455,140 @@ impl RtrCache {
self.created_at.clone()
}
pub fn get_deltas_since(&self, client_serial: u32) -> SerialResult {
if client_serial == self.serial {
debug!(
"RTR cache delta query is already up to date: client_serial={}, cache_serial={}",
client_serial, self.serial
);
pub fn get_deltas_since_for_version(&self, version: u8, client_serial: u32) -> SerialResult {
let state = &self.versions[version_index(version)];
if client_serial == state.serial {
return SerialResult::UpToDate;
}
if matches!(
serial_cmp(client_serial, self.serial),
serial_cmp(client_serial, state.serial),
Some(Ordering::Greater) | None
) {
warn!(
"RTR cache delta query requires reset due to invalid/newer client serial: client_serial={}, cache_serial={}",
client_serial, self.serial
);
return SerialResult::ResetRequired;
}
let deltas = match self.collect_deltas_since(client_serial) {
let deltas = match collect_deltas_since(state, client_serial) {
Some(deltas) => deltas,
None => {
warn!(
"RTR cache delta query requires reset because requested serial is outside delta window: client_serial={}, cache_serial={}, delta_window={:?}",
client_serial,
self.serial,
self.delta_window()
);
return SerialResult::ResetRequired;
}
None => return SerialResult::ResetRequired,
};
if deltas.is_empty() {
debug!(
"RTR cache delta query resolved to no deltas: client_serial={}, cache_serial={}",
client_serial, self.serial
);
return SerialResult::UpToDate;
}
let merged = self.merge_deltas_minimally(&deltas);
let merged = merge_deltas_minimally(state.serial, &deltas);
if merged.is_empty() {
debug!(
"RTR cache merged delta query to empty result: client_serial={}, cache_serial={}, source_deltas={}",
client_serial,
self.serial,
deltas.len()
);
SerialResult::UpToDate
} else {
info!(
"RTR cache serving delta query: client_serial={}, cache_serial={}, source_deltas={}, merged_announced={}, merged_withdrawn={}",
client_serial,
self.serial,
deltas.len(),
merged.announced().len(),
merged.withdrawn().len()
);
SerialResult::Delta(merged)
}
}
}
fn collect_deltas_since(&self, client_serial: u32) -> Option<Vec<Arc<Delta>>> {
if self.deltas.is_empty() {
return None;
}
let oldest_serial = self.deltas.front().unwrap().serial();
let min_supported = oldest_serial.wrapping_sub(1);
if matches!(
serial_cmp(client_serial, min_supported),
Some(Ordering::Less) | None
) {
return None;
}
let mut result = Vec::new();
for delta in &self.deltas {
if serial_gt(delta.serial(), client_serial) {
result.push(delta.clone());
}
}
if let Some(first) = result.first() {
if first.serial() != client_serial.wrapping_add(1) {
return None;
}
}
Some(result)
fn collect_deltas_since(state: &VersionState, client_serial: u32) -> Option<Vec<Arc<Delta>>> {
if state.deltas.is_empty() {
return None;
}
fn merge_deltas_minimally(&self, deltas: &[Arc<Delta>]) -> Delta {
let mut states = BTreeMap::<ChangeKey, LogicalState>::new();
let oldest_serial = state.deltas.front().unwrap().serial();
let min_supported = oldest_serial.wrapping_sub(1);
for delta in deltas {
for payload in delta.withdrawn() {
let key = change_key(payload);
let state = states.entry(key).or_insert_with(LogicalState::new);
if matches!(
serial_cmp(client_serial, min_supported),
Some(Ordering::Less) | None
) {
return None;
}
if state.before.is_none() && state.after.is_none() {
state.before = Some(payload.clone());
}
state.after = None;
}
for payload in delta.announced() {
let key = change_key(payload);
let state = states.entry(key).or_insert_with(LogicalState::new);
state.after = Some(payload.clone());
let mut result = Vec::new();
for delta in &state.deltas {
if serial_gt(delta.serial(), client_serial) {
result.push(delta.clone());
}
}
if let Some(first) = result.first() {
if first.serial() != client_serial.wrapping_add(1) {
return None;
}
}
Some(result)
}
fn merge_deltas_minimally(current_serial: u32, deltas: &[Arc<Delta>]) -> Delta {
let mut states = BTreeMap::<ChangeKey, LogicalState>::new();
for delta in deltas {
for payload in delta.withdrawn() {
let key = change_key(payload);
let state = states.entry(key).or_insert_with(LogicalState::new);
if state.before.is_none() && state.after.is_none() {
state.before = Some(payload.clone());
}
state.after = None;
}
let mut announced = Vec::new();
let mut withdrawn = Vec::new();
for payload in delta.announced() {
let key = change_key(payload);
let state = states.entry(key).or_insert_with(LogicalState::new);
state.after = Some(payload.clone());
}
}
for (_key, state) in states {
match (state.before, state.after) {
(None, None) => {}
(None, Some(new_payload)) => {
announced.push(new_payload);
}
(Some(old_payload), None) => {
withdrawn.push(old_payload);
}
(Some(old_payload), Some(new_payload)) => {
if old_payload != new_payload {
if matches!(old_payload, Payload::Aspa(_))
&& matches!(new_payload, Payload::Aspa(_))
{
announced.push(new_payload);
} else {
withdrawn.push(old_payload);
announced.push(new_payload);
}
let mut announced = Vec::new();
let mut withdrawn = Vec::new();
for (_key, state) in states {
match (state.before, state.after) {
(None, None) => {}
(None, Some(new_payload)) => announced.push(new_payload),
(Some(old_payload), None) => withdrawn.push(old_payload),
(Some(old_payload), Some(new_payload)) => {
if old_payload != new_payload {
if matches!(old_payload, Payload::Aspa(_))
&& matches!(new_payload, Payload::Aspa(_))
{
announced.push(new_payload);
} else {
withdrawn.push(old_payload);
announced.push(new_payload);
}
}
}
}
}
Delta::new(self.serial, announced, withdrawn)
Delta::new(current_serial, announced, withdrawn)
}
fn project_snapshot_for_version(snapshot: &Snapshot, version: u8) -> Snapshot {
let mut payloads = Vec::new();
for payload in snapshot.payloads() {
if let Some(projected) = project_payload_for_version(&payload, version) {
payloads.push(projected);
}
}
Snapshot::from_payloads(payloads)
}
fn project_payload_for_version(payload: &Payload, version: u8) -> Option<Payload> {
match payload {
Payload::RouteOrigin(origin) => Some(Payload::RouteOrigin(origin.clone())),
Payload::RouterKey(key) => {
if version >= 1 {
Some(Payload::RouterKey(key.clone()))
} else {
None
}
}
Payload::Aspa(aspa) => {
if version >= 2 {
Some(Payload::Aspa(aspa.clone()))
} else {
None
}
}
}
}
@@ -659,12 +656,12 @@ pub enum SerialResult {
pub(super) struct AppliedUpdate {
pub(super) availability: CacheAvailability,
pub(super) snapshot: Snapshot,
pub(super) serial: u32,
pub(super) session_ids: SessionIds,
pub(super) delta: Option<Arc<Delta>>,
pub(super) delta_window: Option<(u32, u32)>,
pub(super) clear_delta_window: bool,
pub(super) snapshots: [Snapshot; VERSION_COUNT],
pub(super) serials: [u32; VERSION_COUNT],
pub(super) session_ids: [u16; VERSION_COUNT],
pub(super) deltas: [Option<Arc<Delta>>; VERSION_COUNT],
pub(super) delta_windows: [Option<(u32, u32)>; VERSION_COUNT],
pub(super) clear_delta_windows: [bool; VERSION_COUNT],
}
fn serial_cmp(a: u32, b: u32) -> Option<Ordering> {

202
src/rtr/cache/store.rs vendored
View File

@@ -7,7 +7,9 @@ use crate::rtr::payload::{Payload, Timing};
use crate::rtr::store::RtrStore;
use super::core::{AppliedUpdate, CacheAvailability, RtrCache, RtrCacheBuilder, SessionIds};
use super::model::Snapshot;
use super::model::{Delta, Snapshot};
const VERSION_COUNT: usize = 3;
impl RtrCache {
pub fn init(
@@ -22,13 +24,10 @@ impl RtrCache {
try_restore_from_store(store, max_delta, prune_delta_by_snapshot_size, timing)?
{
tracing::info!(
"RTR cache restored from store: availability={:?}, session_ids={:?}, serial={}, snapshot(route_origins={}, router_keys={}, aspas={})",
"RTR cache restored from store: availability={:?}, session_ids={:?}, serials={:?}",
cache.availability(),
cache.session_ids(),
cache.serial(),
cache.snapshot().origins().len(),
cache.snapshot().router_keys().len(),
cache.snapshot().aspas().len()
cache.serials()
);
return Ok(cache);
}
@@ -36,43 +35,37 @@ impl RtrCache {
tracing::warn!("RTR cache store unavailable or invalid, fallback to file loader");
let payloads = file_loader()?;
let session_ids = SessionIds::random_distinct();
let snapshot = Snapshot::from_payloads(payloads);
let availability = if snapshot.is_empty() {
let source_snapshot = Snapshot::from_payloads(payloads);
let availability = if source_snapshot.is_empty() {
CacheAvailability::NoDataAvailable
} else {
CacheAvailability::Ready
};
let serial = if snapshot.is_empty() { 0 } else { 1 };
let session_ids = SessionIds::random_distinct();
let serial = if source_snapshot.is_empty() { 0 } else { 1 };
if snapshot.is_empty() {
tracing::warn!(
"RTR cache initialized without usable data: session_ids={:?}, serial={}",
session_ids,
serial
);
} else {
tracing::info!(
"RTR cache initialized from file loader: session_ids={:?}, serial={}",
session_ids,
serial
);
}
let snapshot_for_store = snapshot.clone();
let session_ids_for_store = session_ids.clone();
let snapshots = std::array::from_fn(|version| {
project_snapshot_for_version(&source_snapshot, version as u8)
});
let serials = [serial; VERSION_COUNT];
let deltas = std::array::from_fn(|_| VecDeque::<Arc<Delta>>::with_capacity(max_delta as usize));
tokio::spawn({
let store = store.clone();
let snapshots_for_store = snapshots.clone();
let session_ids_for_store = session_ids.as_array();
async move {
if let Err(e) = store.save_cache_state(
let deltas_none: [Option<&Delta>; 3] = [None, None, None];
let windows_none: [Option<(u32, u32)>; 3] = [None, None, None];
let clear = [true, true, true];
if let Err(e) = store.save_cache_state_versioned(
availability,
&snapshot_for_store,
&snapshots_for_store,
&session_ids_for_store,
serial,
None,
None,
true,
&serials,
&deltas_none,
&windows_none,
&clear,
) {
tracing::error!("persist cache state failed: {:?}", e);
}
@@ -85,8 +78,9 @@ impl RtrCache {
.max_delta(max_delta)
.prune_delta_by_snapshot_size(prune_delta_by_snapshot_size)
.timing(timing)
.serial(serial)
.snapshot(snapshot)
.serials(serials)
.snapshots(snapshots)
.deltas_by_version(deltas)
.build())
}
@@ -94,7 +88,6 @@ impl RtrCache {
if let Some(update) = self.apply_update(new_payloads)? {
spawn_store_sync(store, update);
}
Ok(())
}
}
@@ -105,101 +98,98 @@ fn try_restore_from_store(
prune_delta_by_snapshot_size: bool,
timing: Timing,
) -> Result<Option<RtrCache>> {
let snapshot = store.get_snapshot()?;
let session_ids = store.get_session_ids()?;
let serial = store.get_serial()?;
let availability = store.get_availability()?;
let (snapshot, session_ids, serial) = match (snapshot, session_ids, serial) {
(Some(snapshot), Some(session_ids), Some(serial)) => (snapshot, session_ids, serial),
_ => {
tracing::warn!("RTR cache store incomplete: snapshot/session_ids/serial missing");
return Ok(None);
let mut snapshots = std::array::from_fn(|_| Snapshot::empty());
let mut session_ids = [0u16; VERSION_COUNT];
let mut serials = [0u32; VERSION_COUNT];
let mut deltas = std::array::from_fn(|_| VecDeque::<Arc<Delta>>::with_capacity(max_delta as usize));
for version in 0u8..=2 {
let idx = version as usize;
let snapshot = store.get_snapshot_for_version(version)?;
let session_id = store.get_session_id_for_version(version)?;
let serial = store.get_serial_for_version(version)?;
let (snapshot, session_id, serial) = match (snapshot, session_id, serial) {
(Some(snapshot), Some(session_id), Some(serial)) => (snapshot, session_id, serial),
_ => return Ok(None),
};
snapshots[idx] = snapshot;
session_ids[idx] = session_id;
serials[idx] = serial;
if availability == Some(CacheAvailability::NoDataAvailable) {
continue;
}
};
let availability = availability.unwrap_or_else(|| {
tracing::warn!("RTR cache store missing availability metadata, defaulting to Ready");
CacheAvailability::Ready
});
let deltas = if availability == CacheAvailability::NoDataAvailable {
tracing::warn!("RTR cache store restored in no-data-available state");
VecDeque::with_capacity(max_delta.into())
} else {
match store.get_delta_window()? {
Some((min_serial, max_serial)) => {
match store.load_delta_window(min_serial, max_serial) {
Ok(deltas) => deltas.into_iter().map(Arc::new).collect(),
Err(err) => {
tracing::warn!(
"RTR cache store delta recovery failed, treat store as unusable: {:?}",
err
);
return Ok(None);
}
}
}
None => {
tracing::info!("RTR cache store has no delta window, restore snapshot only");
VecDeque::with_capacity(max_delta.into())
if let Some((min_serial, max_serial)) = store.get_delta_window_for_version(version)? {
let mut loaded = store.load_delta_window_for_version(version, min_serial, max_serial)?;
let max_keep = usize::from(max_delta.max(1));
if loaded.len() > max_keep {
let drop_count = loaded.len() - max_keep;
let dropped_serials = loaded
.iter()
.take(drop_count)
.map(Delta::serial)
.collect::<Vec<_>>();
loaded.drain(..drop_count);
tracing::warn!(
"RTR cache restore truncated persisted deltas to max_delta: version={}, max_delta={}, dropped_count={}, dropped_serials={:?}",
version,
max_delta,
drop_count,
dropped_serials
);
}
deltas[idx] = loaded.into_iter().map(Arc::new).collect();
}
};
}
let availability = availability.unwrap_or(CacheAvailability::Ready);
Ok(Some(
RtrCacheBuilder::new()
.availability(availability)
.session_ids(session_ids)
.session_ids(SessionIds::from_array(session_ids))
.max_delta(max_delta)
.prune_delta_by_snapshot_size(prune_delta_by_snapshot_size)
.timing(timing)
.serial(serial)
.snapshot(snapshot)
.deltas(deltas)
.serials(serials)
.snapshots(snapshots)
.deltas_by_version(deltas)
.build(),
))
}
fn spawn_store_sync(store: &RtrStore, update: AppliedUpdate) {
let AppliedUpdate {
availability,
snapshot,
serial,
session_ids,
delta,
delta_window,
clear_delta_window,
} = update;
tokio::spawn({
let store = store.clone();
async move {
tracing::debug!(
"persisting RTR cache state: availability={:?}, serial={}, session_ids={:?}, delta_present={}, delta_window={:?}, clear_delta_window={}, snapshot(route_origins={}, router_keys={}, aspas={})",
availability,
serial,
session_ids,
delta.is_some(),
delta_window,
clear_delta_window,
snapshot.origins().len(),
snapshot.router_keys().len(),
snapshot.aspas().len()
);
if let Err(e) = store.save_cache_state(
availability,
&snapshot,
&session_ids,
serial,
delta.as_deref(),
delta_window,
clear_delta_window,
let delta_refs: [Option<&Delta>; 3] =
std::array::from_fn(|idx| update.deltas[idx].as_deref());
if let Err(e) = store.save_cache_state_versioned(
update.availability,
&update.snapshots,
&update.session_ids,
&update.serials,
&delta_refs,
&update.delta_windows,
&update.clear_delta_windows,
) {
tracing::error!("persist cache state failed: {:?}", e);
} else {
tracing::debug!("persist RTR cache state completed: serial={}", serial);
}
}
});
}
fn project_snapshot_for_version(snapshot: &Snapshot, version: u8) -> Snapshot {
let mut payloads = Vec::new();
for payload in snapshot.payloads() {
match payload {
Payload::RouteOrigin(_) => payloads.push(payload),
Payload::RouterKey(_) if version >= 1 => payloads.push(payload),
Payload::Aspa(_) if version >= 2 => payloads.push(payload),
_ => {}
}
}
Snapshot::from_payloads(payloads)
}

24
src/rtr/session.rs
View File

@@ -558,10 +558,10 @@ where
.read()
.map_err(|_| anyhow!("cache read lock poisoned"))?;
let data_available = cache.is_data_available();
let snapshot = cache.snapshot();
let snapshot = cache.snapshot_for_version(version);
let payloads = snapshot.payloads_for_rtr();
let session_id = cache.session_id_for_version(version);
let serial = cache.serial();
let serial = cache.serial_for_version(version);
(data_available, payloads, session_id, serial)
};
@@ -644,7 +644,7 @@ where
.cache
.read()
.map_err(|_| anyhow!("cache read lock poisoned"))?;
cache.get_deltas_since(client_serial)
cache.get_deltas_since_for_version(version, client_serial)
};
match serial_result {
@@ -665,7 +665,10 @@ where
.cache
.read()
.map_err(|_| anyhow!("cache read lock poisoned"))?;
(cache.session_id_for_version(version), cache.serial())
(
cache.session_id_for_version(version),
cache.serial_for_version(version),
)
};
self.write_end_of_data(current_session, current_serial)
@@ -687,7 +690,10 @@ where
.cache
.read()
.map_err(|_| anyhow!("cache read lock poisoned"))?;
(cache.session_id_for_version(version), cache.serial())
(
cache.session_id_for_version(version),
cache.serial_for_version(version),
)
};
self.write_cache_response(current_session).await?;
@@ -734,7 +740,10 @@ where
.cache
.read()
.map_err(|_| anyhow!("cache read lock poisoned"))?;
(cache.session_id_for_version(version), cache.serial())
(
cache.session_id_for_version(version),
cache.serial_for_version(version),
)
};
debug!(
@@ -772,7 +781,8 @@ where
.cache
.read()
.ok()
.map(|cache| cache.serial().to_string())
.and_then(|cache| self.version.map(|version| cache.serial_for_version(version)))
.map(|serial| serial.to_string())
.unwrap_or_else(|| "<unavailable>".to_string());
let session_id = self
.version

664
src/rtr/store.rs
View File

@@ -1,41 +1,49 @@
use anyhow::{Result, anyhow};
use rocksdb::{ColumnFamilyDescriptor, DB, Direction, IteratorMode, Options, WriteBatch};
use serde::{Serialize, de::DeserializeOwned};
use rocksdb::{ColumnFamilyDescriptor, DB, IteratorMode, Options, WriteBatch};
use serde::de::DeserializeOwned;
use std::path::Path;
use std::sync::Arc;
use tokio::task;
use tracing::{debug, info, warn};
use tracing::{info, warn};
use crate::rtr::cache::{CacheAvailability, Delta, SessionIds, Snapshot};
use crate::rtr::state::State;
use crate::rtr::cache::{CacheAvailability, Delta, Snapshot};
const CF_META: &str = "meta";
const CF_SNAPSHOT: &str = "snapshot";
const CF_DELTA: &str = "delta";
const META_STATE: &[u8] = b"state";
const META_SESSION_IDS: &[u8] = b"session_ids";
const META_SERIAL: &[u8] = b"serial";
const META_AVAILABILITY: &[u8] = b"availability";
const META_DELTA_MIN: &[u8] = b"delta_min";
const META_DELTA_MAX: &[u8] = b"delta_max";
const META_SESSION_ID_PREFIX: &str = "session_id_v";
const META_SERIAL_PREFIX: &str = "serial_v";
const META_DELTA_MIN_PREFIX: &str = "delta_min_v";
const META_DELTA_MAX_PREFIX: &str = "delta_max_v";
const SNAPSHOT_CURRENT_PREFIX: &str = "current_v";
const DELTA_KEY_PREFIX: u8 = b'd';
const DELTA_KEY_V2_PREFIX: u8 = b'D';
fn delta_key(serial: u32) -> [u8; 5] {
let mut key = [0u8; 5];
key[0] = DELTA_KEY_PREFIX;
key[1..].copy_from_slice(&serial.to_be_bytes());
fn delta_key_v2(version: u8, serial: u32) -> [u8; 6] {
let mut key = [0u8; 6];
key[0] = DELTA_KEY_V2_PREFIX;
key[1] = version;
key[2..].copy_from_slice(&serial.to_be_bytes());
key
}
fn delta_key_serial(key: &[u8]) -> Option<u32> {
if key.len() != 5 || key[0] != DELTA_KEY_PREFIX {
fn delta_key_v2_serial(key: &[u8]) -> Option<(u8, u32)> {
if key.len() != 6 || key[0] != DELTA_KEY_V2_PREFIX {
return None;
}
let version = key[1];
let mut bytes = [0u8; 4];
bytes.copy_from_slice(&key[1..]);
Some(u32::from_be_bytes(bytes))
bytes.copy_from_slice(&key[2..]);
Some((version, u32::from_be_bytes(bytes)))
}
fn meta_key(prefix: &str, version: u8) -> Vec<u8> {
format!("{}{}", prefix, version).into_bytes()
}
fn snapshot_key(version: u8) -> Vec<u8> {
format!("{}{}", SNAPSHOT_CURRENT_PREFIX, version).into_bytes()
}
#[derive(Clone)]
@@ -44,7 +52,6 @@ pub struct RtrStore {
}
impl RtrStore {
/// Open or create DB with required column families.
pub fn open<P: AsRef<Path>>(path: P) -> Result<Self> {
let path_ref = path.as_ref();
let mut opts = Options::default();
@@ -60,22 +67,9 @@ impl RtrStore {
info!("opening RTR RocksDB store at {}", path_ref.display());
let db = Arc::new(DB::open_cf_descriptors(&opts, path_ref, cfs)?);
info!("opened RTR RocksDB store at {}", path_ref.display());
Ok(Self { db })
}
/// Common serialize/put.
fn put_cf<T: Serialize>(&self, cf: &str, key: &[u8], value: &T) -> Result<()> {
let cf_handle = self
.db
.cf_handle(cf)
.ok_or_else(|| anyhow!("CF not found"))?;
let data = serde_json::to_vec(value)?;
self.db.put_cf(cf_handle, key, data)?;
Ok(())
}
/// Common get/deserialize.
fn get_cf<T: DeserializeOwned>(&self, cf: &str, key: &[u8]) -> Result<Option<T>> {
let cf_handle = self
.db
@@ -89,211 +83,134 @@ impl RtrStore {
}
}
/// Common delete.
fn delete_cf(&self, cf: &str, key: &[u8]) -> Result<()> {
let cf_handle = self
.db
.cf_handle(cf)
.ok_or_else(|| anyhow!("CF not found"))?;
self.db.delete_cf(cf_handle, key)?;
Ok(())
}
// ===============================
// Meta/state
// ===============================
pub fn set_state(&self, state: &State) -> Result<()> {
self.put_cf(CF_META, META_STATE, &state)
}
pub fn get_state(&self) -> Result<Option<State>> {
self.get_cf(CF_META, META_STATE)
}
pub fn set_meta(&self, meta: &State) -> Result<()> {
self.set_state(meta)
}
pub fn get_meta(&self) -> Result<Option<State>> {
self.get_state()
}
pub fn set_session_ids(&self, session_ids: &SessionIds) -> Result<()> {
self.put_cf(CF_META, META_SESSION_IDS, session_ids)
}
pub fn get_session_ids(&self) -> Result<Option<SessionIds>> {
self.get_cf(CF_META, META_SESSION_IDS)
}
pub fn set_serial(&self, serial: u32) -> Result<()> {
self.put_cf(CF_META, META_SERIAL, &serial)
}
pub fn get_serial(&self) -> Result<Option<u32>> {
self.get_cf(CF_META, META_SERIAL)
}
pub fn set_availability(&self, availability: CacheAvailability) -> Result<()> {
self.put_cf(CF_META, META_AVAILABILITY, &availability)
}
pub fn get_availability(&self) -> Result<Option<CacheAvailability>> {
self.get_cf(CF_META, META_AVAILABILITY)
}
pub fn set_delta_window(&self, min_serial: u32, max_serial: u32) -> Result<()> {
debug!(
"RTR store persisting delta window metadata: min_serial={}, max_serial={}",
min_serial, max_serial
);
let meta_cf = self
.db
.cf_handle(CF_META)
.ok_or_else(|| anyhow!("CF_META not found"))?;
let mut batch = WriteBatch::default();
batch.put_cf(meta_cf, META_DELTA_MIN, serde_json::to_vec(&min_serial)?);
batch.put_cf(meta_cf, META_DELTA_MAX, serde_json::to_vec(&max_serial)?);
self.db.write(batch)?;
Ok(())
pub fn get_session_id_for_version(&self, version: u8) -> Result<Option<u16>> {
let key = meta_key(META_SESSION_ID_PREFIX, version);
self.get_cf(CF_META, &key)
}
pub fn clear_delta_window(&self) -> Result<()> {
debug!("RTR store clearing delta window metadata");
let meta_cf = self
.db
.cf_handle(CF_META)
.ok_or_else(|| anyhow!("CF_META not found"))?;
let mut batch = WriteBatch::default();
batch.delete_cf(meta_cf, META_DELTA_MIN);
batch.delete_cf(meta_cf, META_DELTA_MAX);
self.db.write(batch)?;
Ok(())
pub fn get_serial_for_version(&self, version: u8) -> Result<Option<u32>> {
let key = meta_key(META_SERIAL_PREFIX, version);
self.get_cf(CF_META, &key)
}
pub fn get_delta_window(&self) -> Result<Option<(u32, u32)>> {
let min: Option<u32> = self.get_cf(CF_META, META_DELTA_MIN)?;
let max: Option<u32> = self.get_cf(CF_META, META_DELTA_MAX)?;
pub fn get_delta_window_for_version(&self, version: u8) -> Result<Option<(u32, u32)>> {
let min_key = meta_key(META_DELTA_MIN_PREFIX, version);
let max_key = meta_key(META_DELTA_MAX_PREFIX, version);
let min: Option<u32> = self.get_cf(CF_META, &min_key)?;
let max: Option<u32> = self.get_cf(CF_META, &max_key)?;
match (min, max) {
(Some(min), Some(max)) => {
debug!(
"RTR store loaded delta window metadata: min_serial={}, max_serial={}",
min, max
);
Ok(Some((min, max)))
}
(Some(min), Some(max)) => Ok(Some((min, max))),
(None, None) => Ok(None),
_ => Err(anyhow!("Inconsistent DB state: delta window mismatch")),
_ => Err(anyhow!(
"Inconsistent DB state: delta window mismatch for version {}",
version
)),
}
}
pub fn delete_state(&self) -> Result<()> {
self.delete_cf(CF_META, META_STATE)
pub fn get_snapshot_for_version(&self, version: u8) -> Result<Option<Snapshot>> {
let key = snapshot_key(version);
self.get_cf(CF_SNAPSHOT, &key)
}
pub fn delete_serial(&self) -> Result<()> {
self.delete_cf(CF_META, META_SERIAL)
pub fn get_delta_for_version(&self, version: u8, serial: u32) -> Result<Option<Delta>> {
self.get_cf(CF_DELTA, &delta_key_v2(version, serial))
}
// ===============================
// Snapshot
// ===============================
pub fn save_snapshot(&self, snapshot: &Snapshot) -> Result<()> {
pub fn load_delta_window_for_version(
&self,
version: u8,
min_serial: u32,
max_serial: u32,
) -> Result<Vec<Delta>> {
let cf_handle = self
.db
.cf_handle(CF_SNAPSHOT)
.ok_or_else(|| anyhow!("CF_SNAPSHOT not found"))?;
let mut batch = WriteBatch::default();
let data = serde_json::to_vec(snapshot)?;
batch.put_cf(cf_handle, b"current", data);
self.db.write(batch)?;
Ok(())
.cf_handle(CF_DELTA)
.ok_or_else(|| anyhow!("CF_DELTA not found"))?;
let iter = self.db.iterator_cf(cf_handle, IteratorMode::Start);
let mut out = Vec::new();
for item in iter {
let (key, value) = item.map_err(|e| anyhow!("rocksdb iterator error: {}", e))?;
let Some((key_version, serial)) = delta_key_v2_serial(key.as_ref()) else {
continue;
};
if key_version != version {
continue;
}
if serial_in_window(serial, min_serial, max_serial) {
let delta: Delta = serde_json::from_slice(value.as_ref())?;
out.push(delta);
}
}
out.sort_by_key(|delta| delta.serial().wrapping_sub(min_serial));
validate_delta_window(&out, min_serial, max_serial)?;
Ok(out)
}
pub fn get_snapshot(&self) -> Result<Option<Snapshot>> {
self.get_cf(CF_SNAPSHOT, b"current")
}
pub fn delete_snapshot(&self) -> Result<()> {
self.delete_cf(CF_SNAPSHOT, b"current")
}
pub fn save_snapshot_and_state(&self, snapshot: &Snapshot, state: &State) -> Result<()> {
let snapshot_cf = self
fn list_delta_keys_for_version(&self, version: u8) -> Result<Vec<Vec<u8>>> {
let cf_handle = self
.db
.cf_handle(CF_SNAPSHOT)
.ok_or_else(|| anyhow!("CF_SNAPSHOT not found"))?;
let meta_cf = self
.db
.cf_handle(CF_META)
.ok_or_else(|| anyhow!("CF_META not found"))?;
let mut batch = WriteBatch::default();
batch.put_cf(snapshot_cf, b"current", serde_json::to_vec(snapshot)?);
batch.put_cf(meta_cf, META_STATE, serde_json::to_vec(state)?);
batch.put_cf(
meta_cf,
META_SESSION_IDS,
serde_json::to_vec(&state.clone().session_ids())?,
);
batch.put_cf(
meta_cf,
META_SERIAL,
serde_json::to_vec(&state.clone().serial())?,
);
self.db.write(batch)?;
Ok(())
.cf_handle(CF_DELTA)
.ok_or_else(|| anyhow!("CF_DELTA not found"))?;
let iter = self.db.iterator_cf(cf_handle, IteratorMode::Start);
let mut keys = Vec::new();
for item in iter {
let (key, _value) = item.map_err(|e| anyhow!("rocksdb iterator error: {}", e))?;
if matches!(delta_key_v2_serial(key.as_ref()), Some((v, _)) if v == version) {
keys.push(key.to_vec());
}
}
Ok(keys)
}
pub fn save_snapshot_and_meta(
fn list_delta_keys_outside_window_for_version(
&self,
snapshot: &Snapshot,
session_ids: &SessionIds,
serial: u32,
) -> Result<()> {
let mut batch = WriteBatch::default();
let snapshot_cf = self
version: u8,
min_serial: u32,
max_serial: u32,
) -> Result<Vec<Vec<u8>>> {
let cf_handle = self
.db
.cf_handle(CF_SNAPSHOT)
.ok_or_else(|| anyhow!("CF_SNAPSHOT not found"))?;
let meta_cf = self
.db
.cf_handle(CF_META)
.ok_or_else(|| anyhow!("CF_META not found"))?;
batch.put_cf(snapshot_cf, b"current", serde_json::to_vec(snapshot)?);
batch.put_cf(meta_cf, META_SESSION_IDS, serde_json::to_vec(session_ids)?);
batch.put_cf(meta_cf, META_SERIAL, serde_json::to_vec(&serial)?);
self.db.write(batch)?;
Ok(())
.cf_handle(CF_DELTA)
.ok_or_else(|| anyhow!("CF_DELTA not found"))?;
let iter = self.db.iterator_cf(cf_handle, IteratorMode::Start);
let mut keys = Vec::new();
for item in iter {
let (key, _value) = item.map_err(|e| anyhow!("rocksdb iterator error: {}", e))?;
let Some((key_version, serial)) = delta_key_v2_serial(key.as_ref()) else {
continue;
};
if key_version != version {
continue;
}
if !serial_in_window(serial, min_serial, max_serial) {
keys.push(key.to_vec());
}
}
Ok(keys)
}
pub fn save_cache_state(
/// Persist full versioned RTR cache state in one atomic batch.
///
/// Write-boundary contract:
/// - Production writes must go through `src/rtr/cache/store.rs`.
/// - Direct callers should be limited to DB contract tests.
/// - Do not introduce ad-hoc write paths outside cache/store, otherwise
/// session_id/serial/snapshot/delta_window consistency can be broken.
pub fn save_cache_state_versioned(
&self,
availability: CacheAvailability,
snapshot: &Snapshot,
session_ids: &SessionIds,
serial: u32,
delta: Option<&Delta>,
delta_window: Option<(u32, u32)>,
clear_delta_window: bool,
snapshots: &[Snapshot; 3],
session_ids: &[u16; 3],
serials: &[u32; 3],
deltas: &[Option<&Delta>; 3],
delta_windows: &[Option<(u32, u32)>; 3],
clear_delta_windows: &[bool; 3],
) -> Result<()> {
debug!(
"RTR store save_cache_state start: availability={:?}, serial={}, session_ids={:?}, delta_present={}, delta_window={:?}, clear_delta_window={}, snapshot(route_origins={}, router_keys={}, aspas={})",
availability,
serial,
session_ids,
delta.is_some(),
delta_window,
clear_delta_window,
snapshot.origins().len(),
snapshot.router_keys().len(),
snapshot.aspas().len()
);
let snapshot_cf = self
.db
.cf_handle(CF_SNAPSHOT)
@@ -306,275 +223,68 @@ impl RtrStore {
.db
.cf_handle(CF_DELTA)
.ok_or_else(|| anyhow!("CF_DELTA not found"))?;
let mut batch = WriteBatch::default();
batch.put_cf(snapshot_cf, b"current", serde_json::to_vec(snapshot)?);
batch.put_cf(meta_cf, META_SESSION_IDS, serde_json::to_vec(session_ids)?);
batch.put_cf(meta_cf, META_SERIAL, serde_json::to_vec(&serial)?);
let mut batch = WriteBatch::default();
batch.put_cf(
meta_cf,
META_AVAILABILITY,
serde_json::to_vec(&availability)?,
);
if let Some(delta) = delta {
debug!(
"RTR store persisting delta: serial={}, announced={}, withdrawn={}",
delta.serial(),
delta.announced().len(),
delta.withdrawn().len()
for version in 0u8..=2 {
let idx = version as usize;
batch.put_cf(
snapshot_cf,
snapshot_key(version),
serde_json::to_vec(&snapshots[idx])?,
);
batch.put_cf(
delta_cf,
delta_key(delta.serial()),
serde_json::to_vec(delta)?,
meta_cf,
meta_key(META_SESSION_ID_PREFIX, version),
serde_json::to_vec(&session_ids[idx])?,
);
batch.put_cf(
meta_cf,
meta_key(META_SERIAL_PREFIX, version),
serde_json::to_vec(&serials[idx])?,
);
}
if clear_delta_window {
let existing_keys = self.list_delta_keys()?;
let existing_serials = summarize_delta_keys(&existing_keys);
info!(
"RTR store clearing persisted delta window: deleting {} delta records, serials={}",
existing_keys.len(),
existing_serials
);
batch.delete_cf(meta_cf, META_DELTA_MIN);
batch.delete_cf(meta_cf, META_DELTA_MAX);
for key in existing_keys {
batch.delete_cf(delta_cf, key);
}
} else if let Some((min_serial, max_serial)) = delta_window {
batch.put_cf(meta_cf, META_DELTA_MIN, serde_json::to_vec(&min_serial)?);
batch.put_cf(meta_cf, META_DELTA_MAX, serde_json::to_vec(&max_serial)?);
// Serial numbers are compared in RFC 1982 ring order, while RocksDB stores
// keys in plain lexicographic order. After wraparound, a window such as
// [u32::MAX, 1] is contiguous in serial space but split in key space, so a
// simple delete_range() would leave stale high-serial keys behind.
let stale_keys = self.list_delta_keys_outside_window(min_serial, max_serial)?;
if !stale_keys.is_empty() {
info!(
"RTR store pruning stale delta records outside window [{}, {}]: count={}, serials={}",
min_serial,
max_serial,
stale_keys.len(),
summarize_delta_keys(&stale_keys)
);
} else {
debug!(
"RTR store found no stale delta records outside window [{}, {}]",
min_serial, max_serial
if let Some(delta) = deltas[idx] {
batch.put_cf(
delta_cf,
delta_key_v2(version, delta.serial()),
serde_json::to_vec(delta)?,
);
}
for key in stale_keys {
batch.delete_cf(delta_cf, key);
if clear_delta_windows[idx] {
batch.delete_cf(meta_cf, meta_key(META_DELTA_MIN_PREFIX, version));
batch.delete_cf(meta_cf, meta_key(META_DELTA_MAX_PREFIX, version));
for key in self.list_delta_keys_for_version(version)? {
batch.delete_cf(delta_cf, key);
}
} else if let Some((min_serial, max_serial)) = delta_windows[idx] {
batch.put_cf(
meta_cf,
meta_key(META_DELTA_MIN_PREFIX, version),
serde_json::to_vec(&min_serial)?,
);
batch.put_cf(
meta_cf,
meta_key(META_DELTA_MAX_PREFIX, version),
serde_json::to_vec(&max_serial)?,
);
for key in self
.list_delta_keys_outside_window_for_version(version, min_serial, max_serial)?
{
batch.delete_cf(delta_cf, key);
}
}
}
self.db.write(batch)?;
debug!("RTR store save_cache_state completed: serial={}", serial);
Ok(())
}
pub fn save_snapshot_and_serial(&self, snapshot: &Snapshot, serial: u32) -> Result<()> {
let mut batch = WriteBatch::default();
let snapshot_cf = self
.db
.cf_handle(CF_SNAPSHOT)
.ok_or_else(|| anyhow!("CF_SNAPSHOT not found"))?;
let meta_cf = self
.db
.cf_handle(CF_META)
.ok_or_else(|| anyhow!("CF_META not found"))?;
batch.put_cf(snapshot_cf, b"current", serde_json::to_vec(snapshot)?);
batch.put_cf(meta_cf, META_SERIAL, serde_json::to_vec(&serial)?);
self.db.write(batch)?;
Ok(())
}
pub async fn save_snapshot_and_serial_async(
self: Arc<Self>,
snapshot: Snapshot,
serial: u32,
) -> Result<()> {
let snapshot_bytes = serde_json::to_vec(&snapshot)?;
let serial_bytes = serde_json::to_vec(&serial)?;
task::spawn_blocking(move || {
let mut batch = WriteBatch::default();
let snapshot_cf = self
.db
.cf_handle(CF_SNAPSHOT)
.ok_or_else(|| anyhow!("CF_SNAPSHOT not found"))?;
let meta_cf = self
.db
.cf_handle(CF_META)
.ok_or_else(|| anyhow!("CF_META not found"))?;
batch.put_cf(snapshot_cf, b"current", snapshot_bytes);
batch.put_cf(meta_cf, META_SERIAL, serial_bytes);
self.db.write(batch)?;
Ok::<_, anyhow::Error>(())
})
.await??;
Ok(())
}
pub fn load_snapshot_and_state(&self) -> Result<Option<(Snapshot, State)>> {
let snapshot: Option<Snapshot> = self.get_snapshot()?;
let state: Option<State> = self.get_state()?;
match (snapshot, state) {
(Some(snap), Some(state)) => Ok(Some((snap, state))),
(None, None) => Ok(None),
_ => Err(anyhow!(
"Inconsistent DB state: snapshot and state mismatch"
)),
}
}
pub fn load_snapshot_and_serial(&self) -> Result<Option<(Snapshot, u32)>> {
let snapshot: Option<Snapshot> = self.get_snapshot()?;
let serial: Option<u32> = self.get_serial()?;
match (snapshot, serial) {
(Some(snap), Some(serial)) => Ok(Some((snap, serial))),
(None, None) => Ok(None),
_ => Err(anyhow!(
"Inconsistent DB state: snapshot and serial mismatch"
)),
}
}
// ===============================
// Delta
// ===============================
pub fn save_delta(&self, delta: &Delta) -> Result<()> {
self.put_cf(CF_DELTA, &delta_key(delta.serial()), delta)
}
pub fn get_delta(&self, serial: u32) -> Result<Option<Delta>> {
self.get_cf(CF_DELTA, &delta_key(serial))
}
pub fn load_deltas_since(&self, serial: u32) -> Result<Vec<Delta>> {
let cf_handle = self
.db
.cf_handle(CF_DELTA)
.ok_or_else(|| anyhow!("CF_DELTA not found"))?;
let start_key = delta_key(serial.wrapping_add(1));
let iter = self.db.iterator_cf(
cf_handle,
IteratorMode::From(&start_key, Direction::Forward),
);
let mut out = Vec::new();
for item in iter {
let (key, value) = item.map_err(|e| anyhow!("rocksdb iterator error: {}", e))?;
let parsed =
delta_key_serial(key.as_ref()).ok_or_else(|| anyhow!("Invalid delta key"))?;
if parsed <= serial {
continue;
}
let delta: Delta = serde_json::from_slice(value.as_ref())?;
out.push(delta);
}
Ok(out)
}
pub fn load_delta_window(&self, min_serial: u32, max_serial: u32) -> Result<Vec<Delta>> {
info!(
"RTR store loading persisted delta window: min_serial={}, max_serial={}",
min_serial, max_serial
);
let cf_handle = self
.db
.cf_handle(CF_DELTA)
.ok_or_else(|| anyhow!("CF_DELTA not found"))?;
let iter = self.db.iterator_cf(cf_handle, IteratorMode::Start);
let mut out = Vec::new();
for item in iter {
let (key, value) = item.map_err(|e| anyhow!("rocksdb iterator error: {}", e))?;
let parsed =
delta_key_serial(key.as_ref()).ok_or_else(|| anyhow!("Invalid delta key"))?;
// Restore by the persisted window bounds instead of load_deltas_since().
// The latter follows lexicographic key order and is not safe across serial
// wraparound, where older high-valued keys may otherwise be pulled back in.
if serial_in_window(parsed, min_serial, max_serial) {
let delta: Delta = serde_json::from_slice(value.as_ref())?;
out.push(delta);
}
}
out.sort_by_key(|delta| delta.serial().wrapping_sub(min_serial));
debug!(
"RTR store loaded delta candidates for window [{}, {}]: count={}, serials={}",
min_serial,
max_serial,
out.len(),
summarize_delta_serials(&out)
);
validate_delta_window(&out, min_serial, max_serial)?;
info!(
"RTR store restored valid delta window: min_serial={}, max_serial={}, count={}, serials={}",
min_serial,
max_serial,
out.len(),
summarize_delta_serials(&out)
);
Ok(out)
}
pub fn delete_delta(&self, serial: u32) -> Result<()> {
self.delete_cf(CF_DELTA, &delta_key(serial))
}
fn list_delta_keys(&self) -> Result<Vec<Vec<u8>>> {
let cf_handle = self
.db
.cf_handle(CF_DELTA)
.ok_or_else(|| anyhow!("CF_DELTA not found"))?;
let iter = self.db.iterator_cf(cf_handle, IteratorMode::Start);
let mut keys = Vec::new();
for item in iter {
let (key, _value) = item.map_err(|e| anyhow!("rocksdb iterator error: {}", e))?;
keys.push(key.to_vec());
}
Ok(keys)
}
fn list_delta_keys_outside_window(
&self,
min_serial: u32,
max_serial: u32,
) -> Result<Vec<Vec<u8>>> {
let cf_handle = self
.db
.cf_handle(CF_DELTA)
.ok_or_else(|| anyhow!("CF_DELTA not found"))?;
let iter = self.db.iterator_cf(cf_handle, IteratorMode::Start);
let mut keys = Vec::new();
for item in iter {
let (key, _value) = item.map_err(|e| anyhow!("rocksdb iterator error: {}", e))?;
let serial =
delta_key_serial(key.as_ref()).ok_or_else(|| anyhow!("Invalid delta key"))?;
if !serial_in_window(serial, min_serial, max_serial) {
keys.push(key.to_vec());
}
}
Ok(keys)
}
}
fn serial_in_window(serial: u32, min_serial: u32, max_serial: u32) -> bool {
@@ -641,41 +351,3 @@ fn validate_delta_window(deltas: &[Delta], min_serial: u32, max_serial: u32) ->
Ok(())
}
fn summarize_delta_keys(keys: &[Vec<u8>]) -> String {
let serials: Vec<u32> = keys
.iter()
.filter_map(|key| delta_key_serial(key))
.collect();
summarize_serials(&serials)
}
fn summarize_delta_serials(deltas: &[Delta]) -> String {
let serials: Vec<u32> = deltas.iter().map(Delta::serial).collect();
summarize_serials(&serials)
}
fn summarize_serials(serials: &[u32]) -> String {
const MAX_INLINE: usize = 12;
if serials.is_empty() {
return "[]".to_string();
}
if serials.len() <= MAX_INLINE {
return format!("{:?}", serials);
}
let head: Vec<u32> = serials.iter().take(6).copied().collect();
let tail: Vec<u32> = serials
.iter()
.rev()
.take(3)
.copied()
.collect::<Vec<_>>()
.into_iter()
.rev()
.collect();
format!("{:?} ... {:?} (total={})", head, tail, serials.len())
}