cu29_logstream/
feedback.rs

1//! Optional advisory feedback. All accounting and adaptation run on stream workers.
2//! Sender and receiver must be built with matching feedback layouts.
3
4use crate::{Error, Result, StreamIdentity};
5use bincode::{Decode, Encode};
6use cu29_clock::{CuDuration, CuTime};
7use cu29_runtime::config::LogStreamFeedbackConfig;
8
9pub const FEEDBACK_BUFFER_BYTES: usize = 320;
10const MAGIC: &[u8; 4] = b"CUFB";
11const LOSS_MARGIN_BP: u64 = 200;
12const HEALTHY_REPORTS: u8 = 3;
13
14/// Local sender policy for feedback cadence, timeout, and FEC adaptation.
15#[derive(Clone, Copy, Debug, PartialEq, Eq, Encode, Decode)]
16pub struct FeedbackPolicy {
17    pub report_interval_ms: u32,
18    pub timeout_ms: u32,
19    pub adaptation: Option<AdaptationBounds>,
20}
21
22/// Return-report cadence and identity advertised to the receiver.
23#[derive(Clone, Copy, Debug, PartialEq, Eq, Encode, Decode)]
24pub struct FeedbackRequirements {
25    pub report_interval_ms: u32,
26    pub destination: [u8; 16],
27}
28
29#[derive(Clone, Copy, Debug, PartialEq, Eq, Encode, Decode)]
30pub struct AdaptationBounds {
31    pub min_repair_every_source_symbols: u16,
32    pub max_repair_every_source_symbols: u16,
33}
34
35impl From<&LogStreamFeedbackConfig> for FeedbackPolicy {
36    fn from(config: &LogStreamFeedbackConfig) -> Self {
37        Self {
38            report_interval_ms: config.report_interval_ms,
39            timeout_ms: config.timeout_ms,
40            adaptation: config.adaptation.map(|bounds| AdaptationBounds {
41                min_repair_every_source_symbols: bounds.min_repair_every_source_symbols,
42                max_repair_every_source_symbols: bounds.max_repair_every_source_symbols,
43            }),
44        }
45    }
46}
47
48impl FeedbackPolicy {
49    pub fn validate(self, baseline: u16) -> Result<()> {
50        if self.report_interval_ms == 0
51            || self.timeout_ms <= self.report_interval_ms
52            || self.adaptation.is_some_and(|b| {
53                b.min_repair_every_source_symbols == 0
54                    || b.min_repair_every_source_symbols > baseline
55                    || baseline > b.max_repair_every_source_symbols
56            })
57        {
58            return Err(Error::InvalidConfig(
59                "invalid feedback cadence or FEC bounds",
60            ));
61        }
62        Ok(())
63    }
64}
65
66/// Cumulative outcomes for a contiguous finalized source-symbol range. Initial
67/// unseen history and the still-live coding window are excluded. ESI wraps at u32.
68#[derive(Clone, Copy, Debug, Default, PartialEq, Eq, Encode, Decode)]
69pub struct SourceOutcomes {
70    pub first_esi: u32,
71    pub finalized: u64,
72    pub received: u64,
73    pub recovered: u64,
74    pub missing: u64,
75}
76
77/// Cumulative measurements since this receiver joined the destination. Taking
78/// deltas between accepted reports tolerates lost feedback datagrams.
79#[derive(Clone, Copy, Debug, Default, PartialEq, Eq, Encode, Decode)]
80pub struct ReceiverReport {
81    pub session_id: [u8; 16],
82    pub sender_id: u32,
83    pub destination: [u8; 16],
84    pub receiver_id: [u8; 16],
85    pub sequence: u64,
86    pub elapsed_us: u64,
87    pub received_bytes: u64,
88    pub received_packets: u64,
89    pub sources: SourceOutcomes,
90    pub invalid_packets: u64,
91    pub duplicate_packets: u64,
92    pub expired_records: u64,
93    pub buffered_records: u32,
94    pub record_capacity: u32,
95    pub latest_copperlist: Option<u64>,
96    pub request_recovery: bool,
97}
98
99/// Stable destination binding without transmitting or looking up strings on workers.
100pub fn destination_key(id: &str) -> [u8; 16] {
101    blake3::hash(id.as_bytes()).as_bytes()[..16]
102        .try_into()
103        .unwrap()
104}
105
106impl ReceiverReport {
107    fn valid(&self) -> bool {
108        self.elapsed_us > 0
109            && self.record_capacity > 0
110            && self.buffered_records <= self.record_capacity
111            && self
112                .sources
113                .received
114                .checked_add(self.sources.recovered)
115                .and_then(|n| n.checked_add(self.sources.missing))
116                == Some(self.sources.finalized)
117    }
118
119    /// Encodes one complete report for an integrity-protected packet transport.
120    pub fn encode_into(&self, output: &mut [u8]) -> Result<usize> {
121        if output.len() < FEEDBACK_BUFFER_BYTES {
122            return Err(Error::BufferTooSmall {
123                needed: FEEDBACK_BUFFER_BYTES,
124                available: output.len(),
125            });
126        }
127        if !self.valid() {
128            return Err(Error::InvalidConfig("invalid receiver report"));
129        }
130        output[..4].copy_from_slice(MAGIC);
131        let len = bincode::encode_into_slice(
132            self,
133            &mut output[4..FEEDBACK_BUFFER_BYTES],
134            bincode::config::standard().with_fixed_int_encoding(),
135        )
136        .map_err(|_| Error::InvalidConfig("feedback report exceeds packet capacity"))?
137            + 4;
138        Ok(len)
139    }
140
141    /// Decodes a complete report after transport integrity verification.
142    /// Reports may arrive late, out of order, or more than once; the feedback
143    /// controller checks their identity, sequence, and cumulative counters.
144    pub fn decode(packet: &[u8]) -> Result<Self> {
145        if packet.len() < MAGIC.len()
146            || packet.len() > FEEDBACK_BUFFER_BYTES
147            || &packet[..4] != MAGIC
148        {
149            return Err(Error::InvalidConfig("invalid feedback framing"));
150        }
151        let end = packet.len();
152        let (report, used): (Self, usize) = bincode::decode_from_slice(
153            &packet[4..end],
154            bincode::config::standard()
155                .with_fixed_int_encoding()
156                .with_limit::<FEEDBACK_BUFFER_BYTES>(),
157        )
158        .map_err(|_| Error::InvalidConfig("invalid feedback payload"))?;
159        if used != end - 4 || !report.valid() {
160            return Err(Error::InvalidConfig("invalid feedback counters"));
161        }
162        Ok(report)
163    }
164}
165
166/// A bounded ring distinguishes original arrivals from symbols recovered by FEC.
167/// Advancing past a symbol finalizes it once; late duplicates cannot heal history.
168pub(crate) struct SourceObserver<const W: usize> {
169    base: Option<u32>,
170    slots: [u8; W],
171    head: usize,
172    outcomes: SourceOutcomes,
173}
174impl<const W: usize> Default for SourceObserver<W> {
175    fn default() -> Self {
176        Self {
177            base: None,
178            slots: [0; W],
179            head: 0,
180            outcomes: SourceOutcomes::default(),
181        }
182    }
183}
184impl<const W: usize> SourceObserver<W> {
185    pub(crate) fn advance(&mut self, base: u32) {
186        let Some(previous) = self.base else {
187            self.base = Some(base);
188            self.outcomes.first_esi = base;
189            return;
190        };
191        let delta = base.wrapping_sub(previous);
192        if delta == 0 || delta >= 1 << 31 {
193            return;
194        }
195        // Long outages cost at most W slot visits, never work proportional to the outage.
196        for _ in 0..(delta as usize).min(W) {
197            match self.slots[self.head] {
198                1 => self.outcomes.received += 1,
199                2 => self.outcomes.recovered += 1,
200                _ => self.outcomes.missing += 1,
201            }
202            self.slots[self.head] = 0;
203            self.head = (self.head + 1) % W;
204        }
205        if delta as usize > W {
206            self.outcomes.missing += u64::from(delta) - W as u64;
207            self.head = (self.head + (delta as usize - W) % W) % W;
208        }
209        self.outcomes.finalized += u64::from(delta);
210        self.base = Some(base);
211    }
212    pub(crate) fn observe(&mut self, esi: u32, original: bool) {
213        let Some(base) = self.base else {
214            return;
215        };
216        let offset = esi.wrapping_sub(base) as usize;
217        if offset >= W {
218            return;
219        }
220        let slot = &mut self.slots[(self.head + offset) % W];
221        if original {
222            *slot = 1;
223        } else if *slot == 0 {
224            *slot = 2;
225        }
226    }
227    pub(crate) fn outcomes(&self) -> SourceOutcomes {
228        self.outcomes
229    }
230}
231
232#[derive(Clone, Copy, Debug, Default, PartialEq, Eq)]
233pub enum FeedbackState {
234    #[default]
235    Waiting,
236    Active,
237    Stale,
238}
239
240#[derive(Clone, Copy, Debug, Default, PartialEq, Eq)]
241pub struct FeedbackSnapshot {
242    pub state: FeedbackState,
243    pub accepted_reports: u64,
244    pub rejected_reports: u64,
245    pub invalid_reports: u64,
246    pub last_received: Option<CuTime>,
247    pub report: Option<ReceiverReport>,
248    pub effective_repair_every_source_symbols: u16,
249    pub baseline_repair_every_source_symbols: u16,
250    pub receiver_rates_available: bool,
251    pub source_metrics_available: bool,
252    pub receiver_bytes_per_second: u64,
253    pub receiver_packets_per_second: u64,
254    pub source_loss_basis_points: u16,
255    pub source_recovery_basis_points: u16,
256}
257
258/// Pure deterministic controller. It cannot alter the send budget or retained storage.
259pub struct FeedbackController {
260    policy: FeedbackPolicy,
261    identity: StreamIdentity,
262    destination: [u8; 16],
263    snapshot: FeedbackSnapshot,
264    last_step: CuTime,
265    smoothed_loss: Option<u64>,
266    healthy: u8,
267}
268impl FeedbackController {
269    pub fn new(
270        policy: FeedbackPolicy,
271        identity: StreamIdentity,
272        destination: [u8; 16],
273        baseline: u16,
274    ) -> Result<Self> {
275        policy.validate(baseline)?;
276        Ok(Self {
277            policy,
278            identity,
279            destination,
280            snapshot: FeedbackSnapshot {
281                effective_repair_every_source_symbols: baseline,
282                baseline_repair_every_source_symbols: baseline,
283                ..FeedbackSnapshot::default()
284            },
285            last_step: CuTime::default(),
286            smoothed_loss: None,
287            healthy: 0,
288        })
289    }
290    pub fn snapshot(&self) -> FeedbackSnapshot {
291        self.snapshot
292    }
293    pub fn invalid_report(&mut self) {
294        self.snapshot.invalid_reports += 1;
295    }
296    pub fn tick(&mut self, now: CuTime) {
297        if self
298            .snapshot
299            .last_received
300            .is_some_and(|last| elapsed(now, last) >= u64::from(self.policy.timeout_ms) * 1_000_000)
301        {
302            self.snapshot.state = FeedbackState::Stale;
303            self.healthy = 0;
304            self.smoothed_loss = None;
305            if elapsed(now, self.last_step) >= u64::from(self.policy.report_interval_ms) * 1_000_000
306            {
307                let baseline = self.snapshot.baseline_repair_every_source_symbols;
308                let current = &mut self.snapshot.effective_repair_every_source_symbols;
309                if *current < baseline {
310                    *current += 1;
311                } else if *current > baseline {
312                    *current -= 1;
313                }
314                self.last_step = now;
315            }
316        }
317    }
318    /// Returns whether an accepted report requests the existing retained recovery bundle.
319    pub fn receive(&mut self, report: ReceiverReport, now: CuTime) -> bool {
320        self.tick(now);
321        let previous = self.snapshot.report;
322        let same_peer = previous.is_some_and(|p| p.receiver_id == report.receiver_id);
323        let invalid = !report.valid()
324            || report.session_id != self.identity.session_id
325            || report.sender_id != self.identity.sender_id
326            || report.destination != self.destination
327            || previous.is_some_and(|p| {
328                if same_peer {
329                    report.sequence <= p.sequence
330                        || report.elapsed_us <= p.elapsed_us
331                        || (p.sources.finalized > 0
332                            && report.sources.first_esi != p.sources.first_esi)
333                        || report.sources.finalized < p.sources.finalized
334                        || report.sources.received < p.sources.received
335                        || report.sources.recovered < p.sources.recovered
336                        || report.sources.missing < p.sources.missing
337                        || report.received_bytes < p.received_bytes
338                        || report.received_packets < p.received_packets
339                } else {
340                    self.snapshot.state != FeedbackState::Stale
341                }
342            })
343            || self.snapshot.last_received.is_some_and(|last| {
344                self.snapshot.state != FeedbackState::Stale
345                    && elapsed(now, last) < u64::from(self.policy.report_interval_ms) * 500_000
346            });
347        if invalid {
348            self.snapshot.rejected_reports += 1;
349            return false;
350        }
351        // First report establishes a measurement baseline, including after receiver restart.
352        if let Some(previous) = previous.filter(|_| same_peer) {
353            let micros = report.elapsed_us - previous.elapsed_us;
354            self.snapshot.receiver_rates_available = true;
355            self.snapshot.receiver_bytes_per_second =
356                rate(report.received_bytes - previous.received_bytes, micros);
357            self.snapshot.receiver_packets_per_second =
358                rate(report.received_packets - previous.received_packets, micros);
359            let n = report.sources.finalized - previous.sources.finalized;
360            if n > 0 {
361                self.snapshot.source_metrics_available = true;
362                let lost = n.saturating_sub(report.sources.received - previous.sources.received);
363                let recovered = report.sources.recovered - previous.sources.recovered;
364                let loss = ((u128::from(lost) * 10_000) / u128::from(n)) as u64;
365                self.snapshot.source_loss_basis_points = loss as u16;
366                self.snapshot.source_recovery_basis_points = if lost == 0 {
367                    10_000
368                } else {
369                    ((u128::from(recovered) * 10_000) / u128::from(lost)).min(10_000) as u16
370                };
371                self.adapt(loss, report.sources.missing > previous.sources.missing);
372            }
373        } else {
374            self.healthy = 0;
375            self.smoothed_loss = None;
376            self.snapshot.receiver_rates_available = false;
377            self.snapshot.source_metrics_available = false;
378            self.snapshot.receiver_bytes_per_second = 0;
379            self.snapshot.receiver_packets_per_second = 0;
380            self.snapshot.source_loss_basis_points = 0;
381            self.snapshot.source_recovery_basis_points = 0;
382        }
383        self.snapshot.report = Some(report);
384        self.snapshot.last_received = Some(now);
385        self.snapshot.state = FeedbackState::Active;
386        self.snapshot.accepted_reports += 1;
387        self.last_step = now;
388        report.request_recovery
389    }
390    fn adapt(&mut self, loss: u64, unrecovered: bool) {
391        let Some(bounds) = self.policy.adaptation else {
392            return;
393        };
394        let smoothed = self.smoothed_loss.map_or(loss, |old| (old * 3 + loss) / 4);
395        self.smoothed_loss = Some(smoothed);
396        let target = ((10_000 - smoothed) / (smoothed + LOSS_MARGIN_BP)).clamp(
397            u64::from(bounds.min_repair_every_source_symbols),
398            u64::from(bounds.max_repair_every_source_symbols),
399        ) as u16;
400        let current = &mut self.snapshot.effective_repair_every_source_symbols;
401        if unrecovered {
402            *current = (*current / 2)
403                .min(target)
404                .max(bounds.min_repair_every_source_symbols);
405            self.healthy = 0;
406        } else if target < *current {
407            *current = target;
408            self.healthy = 0;
409        } else if target > *current {
410            self.healthy += 1;
411            if self.healthy >= HEALTHY_REPORTS {
412                *current += 1;
413                self.healthy = 0;
414            }
415        } else {
416            self.healthy = 0;
417        }
418    }
419}
420fn elapsed(now: CuTime, then: CuTime) -> u64 {
421    now.as_nanos().saturating_sub(then.as_nanos())
422}
423fn rate(count: u64, micros: u64) -> u64 {
424    (u128::from(count) * 1_000_000 / u128::from(micros.max(1))).min(u128::from(u64::MAX)) as u64
425}
426
427/// Receiver-side cadence and identity. The caller gathers counters from its router
428/// and attempts one datagram send; backpressure never queues feedback history.
429pub struct FeedbackReporter {
430    pub requirements: FeedbackRequirements,
431    identity: StreamIdentity,
432    destination: [u8; 16],
433    receiver_id: [u8; 16],
434    started: CuTime,
435    next_report: CuTime,
436    sequence: u64,
437}
438impl FeedbackReporter {
439    pub fn new(
440        manifest: &crate::SessionManifest,
441        receiver_id: [u8; 16],
442        now: CuTime,
443    ) -> Option<Self> {
444        let requirements = manifest.requirements.feedback?;
445        Some(Self {
446            requirements,
447            identity: manifest.identity,
448            destination: requirements.destination,
449            receiver_id,
450            started: now,
451            next_report: now + CuDuration::from_millis(u64::from(requirements.report_interval_ms)),
452            sequence: 0,
453        })
454    }
455    pub fn report(&mut self, now: CuTime, mut counters: ReceiverReport) -> Option<ReceiverReport> {
456        if now < self.next_report {
457            return None;
458        }
459        self.next_report =
460            now + CuDuration::from_millis(u64::from(self.requirements.report_interval_ms));
461        self.sequence = self.sequence.saturating_add(1);
462        counters.session_id = self.identity.session_id;
463        counters.sender_id = self.identity.sender_id;
464        counters.destination = self.destination;
465        counters.receiver_id = self.receiver_id;
466        counters.sequence = self.sequence;
467        counters.elapsed_us = (elapsed(now, self.started) / 1000).max(1);
468        Some(counters)
469    }
470}
471
472#[cfg(test)]
473mod tests {
474    use super::*;
475    #[test]
476    fn observer_handles_wrap_repair_then_original_and_long_outage() {
477        let mut observer = SourceObserver::<4>::default();
478        observer.advance(u32::MAX - 1);
479        observer.observe(u32::MAX - 1, true);
480        observer.observe(u32::MAX, false);
481        observer.observe(u32::MAX, true);
482        observer.observe(0, false);
483        observer.advance(2);
484        assert_eq!(
485            observer.outcomes(),
486            SourceOutcomes {
487                first_esi: u32::MAX - 1,
488                finalized: 4,
489                received: 2,
490                recovered: 1,
491                missing: 1,
492            }
493        );
494        observer.advance(1_000_002);
495        assert_eq!(observer.outcomes().finalized, 1_000_004);
496        assert_eq!(observer.outcomes().missing, 1_000_001);
497        observer.observe(1_000_002, true);
498        observer.advance(1_000_003);
499        assert_eq!(observer.outcomes().received, 3);
500    }
501}
cu29_logstream/feedback.rs

cu29_logstream/
feedback.rs
