ozzy_replication/flow/probe.rs
1//! Small status retries only. No timer here schedules payloads or supplies votes.
2
3use std::time::Duration;
4
5use ozzy_proto::RequestId;
6
7use super::prefix_valid;
8use crate::{Digest, OpNumber, Prefix, Scope};
9
10/// One immutable status request, including the tail needed to detect a lost last batch.
11#[derive(Debug, Clone, Copy, PartialEq, Eq)]
12pub struct Probe {
13 /// Exact normal scope; this request never changes a view or authorizes recovery.
14 pub scope: Scope,
15 /// Unique exchange identity, frozen across retries and retired after completion.
16 pub request_id: RequestId,
17 /// Primary's outstanding tail at request creation, not a commit assertion.
18 pub tail: Prefix,
19 /// Highest locally available operation, including unsent publications.
20 /// Availability hint only, never a repair or commit boundary.
21 pub available: OpNumber,
22}
23
24/// Independent status-query pacing, not election or durable-progress timing.
25#[derive(Debug, Clone, Copy, PartialEq, Eq)]
26pub struct ProbeTiming {
27 /// Delay before the first retry and between unchanged completed exchanges.
28 pub initial: Duration,
29 /// Maximum retry delay after repeated unanswered status requests.
30 pub maximum: Duration,
31}
32
33/// Rejected probe transition. Errors preserve the pending request and deadlines.
34#[derive(Debug, Clone, Copy, PartialEq, Eq, thiserror::Error)]
35pub enum ProbeError {
36 /// Zero identity, malformed scope/tail, or invalid initial/maximum delay.
37 #[error("invalid replica probe parameters")]
38 Invalid,
39 /// Retire old correlation explicitly before polling another view/group.
40 #[error("replica probe scope changed without invalidation")]
41 Scope,
42 /// Adapter-supplied monotonic time moved backward.
43 #[error("replica probe clock regressed")]
44 Clock,
45 /// A retry/completion deadline cannot be represented; no wraparound is allowed.
46 #[error("replica probe deadline exhausted")]
47 TimeExhausted,
48 /// The local request-identity sequence cannot advance; never reuse retired IDs.
49 #[error("replica probe request identities exhausted")]
50 IdsExhausted,
51}
52
53/// Allocation-free, one-outstanding-request scheduler for one authenticated peer.
54///
55/// Poll only when the adapter needs receipt status. Opening is immediate;
56/// unanswered small requests back off to an explicit ceiling. Repeated calls at
57/// one instant emit at most one pending request. New work after a completed
58/// exchange opens immediately; unchanged status retains its pacing delay.
59/// A fresh valid response resets pacing.
60/// No observation changes durable ACKs, quorum state, or commit-progress deadlines.
61#[derive(Debug)]
62pub struct ProbeScheduler {
63 timing: ProbeTiming,
64 next_id: u128,
65 now: Duration,
66 due: Duration,
67 delay: Duration,
68 pending: Option<Probe>,
69 advertised: OpNumber,
70}
71
72impl ProbeScheduler {
73 /// Start a peer-local request sequence and clock without allocating or reading time.
74 ///
75 /// Supply a fresh process/session-specific starting ID, never a reused constant
76 /// across actor lifetimes. Keep this scheduler across view/receive-epoch changes;
77 /// `invalidate` retires correlation without resetting the identity sequence.
78 pub fn new(first: RequestId, timing: ProbeTiming, now: Duration) -> Result<Self, ProbeError> {
79 let next_id = u128::from_be_bytes(*first.as_bytes());
80 if next_id == 0 || timing.initial.is_zero() || timing.maximum < timing.initial {
81 return Err(ProbeError::Invalid);
82 }
83 now.checked_add(timing.maximum)
84 .ok_or(ProbeError::TimeExhausted)?;
85 Ok(Self {
86 timing,
87 next_id,
88 now,
89 due: now,
90 delay: timing.initial,
91 pending: None,
92 advertised: OpNumber(0),
93 })
94 }
95
96 /// Currently outstanding frozen request, for exact response/history-work matching.
97 pub const fn pending(&self) -> Option<Probe> {
98 self.pending
99 }
100
101 /// Emit a new small probe when due, or retry the exact pending request.
102 ///
103 /// Changed `tail` or `available` does not mutate a live exchange. After
104 /// completion new availability bypasses the idle delay and captures both.
105 /// A returned request remains owned
106 /// here even if transport admission fails; retry keeps its request identity.
107 pub fn poll(
108 &mut self,
109 scope: Scope,
110 tail: Prefix,
111 available: OpNumber,
112 now: Duration,
113 ) -> Result<Option<Probe>, ProbeError> {
114 self.check_time(now)?;
115 if scope.group_id.as_bytes() == &[0; 16]
116 || scope.configuration_digest == Digest::ZERO
117 || !prefix_valid(tail)
118 || tail.op.0 == u64::MAX
119 || available < tail.op
120 || available.0 == u64::MAX
121 {
122 return Err(ProbeError::Invalid);
123 }
124 if self.pending.is_some_and(|request| request.scope != scope) {
125 return Err(ProbeError::Scope);
126 }
127 let new_work = self.pending.is_none() && available > self.advertised;
128 if now < self.due && !new_work {
129 self.now = now;
130 return Ok(None);
131 }
132 let (request, next_id) = if let Some(request) = self.pending {
133 (request, self.next_id)
134 } else {
135 let next_id = self
136 .next_id
137 .checked_add(1)
138 .ok_or(ProbeError::IdsExhausted)?;
139 (
140 Probe {
141 scope,
142 tail,
143 available,
144
145 request_id: RequestId::from_bytes(self.next_id.to_be_bytes()),
146 },
147 next_id,
148 )
149 };
150 let due = now
151 .checked_add(self.delay)
152 .ok_or(ProbeError::TimeExhausted)?;
153 self.now = now;
154 self.due = due;
155 self.delay = self.delay.saturating_mul(2).min(self.timing.maximum);
156 self.next_id = next_id;
157 self.pending = Some(request);
158 self.advertised = request.available;
159 Ok(Some(request))
160 }
161
162 /// Retire a validated response only if its scope and ID name the live request.
163 ///
164 /// Call after authenticated decoding and required report/history checks, not
165 /// on arbitrary network activity. Stale replies do not extend retry deadlines.
166 /// This method does not decide whether a newly reported receive epoch is valid.
167 pub fn complete(
168 &mut self,
169 scope: Scope,
170 id: RequestId,
171 now: Duration,
172 ) -> Result<bool, ProbeError> {
173 let Some(due) = self.completion_deadline(scope, id, now)? else {
174 self.now = now;
175 return Ok(false);
176 };
177 self.now = now;
178 self.due = due;
179 self.delay = self.timing.initial;
180 self.pending = None;
181 Ok(true)
182 }
183
184 pub(super) fn completion_deadline(
185 &self,
186 scope: Scope,
187 id: RequestId,
188 now: Duration,
189 ) -> Result<Option<Duration>, ProbeError> {
190 self.check_time(now)?;
191 if self
192 .pending
193 .is_none_or(|request| request.scope != scope || request.request_id != id)
194 {
195 return Ok(None);
196 }
197 now.checked_add(self.timing.initial)
198 .map(Some)
199 .ok_or(ProbeError::TimeExhausted)
200 }
201
202 /// Retire correlation after role/session replacement, keeping identities monotonic.
203 /// The next poll can open immediately; old responses cannot complete that request.
204 pub fn invalidate(&mut self, now: Duration) -> Result<(), ProbeError> {
205 self.check_time(now)?;
206 self.now = now;
207 self.due = now;
208 self.delay = self.timing.initial;
209 self.pending = None;
210 Ok(())
211 }
212
213 /// Retry the current immutable exchange sooner after a new receiver hint.
214 /// Preserve its identity, frozen tail, and any correlated history lookup.
215 pub(super) fn expedite(&mut self, now: Duration) -> Result<(), ProbeError> {
216 self.check_time(now)?;
217 self.now = now;
218 self.due = self.due.min(now);
219 self.delay = self.timing.initial;
220 Ok(())
221 }
222
223 fn check_time(&self, now: Duration) -> Result<(), ProbeError> {
224 if now < self.now {
225 Err(ProbeError::Clock)
226 } else {
227 Ok(())
228 }
229 }
230}