#![allow(dead_code)]
use crate::ring::PeerKeyLocation;
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub(crate) enum Action {
Subscribe,
Renew,
Unsubscribe,
Collapse,
Announce,
Retract,
ReRootSearch,
}
#[derive(Debug, Clone)]
pub(crate) struct ReconcileInputs {
pub computed_upstream: Option<PeerKeyLocation>,
pub has_local_client: bool,
pub has_farther_downstream_subscriber: bool,
pub has_recent_local_client_access: bool,
pub state_present: bool,
pub is_subscribed: bool,
pub is_advertised: bool,
pub is_verified_root: bool,
pub actively_acquiring: bool,
}
pub(crate) fn reconcile(inputs: &ReconcileInputs) -> Vec<Action> {
let contract_in_use = contract_in_use(inputs);
if !contract_in_use {
let mut actions = Vec::new();
if inputs.is_subscribed {
actions.push(Action::Collapse);
if inputs.computed_upstream.is_some() {
actions.push(Action::Unsubscribe);
}
}
if inputs.is_advertised {
actions.push(Action::Retract);
}
return actions;
}
if !inputs.state_present && !inputs.is_subscribed {
return Vec::new();
}
let has_host_role =
inputs.computed_upstream.is_some() || inputs.is_verified_root || inputs.actively_acquiring;
let mut actions = Vec::new();
if inputs.is_subscribed {
actions.push(Action::Renew);
}
if inputs.computed_upstream.is_some() && !inputs.is_subscribed {
actions.push(Action::Subscribe);
}
if inputs.computed_upstream.is_none() && !inputs.is_verified_root && !inputs.actively_acquiring
{
actions.push(Action::ReRootSearch);
}
if inputs.state_present && has_host_role && !inputs.is_advertised {
actions.push(Action::Announce);
}
actions
}
#[derive(Debug, Default, Clone, Copy, PartialEq, Eq)]
pub(crate) struct ReconcileActionDivergence {
pub subscribe: bool,
pub renew: bool,
pub unsubscribe: bool,
pub collapse: bool,
pub announce: bool,
pub retract: bool,
pub reroot_search: bool,
}
impl ReconcileActionDivergence {
pub fn any(&self) -> bool {
self.subscribe
|| self.renew
|| self.unsubscribe
|| self.collapse
|| self.announce
|| self.retract
|| self.reroot_search
}
}
pub(crate) fn action_set_divergence(
reconcile_actions: &[Action],
actual_actions: &[Action],
) -> ReconcileActionDivergence {
fn flag(div: &mut ReconcileActionDivergence, action: Action) {
match action {
Action::Subscribe => div.subscribe = true,
Action::Renew => div.renew = true,
Action::Unsubscribe => div.unsubscribe = true,
Action::Collapse => div.collapse = true,
Action::Announce => div.announce = true,
Action::Retract => div.retract = true,
Action::ReRootSearch => div.reroot_search = true,
}
}
let mut div = ReconcileActionDivergence::default();
for &a in reconcile_actions {
if !actual_actions.contains(&a) {
flag(&mut div, a);
}
}
for &a in actual_actions {
if !reconcile_actions.contains(&a) {
flag(&mut div, a);
}
}
div
}
pub(crate) fn action_set_divergence_focused(
reconcile_actions: &[Action],
actual_actions: &[Action],
relevant: &[Action],
) -> ReconcileActionDivergence {
let r: Vec<Action> = reconcile_actions
.iter()
.copied()
.filter(|a| relevant.contains(a))
.collect();
let a: Vec<Action> = actual_actions
.iter()
.copied()
.filter(|a| relevant.contains(a))
.collect();
action_set_divergence(&r, &a)
}
pub(crate) fn contract_in_use(inputs: &ReconcileInputs) -> bool {
inputs.has_local_client
|| inputs.has_farther_downstream_subscriber
|| inputs.has_recent_local_client_access
}
pub(crate) fn wants_renewal(inputs: &ReconcileInputs) -> bool {
contract_in_use(inputs)
}
pub(crate) fn wants_collapse(inputs: &ReconcileInputs) -> bool {
!contract_in_use(inputs)
}
#[cfg(test)]
mod tests {
use super::*;
use crate::ring::PeerKeyLocation;
use crate::ring::location::Distance;
use crate::transport::TransportKeypair;
use std::cmp::Ordering;
use std::net::SocketAddr;
fn upstream() -> Option<PeerKeyLocation> {
let addr: SocketAddr = "127.0.0.1:9001".parse().unwrap();
let pk = TransportKeypair::new().public().clone();
Some(PeerKeyLocation::new(pk, addr))
}
fn base() -> ReconcileInputs {
ReconcileInputs {
computed_upstream: None,
has_local_client: false,
has_farther_downstream_subscriber: false,
has_recent_local_client_access: false,
state_present: true,
is_subscribed: false,
is_advertised: false,
is_verified_root: false,
actively_acquiring: false,
}
}
#[test]
fn reconcile_table() {
use Action::*;
let cases: Vec<(&str, ReconcileInputs, Vec<Action>)> = vec![
(
"no state, not subscribed, even with demand+upstream ⇒ [] \
(initial acquisition is the op path's job, not reconcile)",
ReconcileInputs {
state_present: false,
has_local_client: true,
computed_upstream: upstream(),
..base()
},
vec![],
),
(
"truly idle: no demand, not subscribed, NOT advertised ⇒ [] \
(nothing to tear down)",
ReconcileInputs { ..base() },
vec![],
),
(
"P2-A: advertised, NOT subscribed, no demand ⇒ [Retract] \
(stale advertisement must be withdrawn)",
ReconcileInputs {
is_advertised: true,
..base()
},
vec![Retract],
),
(
"P2-A concrete: verified root that announced (advertised, not subscribed) \
loses its last client ⇒ [Retract]",
ReconcileInputs {
is_advertised: true,
is_verified_root: true,
..base()
},
vec![Retract],
),
(
"P2-A: advertised, not subscribed, no state, no demand ⇒ [Retract] \
(teardown needs no body)",
ReconcileInputs {
state_present: false,
is_advertised: true,
..base()
},
vec![Retract],
),
(
"H1: no state, SUBSCRIBED, demand gone, has upstream ⇒ collapse + unsubscribe \
(do NOT leak the upstream subscription)",
ReconcileInputs {
state_present: false,
is_subscribed: true,
computed_upstream: upstream(),
..base()
},
vec![Collapse, Unsubscribe],
),
(
"M3/L1: demand gone, subscribed, has upstream, advertised \
⇒ collapse + unsubscribe + retract",
ReconcileInputs {
computed_upstream: upstream(),
is_subscribed: true,
is_advertised: true,
..base()
},
vec![Collapse, Unsubscribe, Retract],
),
(
"demand gone, subscribed, has upstream, not advertised ⇒ collapse + unsubscribe",
ReconcileInputs {
computed_upstream: upstream(),
is_subscribed: true,
..base()
},
vec![Collapse, Unsubscribe],
),
(
"demand gone, subscribed, no upstream (root lapsing), advertised \
⇒ collapse + retract (no wire unsubscribe with no upstream)",
ReconcileInputs {
is_subscribed: true,
is_verified_root: true,
is_advertised: true,
..base()
},
vec![Collapse, Retract],
),
(
"steady-state host: upstream, subscribed, advertised, in use ⇒ [Renew]",
ReconcileInputs {
computed_upstream: upstream(),
has_local_client: true,
is_subscribed: true,
is_advertised: true,
..base()
},
vec![Renew],
),
(
"host-with-upstream, not subscribed, not advertised ⇒ subscribe + announce",
ReconcileInputs {
computed_upstream: upstream(),
has_local_client: true,
..base()
},
vec![Subscribe, Announce],
),
(
"host-with-upstream, not subscribed, already advertised ⇒ subscribe only",
ReconcileInputs {
computed_upstream: upstream(),
has_farther_downstream_subscriber: true,
is_advertised: true,
..base()
},
vec![Subscribe],
),
(
"host-with-upstream, subscribed, not advertised ⇒ renew + announce",
ReconcileInputs {
computed_upstream: upstream(),
has_local_client: true,
is_subscribed: true,
..base()
},
vec![Renew, Announce],
),
(
"verified root, in use, subscribed, advertised ⇒ [Renew] \
(no subscribe, no collapse while in use)",
ReconcileInputs {
has_local_client: true,
is_subscribed: true,
is_advertised: true,
is_verified_root: true,
..base()
},
vec![Renew],
),
(
"verified root, in use, subscribed, not advertised ⇒ renew + announce",
ReconcileInputs {
has_local_client: true,
is_subscribed: true,
is_verified_root: true,
..base()
},
vec![Renew, Announce],
),
(
"L1: verified root, in use, NOT subscribed, not advertised ⇒ [Announce] \
(root has body + demand, advertises; no lease to renew)",
ReconcileInputs {
has_local_client: true,
is_verified_root: true,
..base()
},
vec![Announce],
),
(
"re-root: had upstream (subscribed), upstream now None, demand intact, not root \
⇒ renew + re-root (serve-during: keep lease AND re-find, NOT collapse)",
ReconcileInputs {
has_local_client: true,
is_subscribed: true,
..base()
},
vec![Renew, ReRootSearch],
),
(
"re-root fresh: demand intact, no upstream, not subscribed, not root \
⇒ [ReRootSearch] (first formation / never rooted)",
ReconcileInputs {
has_farther_downstream_subscriber: true,
..base()
},
vec![ReRootSearch],
),
(
"P2-B: acquiring, in use, no upstream/root, not subscribed, state present, \
not advertised ⇒ [Announce] (body arrived, acquiring flag not yet cleared)",
ReconcileInputs {
has_local_client: true,
actively_acquiring: true,
..base()
},
vec![Announce],
),
(
"P2-B: acquiring + subscribed, in use, no upstream/root, state present, \
not advertised ⇒ [Renew, Announce]",
ReconcileInputs {
has_local_client: true,
is_subscribed: true,
actively_acquiring: true,
..base()
},
vec![Renew, Announce],
),
(
"M1: acquiring, in use, NO state yet, not subscribed ⇒ [] \
(never announce before the body arrives)",
ReconcileInputs {
state_present: false,
has_local_client: true,
actively_acquiring: true,
..base()
},
vec![],
),
];
for (name, inputs, expected) in cases {
assert_eq!(reconcile(&inputs), expected, "case: {name}");
}
}
#[test]
fn wants_renewal_drives_interest_gate() {
assert!(wants_renewal(&ReconcileInputs {
computed_upstream: upstream(),
has_local_client: true,
is_subscribed: true,
is_advertised: true,
..base()
}));
assert!(wants_renewal(&ReconcileInputs {
computed_upstream: upstream(),
has_local_client: true,
..base()
}));
assert!(wants_renewal(&ReconcileInputs {
has_local_client: true,
is_subscribed: true,
..base()
}));
assert!(wants_renewal(&ReconcileInputs {
has_farther_downstream_subscriber: true,
..base()
}));
assert!(wants_renewal(&ReconcileInputs {
has_local_client: true,
is_advertised: true,
is_verified_root: true,
..base()
}));
assert!(wants_renewal(&ReconcileInputs {
has_local_client: true,
is_verified_root: true,
..base()
}));
assert!(!wants_renewal(&ReconcileInputs {
computed_upstream: upstream(),
is_subscribed: true,
..base()
}));
assert!(!wants_renewal(&ReconcileInputs { ..base() }));
assert!(!wants_renewal(&ReconcileInputs {
is_advertised: true,
..base()
}));
}
#[test]
fn wants_collapse_is_exact_inverse_of_renewal() {
let snapshots = [
ReconcileInputs {
computed_upstream: upstream(),
has_local_client: true,
is_subscribed: true,
is_advertised: true,
..base()
},
ReconcileInputs {
has_farther_downstream_subscriber: true,
is_subscribed: true,
..base()
},
ReconcileInputs {
has_local_client: true,
is_verified_root: true,
..base()
},
ReconcileInputs {
computed_upstream: upstream(),
is_subscribed: true,
..base()
},
ReconcileInputs { ..base() },
ReconcileInputs {
is_advertised: true,
..base()
},
];
for s in &snapshots {
assert_eq!(
wants_collapse(s),
!wants_renewal(s),
"collapse gate must be the exact inverse of the renewal gate: {s:?}"
);
}
assert!(!wants_collapse(&ReconcileInputs {
has_local_client: true,
is_subscribed: true,
..base()
}));
assert!(!wants_collapse(&ReconcileInputs {
has_farther_downstream_subscriber: true,
is_subscribed: true,
..base()
}));
assert!(wants_collapse(&ReconcileInputs {
computed_upstream: upstream(),
is_subscribed: true,
is_advertised: true,
..base()
}));
}
#[test]
fn recent_local_access_alone_renews_not_collapses() {
let recent_access_only = ReconcileInputs {
has_recent_local_client_access: true,
has_local_client: false,
has_farther_downstream_subscriber: false,
..base()
};
assert!(
wants_renewal(&recent_access_only),
"recent local GET/PUT access alone must RENEW (invariant 3: reads/PUTs \
are permanent demand) — the P6 flip regression"
);
assert!(
!wants_collapse(&recent_access_only),
"recent local GET/PUT access alone must NOT collapse — collapse is the \
exact inverse of the demand-inclusive renewal gate"
);
assert!(contract_in_use(&recent_access_only));
assert!(!reconcile(&recent_access_only).contains(&Action::Collapse));
let idle = ReconcileInputs {
has_recent_local_client_access: false,
..base()
};
assert!(!wants_renewal(&idle));
assert!(wants_collapse(&idle));
}
#[test]
fn action_set_divergence_by_membership() {
use Action::*;
let d = action_set_divergence(&[Collapse, Unsubscribe], &[Collapse, Unsubscribe]);
assert!(!d.any(), "identical sets must not diverge");
let d = action_set_divergence(&[Unsubscribe, Collapse, Collapse], &[Collapse, Unsubscribe]);
assert!(!d.any(), "set membership ignores order and duplicates");
let d = action_set_divergence(&[Collapse, Unsubscribe, Retract], &[Collapse, Unsubscribe]);
assert_eq!(
d,
ReconcileActionDivergence {
retract: true,
..Default::default()
},
"only Retract should diverge (present in reconcile, absent in actual)"
);
let d = action_set_divergence(&[Subscribe], &[Renew]);
assert_eq!(
d,
ReconcileActionDivergence {
subscribe: true,
renew: true,
..Default::default()
}
);
let d = action_set_divergence(&[], &[Collapse]);
assert_eq!(
d,
ReconcileActionDivergence {
collapse: true,
..Default::default()
}
);
let d = action_set_divergence(&[Renew], &[Renew]);
assert!(!d.any());
}
#[test]
fn action_set_divergence_focused_ignores_irrelevant_classes() {
use Action::*;
let d = action_set_divergence_focused(&[Renew, ReRootSearch], &[], &[ReRootSearch]);
assert_eq!(
d,
ReconcileActionDivergence {
reroot_search: true,
..Default::default()
},
"focused compare must ignore the irrelevant Renew and flag only ReRootSearch"
);
let d = action_set_divergence_focused(&[Renew, Announce], &[Announce], &[Announce]);
assert!(
!d.any(),
"focused Announce compare agrees; the Renew is out of scope"
);
let d = action_set_divergence_focused(&[Renew, Collapse], &[], &[Collapse]);
assert_eq!(
d,
ReconcileActionDivergence {
collapse: true,
..Default::default()
}
);
}
#[test]
fn distance_partialeq_is_fuzzy_but_cmp_is_exact() {
let x = Distance::new(0.3);
let y = Distance::new(f64::from_bits(0.3_f64.to_bits() + 1));
assert_eq!(
x, y,
"epsilon PartialEq treats one-ULP-apart distances as equal"
);
assert_ne!(
x.cmp(&y),
Ordering::Equal,
"exact cmp does NOT — reconcile must never mix epsilon `==` with this ordering"
);
}
#[test]
fn reconcile_core_stays_pure() {
const SRC: &str = include_str!("reconcile.rs");
let prod = &SRC[..SRC.find("#[cfg(test)]").unwrap_or(SRC.len())];
for forbidden in ["fn drive", "fn apply_action", "fn apply_actions"] {
assert!(
!prod.contains(forbidden),
"reconcile.rs must not define `{forbidden}` — the pure decision core \
stays side-effect-free; drivers live at the call sites (the flip)"
);
}
}
}