cu29_logstream/
twin.rs

1//! Ordered, bounded live replay, independent of archive writes and presentation loss.
2use crate::{
3    StreamIdentity,
4    capture::{CaptureDataSet, CapturedList},
5    telemetry::TelemetryPublisher,
6};
7use crossbeam_queue::ArrayQueue;
8use cu29_clock::RobotClockMock;
9use cu29_runtime::{copperlist::CopperList, curuntime::KeyFrame};
10use cu29_traits::{CopperListTuple, CuResult};
11use std::{
12    num::NonZeroUsize,
13    sync::{
14        Arc,
15        atomic::{AtomicU64, Ordering},
16        mpsc::{self, SyncSender, TrySendError},
17    },
18    thread::{self, JoinHandle},
19    time::{Duration, Instant},
20};
21
22/// Implemented by `#[copper_runtime(..., sim_mode = true)]` with logstream enabled.
23pub trait LiveReplay: Sized + 'static {
24    type DataSet: CaptureDataSet + Send + 'static;
25    /// Index into the generated application's shared mission catalog.
26    #[doc(hidden)]
27    const MISSION_INDEX: u32;
28    /// Seal locally generated application metadata and the optional decode catalog.
29    #[doc(hidden)]
30    fn seal_archive_metadata(logger: &mut cu29_unifiedlog::UnifiedLoggerWrite) -> CuResult<()>;
31    fn build_twin() -> CuResult<(Self, RobotClockMock)>;
32    fn stop_twin(&mut self) -> CuResult<()> {
33        Ok(())
34    }
35    fn replay_capture(
36        &mut self,
37        clock: &RobotClockMock,
38        capture: &mut CopperList<Self::DataSet>,
39        keyframe: Option<&KeyFrame>,
40    ) -> CuResult<()>;
41}
42
43#[derive(Clone, Copy, Debug, Default, PartialEq, Eq)]
44pub enum ReconstructionState {
45    #[default]
46    Waiting,
47    Recovering,
48    Reconstructed,
49    Verified,
50    Diverged,
51}
52
53#[derive(Clone, Copy, Debug, Default, PartialEq, Eq)]
54pub struct TwinStatus {
55    pub state: ReconstructionState,
56    pub reconstructed: u64,
57    pub verified: u64,
58    pub divergences: u64,
59    pub queue_overflows: u64,
60}
61
62/// Compose receiver/archive status with independently available reconstruction health.
63pub trait TwinReceiverStatus: Copy + PartialEq + Send + 'static {
64    fn with_twin(self, status: TwinStatus) -> Self;
65}
66
67pub struct TwinFrame<P: CopperListTuple> {
68    pub identity: StreamIdentity,
69    pub received_at: Instant,
70    pub copperlist: CopperList<P>,
71}
72
73/// Synchronous replay engine, also usable for ordinary offline capture replay.
74pub struct LiveTwin<A: LiveReplay> {
75    app: A,
76    clock: RobotClockMock,
77    next: Option<u64>,
78    pub status: TwinStatus,
79}
80impl<A: LiveReplay> LiveTwin<A> {
81    pub fn new() -> CuResult<Self> {
82        let (app, clock) = A::build_twin()?;
83        Ok(Self {
84            app,
85            clock,
86            next: None,
87            status: TwinStatus::default(),
88        })
89    }
90    pub fn recover(&mut self) {
91        self.next = None;
92        self.status.state = ReconstructionState::Recovering;
93    }
94    pub fn reconstruct(
95        &mut self,
96        mut capture: CapturedList<A::DataSet>,
97        keyframe: Option<&KeyFrame>,
98    ) -> CuResult<Option<CopperList<A::DataSet>>> {
99        let id = capture.copperlist.id;
100        if self.next != Some(id) && keyframe.is_none_or(|k| k.culistid != id) {
101            if self.status.state != ReconstructionState::Diverged {
102                self.recover();
103            }
104            return Ok(None);
105        }
106        let result = self
107            .app
108            .replay_capture(&self.clock, &mut capture.copperlist, keyframe);
109        #[cfg(feature = "verify-reconstruction")]
110        let result = result.and_then(|()| {
111            capture
112                .verify()
113                .map_err(|e| cu29_traits::CuError::from(e.to_string()))
114        });
115        if let Err(error) = result {
116            self.next = None;
117            self.status.state = ReconstructionState::Diverged;
118            self.status.divergences += 1;
119            return Err(error);
120        }
121        self.next = id.checked_add(1);
122        self.status.state = ReconstructionState::Reconstructed;
123        #[cfg(feature = "verify-reconstruction")]
124        if capture.digest.is_some() {
125            self.status.state = ReconstructionState::Verified;
126        }
127        self.status.reconstructed += 1;
128        if self.status.state == ReconstructionState::Verified {
129            self.status.verified += 1;
130        }
131        Ok(Some(capture.copperlist))
132    }
133}
134
135impl<A: LiveReplay> Drop for LiveTwin<A> {
136    fn drop(&mut self) {
137        let _ = self.app.stop_twin();
138    }
139}
140
141enum Input<P: CopperListTuple> {
142    Capture {
143        capture: CapturedList<P>,
144        identity: StreamIdentity,
145        received_at: Instant,
146    },
147    RecoveryPoint(KeyFrame),
148}
149struct Work<P: CopperListTuple> {
150    input: Input<P>,
151    generation: u64,
152}
153struct Pending<P: CopperListTuple> {
154    capture: CapturedList<P>,
155    identity: StreamIdentity,
156    received_at: Instant,
157}
158
159/// Nonblocking archive-to-replay handoff. The worker consumes every admitted
160/// iteration; only successfully reconstructed results enter the UI ring.
161/// Payload verification is opt-in through `verify-reconstruction`.
162/// Storage bounds are `capacity` queued events, `capacity` pending captures,
163/// one pending recovery point and one executing frame, in addition to the telemetry ring.
164/// Overflow invalidates replay until an admitted matching recovery point, never archival.
165#[doc(hidden)]
166pub struct TwinWorker<P: CopperListTuple, S> {
167    sender: Option<SyncSender<Work<P>>>,
168    statuses: Arc<ArrayQueue<S>>,
169    generation: Arc<AtomicU64>,
170    overflows: Arc<AtomicU64>,
171    worker: Option<JoinHandle<TwinStatus>>,
172}
173
174impl<P: CaptureDataSet + Send + 'static, S: TwinReceiverStatus> TwinWorker<P, S> {
175    pub fn spawn<A: LiveReplay<DataSet = P>>(
176        capacity: NonZeroUsize,
177        mut publisher: TelemetryPublisher<TwinFrame<P>, S>,
178    ) -> CuResult<Self> {
179        let (sender, receiver) = mpsc::sync_channel::<Work<P>>(capacity.get());
180        let statuses = Arc::new(ArrayQueue::new(1));
181        let pending_status = statuses.clone();
182        let generation = Arc::new(AtomicU64::new(0));
183        let epoch = generation.clone();
184        let overflows = Arc::new(AtomicU64::new(0));
185        let drops = overflows.clone();
186        let (ready_tx, ready_rx) = mpsc::sync_channel(1);
187        let worker = thread::Builder::new()
188            .name("copper-live-twin".into())
189            .spawn(move || {
190                let mut twin = match LiveTwin::<A>::new() {
191                    Ok(twin) => {
192                        let _ = ready_tx.send(Ok(()));
193                        twin
194                    }
195                    Err(e) => {
196                        let _ = ready_tx.send(Err(e));
197                        return TwinStatus::default();
198                    }
199                };
200                let mut current_generation = 0;
201                let mut pending: std::collections::VecDeque<Pending<P>> =
202                    std::collections::VecDeque::with_capacity(capacity.get());
203                let mut recovery_point: Option<KeyFrame> = None;
204                let mut receiver_status = publisher.status();
205                loop {
206                    if let Some(status) = pending_status.pop() {
207                        receiver_status = status;
208                    }
209                    let latest_generation = epoch.load(Ordering::Acquire);
210                    if current_generation != latest_generation {
211                        twin.recover();
212                        pending.clear();
213                        recovery_point = None;
214                        current_generation = latest_generation;
215                    }
216                    twin.status.queue_overflows = drops.load(Ordering::Relaxed);
217                    publisher.set_status(receiver_status.with_twin(twin.status));
218                    match receiver.recv_timeout(Duration::from_millis(10)) {
219                        Ok(work) => {
220                            let latest = epoch.load(Ordering::Acquire);
221                            if current_generation != latest {
222                                twin.recover();
223                                pending.clear();
224                                recovery_point = None;
225                                current_generation = latest;
226                            }
227                            if work.generation != current_generation {
228                                continue;
229                            }
230                            match work.input {
231                                Input::Capture {
232                                    capture,
233                                    identity,
234                                    received_at,
235                                } => {
236                                    if pending.len() == capacity.get() {
237                                        pending.pop_front();
238                                        drops.fetch_add(1, Ordering::Relaxed);
239                                        twin.recover();
240                                    }
241                                    pending.push_back(Pending {
242                                        capture,
243                                        identity,
244                                        received_at,
245                                    });
246                                }
247                                Input::RecoveryPoint(frame) => {
248                                    if recovery_point
249                                        .as_ref()
250                                        .is_none_or(|old| old.culistid < frame.culistid)
251                                    {
252                                        recovery_point = Some(frame);
253                                    }
254                                }
255                            }
256                            while let Some(first) = pending.front() {
257                                if twin.next != Some(first.capture.copperlist.id) {
258                                    let Some(frame) = &recovery_point else {
259                                        break;
260                                    };
261                                    let Some(position) = pending.iter().position(|item| {
262                                        item.capture.copperlist.id == frame.culistid
263                                    }) else {
264                                        break;
265                                    };
266                                    pending.drain(..position);
267                                }
268                                let item = pending.pop_front().unwrap();
269                                let keyframe = if recovery_point.as_ref().is_some_and(|frame| {
270                                    frame.culistid == item.capture.copperlist.id
271                                }) {
272                                    recovery_point.take()
273                                } else {
274                                    None
275                                };
276                                if let Ok(Some(copperlist)) =
277                                    twin.reconstruct(item.capture, keyframe.as_ref())
278                                    && epoch.load(Ordering::Acquire) == current_generation
279                                {
280                                    publisher.publish(TwinFrame {
281                                        identity: item.identity,
282                                        received_at: item.received_at,
283                                        copperlist,
284                                    });
285                                }
286                            }
287                        }
288                        Err(mpsc::RecvTimeoutError::Timeout) => {}
289                        Err(mpsc::RecvTimeoutError::Disconnected) => break,
290                    }
291                }
292                if let Some(status) = pending_status.pop() {
293                    receiver_status = status;
294                }
295                publisher.set_status(receiver_status.with_twin(twin.status));
296                twin.status
297            })
298            .map_err(|e| cu29_traits::CuError::new_with_cause("spawn live twin", e))?;
299        ready_rx
300            .recv()
301            .map_err(|_| cu29_traits::CuError::from("live twin initialization failed"))??;
302        Ok(Self {
303            sender: Some(sender),
304            statuses,
305            generation,
306            overflows,
307            worker: Some(worker),
308        })
309    }
310    pub fn finish(mut self) -> CuResult<TwinStatus> {
311        self.sender.take();
312        self.worker
313            .take()
314            .ok_or_else(|| cu29_traits::CuError::from("Twin worker already joined"))?
315            .join()
316            .map_err(|_| cu29_traits::CuError::from("Twin replay worker panicked"))
317    }
318
319    /// Route an already archived event into live replay. This is the usual
320    /// ground-station integration: `twin.accept(event, archive.accept(event)?)`.
321    /// Archival must succeed first; replay never retries or delays that writer.
322    pub fn accept(
323        &self,
324        event: &crate::SessionEventRef<'_>,
325        capture: Option<CapturedList<P>>,
326    ) -> crate::Result<()> {
327        match event {
328            crate::SessionEvent::Manifest(_) | crate::SessionEvent::Gap { .. } => self.recover(),
329            crate::SessionEvent::VerifiedRecoveryPoint { keyframe, .. } => {
330                self.recovery_point(crate::decode_keyframe(keyframe.decoded().payload)?)
331            }
332            crate::SessionEvent::ContinuousRecord { identity, .. } => {
333                if let Some(capture) = capture {
334                    self.submit(capture, *identity, Instant::now());
335                }
336            }
337            _ => {}
338        }
339        Ok(())
340    }
341    pub fn set_status(&self, status: S) {
342        self.statuses.force_push(status);
343    }
344    pub fn recover(&self) {
345        self.generation.fetch_add(1, Ordering::AcqRel);
346    }
347    pub fn submit(&self, capture: CapturedList<P>, identity: StreamIdentity, received_at: Instant) {
348        self.send(Input::Capture {
349            capture,
350            identity,
351            received_at,
352        });
353    }
354    /// Accept only keyframes from the session router's VerifiedRecoveryPoint event.
355    /// Control objects may arrive before or after their captured boundary.
356    pub fn recovery_point(&self, frame: KeyFrame) {
357        self.send(Input::RecoveryPoint(frame));
358    }
359    fn send(&self, input: Input<P>) {
360        let work = Work {
361            input,
362            generation: self.generation.load(Ordering::Acquire),
363        };
364        if let Err(TrySendError::Full(_)) = self.sender.as_ref().unwrap().try_send(work) {
365            self.overflows.fetch_add(1, Ordering::Relaxed);
366            self.recover();
367        }
368    }
369}
370impl<P: CopperListTuple, S> Drop for TwinWorker<P, S> {
371    fn drop(&mut self) {
372        self.sender.take();
373        if let Some(worker) = self.worker.take() {
374            let _ = worker.join();
375        }
376    }
377}
cu29_logstream/twin.rs

cu29_logstream/
twin.rs
