cu29_logstream/
twin_session.rs

1//! Copper-owned receive, archive, and reconstruction lifecycle for one robot session.
2
3use crate::telemetry::{TelemetryReader, telemetry_channel};
4use crate::twin::{LiveReplay, TwinFrame, TwinReceiverStatus, TwinStatus, TwinWorker};
5use crate::{
6    CaptureArchive, CuStreamRx, SessionEvent, SessionRouter, SessionRouterLimits, StreamIdentity,
7};
8use cu29_traits::{CuError, CuResult};
9use cu29_unifiedlog::{UnifiedLogger, UnifiedLoggerBuilder};
10use std::marker::PhantomData;
11use std::num::NonZeroUsize;
12use std::path::{Path, PathBuf};
13use std::sync::Arc;
14use std::sync::atomic::{AtomicBool, Ordering};
15use std::thread::{self, JoinHandle};
16use std::time::{Duration, Instant};
17
18const REPLAY_CAPACITY: NonZeroUsize = NonZeroUsize::new(32).unwrap();
19const FRAME_CAPACITY: NonZeroUsize = NonZeroUsize::new(64).unwrap();
20const SLAB_BYTES: usize = 16 * 1024 * 1024;
21// Includes the native section header and continuity envelope around a finite record.
22const SECTION_BYTES: usize = 128 * 1024;
23const CONTINUITY_OVERHEAD: usize = 32;
24const POLL_INTERVAL: Duration = Duration::from_millis(1);
25
26/// Recording state is independent of reconstruction and display consumption.
27#[derive(Clone, Copy, Debug, Default, PartialEq, Eq)]
28pub enum CuTwinRecordingState {
29    #[default]
30    Waiting,
31    Recording,
32    Closed,
33    Failed,
34}
35
36/// Receiver and reconstruction health, available even when the view is paused.
37#[derive(Clone, Copy, Debug, Default, PartialEq, Eq)]
38pub struct CuTwinStatus {
39    pub latest: Option<u64>,
40    pub recovery_point: Option<u64>,
41    pub gaps: usize,
42    pub packets: usize,
43    pub archived: u64,
44    pub structured_logs: u64,
45    pub identity: Option<StreamIdentity>,
46    pub last_packet: Option<Instant>,
47    pub state: CuTwinRecordingState,
48    pub twin: TwinStatus,
49    pub feedback_reports_sent: u64,
50    pub feedback_reports_dropped: u64,
51    pub feedback_failed: bool,
52}
53
54impl TwinReceiverStatus for CuTwinStatus {
55    fn with_twin(mut self, status: TwinStatus) -> Self {
56        self.twin = status;
57        self
58    }
59}
60
61/// Builder for a generated application's live twin. Copper owns all worker,
62/// routing, archive, and recovery plumbing; callers supply a packet transport.
63/// Receiver bounds match the default 1200-byte-MTU, 64-symbol streaming profile.
64/// One handle accepts one sender session, with a fresh archive path.
65pub struct CuTwinBuilder<A, R> {
66    rx: R,
67    feedback_tx: Option<Box<dyn crate::CuFeedbackTx>>,
68    log_base: Option<PathBuf>,
69    log_rollover: Option<usize>,
70    frame_capacity: NonZeroUsize,
71    log_capacity: NonZeroUsize,
72    replay_capacity: NonZeroUsize,
73    slab_bytes: usize,
74    section_bytes: usize,
75    receiver_limits: SessionRouterLimits,
76    reconstruct: bool,
77    app: PhantomData<fn() -> A>,
78}
79
80impl<A: LiveReplay, R: CuStreamRx + 'static> CuTwinBuilder<A, R> {
81    /// Enable feedback only when the sender advertises support. The caller owns endpoint selection.
82    pub fn with_feedback<T: crate::CuFeedbackTx + 'static>(mut self, tx: T) -> Self {
83        self.feedback_tx = Some(Box::new(tx));
84        self
85    }
86
87    pub fn with_log_path(mut self, path: impl AsRef<Path>) -> Self {
88        self.log_base = Some(path.as_ref().to_path_buf());
89        self
90    }
91
92    /// Bound archive storage in bytes, overwriting the oldest data sections.
93    /// Static application metadata and the decode catalog remain available.
94    /// Capacity must exceed the configured section size and be a multiple of 512 bytes.
95    pub fn with_log_rollover(mut self, bytes: usize) -> Self {
96        self.log_rollover = Some(bytes);
97        self
98    }
99
100    /// Sets the size of each archive backing file (slab), in bytes. Defaults to 16 MiB.
101    ///
102    /// The archive adds another slab when the current one fills; this is not a
103    /// limit on total recording size. A slab must fit the largest section used
104    /// to store CopperLists, keyframes, or structured logs. Prefer a slab roughly
105    /// 10–100 times larger than a section to hold many sections per file and
106    /// reduce file turnover. Larger slabs reserve more disk space at a time.
107    /// Both slab and section sizes must be multiples of 512 bytes.
108    pub fn with_slab_size(mut self, bytes: usize) -> Self {
109        self.slab_bytes = bytes;
110        self
111    }
112
113    /// Sets the size of an archive section, in bytes. Defaults to 128 KiB.
114    ///
115    /// Sections group records such as CopperLists, keyframes, and structured log
116    /// entries inside a slab. A complete record must fit within one section.
117    /// Choose at least the larger of `max_record_bytes` and
118    /// `finite_objects.max_object_bytes` from [`Self::with_receiver_limits`],
119    /// plus 512 bytes for the section header and 32 bytes for the archive's
120    /// continuity envelope. Round up to a multiple of 512 bytes.
121    ///
122    /// Increase this alongside receive bounds when captures or task-state
123    /// snapshots grow. The section must fit in [`Self::with_slab_size`]; keeping
124    /// slabs much larger lets each file hold many sections.
125    pub fn with_section_size(mut self, bytes: usize) -> Self {
126        self.section_bytes = bytes;
127        self
128    }
129
130    /// Sets receive size and buffering limits for one sender session.
131    /// Defaults to [`SessionRouterLimits::default`].
132    ///
133    /// `max_record_bytes` bounds one complete serialized CopperList record
134    /// (default: 4 KiB). `finite_objects.max_object_bytes` bounds one manifest,
135    /// keyframe, or structured log record (default: 64 KiB). Match or exceed
136    /// the sender's advertised bounds so its captures and recovery snapshots
137    /// can be received, and increase [`Self::with_section_size`] to fit them.
138    ///
139    /// The remaining limits bound records, packets, recovery data, and objects
140    /// buffered during decoding. Larger sizes and concurrency limits increase
141    /// receiver memory use; they are not a total process memory budget.
142    /// `max_sessions` must be 1. The packet profile remains a 1200-byte MTU with
143    /// at most 1128-byte symbols and 64 FEC equations.
144    pub fn with_receiver_limits(mut self, limits: SessionRouterLimits) -> Self {
145        self.receiver_limits = limits;
146        self
147    }
148
149    /// Sets buffering between recording and task replay. Defaults to 32.
150    ///
151    /// This bounds both the queued replay events (captured CopperLists and
152    /// recovery keyframes) and the pending CopperLists waiting for matching
153    /// replay state. One recovery keyframe and one executing CopperList are
154    /// retained in addition to these buffers.
155    ///
156    /// Increase this to absorb short replay slowdowns; reduce it to retain fewer
157    /// payloads in memory. Overflow interrupts reconstruction until a matching
158    /// recovery point is available. Received data continues to be recorded in
159    /// the archive. This setting is unused with [`Self::archive_only`].
160    pub fn with_replay_capacity(mut self, capacity: NonZeroUsize) -> Self {
161        self.replay_capacity = capacity;
162        self
163    }
164
165    /// Sets the number of unread structured log entries buffered for
166    /// [`CuTwin::take_log_reader`]. Defaults to 64 entries.
167    ///
168    /// These are the robot's structured logging calls (such as `info!` and
169    /// `debug!`), published after being saved in the archive. Increase this to
170    /// let the log reader pause longer at the cost of retaining more entries in
171    /// memory. When full, the buffer replaces the oldest unread entry. Reading
172    /// slowly never blocks recording; overwritten entries remain in the archive.
173    pub fn with_log_capacity(mut self, capacity: NonZeroUsize) -> Self {
174        self.log_capacity = capacity;
175        self
176    }
177
178    /// Sets the number of unread reconstructed CopperLists buffered for the
179    /// returned [`CuTwinReader`]. Defaults to 64.
180    ///
181    /// Each [`TwinFrame`] contains one CopperList with its captured inputs and
182    /// locally reconstructed outputs, plus sender identity and receive time.
183    /// Increase this to absorb pauses in a UI or analysis reader; reduce it to
184    /// retain fewer message payloads in memory. When full, the buffer replaces
185    /// the oldest unread CopperList and the reader reports how many it missed.
186    /// Recording and task replay continue independently of reader consumption.
187    /// Network packets and recovery keyframes have separate receive/replay
188    /// buffers; this capacity counts completed CopperLists.
189    pub fn with_frame_capacity(mut self, capacity: NonZeroUsize) -> Self {
190        self.frame_capacity = capacity;
191        self
192    }
193
194    /// Record the native capture without starting task reconstruction.
195    pub fn archive_only(mut self) -> Self {
196        self.reconstruct = false;
197        self
198    }
199
200    pub fn spawn(self) -> CuResult<(CuTwin<A>, CuTwinReader<A>)> {
201        validate_sizes(self.slab_bytes, self.section_bytes, self.receiver_limits)?;
202        // Validate and construct routing before creating directories or starting workers.
203        let mut router =
204            SessionRouter::<1128, 64, 64>::new(self.receiver_limits).map_err(stream_error)?;
205        if self
206            .log_rollover
207            .is_some_and(|bytes| bytes <= self.section_bytes || !bytes.is_multiple_of(512))
208        {
209            return Err(CuError::from(
210                "Twin log rollover capacity must exceed the configured section size and be a multiple of 512 bytes",
211            ));
212        }
213        let path = self
214            .log_base
215            .ok_or_else(|| CuError::from("Copper twin requires a log path"))?;
216        if cu29_runtime::replay::first_slab_path(&path)?.exists() || path.exists() {
217            return Err(CuError::from(
218                "Copper twin archive already exists; choose a fresh log path",
219            ));
220        }
221        if let Some(parent) = path
222            .parent()
223            .filter(|parent| !parent.as_os_str().is_empty())
224        {
225            std::fs::create_dir_all(parent)
226                .map_err(|e| CuError::new_with_cause("Create twin log directory", e))?;
227        }
228        let (publisher, reader) = telemetry_channel(self.frame_capacity, CuTwinStatus::default());
229        let (mut log_publisher, log_reader) = telemetry_channel(self.log_capacity, ());
230        let stop = Arc::new(AtomicBool::new(false));
231        let stopping = stop.clone();
232        let (ready_tx, ready_rx) = std::sync::mpsc::sync_channel(1);
233        let worker = thread::Builder::new()
234            .name("copper-twin-receiver".into())
235            .spawn(move || {
236                let mut status = CuTwinStatus::default();
237                let (replay, mut publisher) = if self.reconstruct {
238                    match TwinWorker::spawn::<A>(self.replay_capacity, publisher) {
239                        Ok(replay) => (Some(replay), None),
240                        Err(error) => {
241                            let _ = ready_tx.send(Err(error.clone()));
242                            return Err(error);
243                        }
244                    }
245                } else {
246                    (None, Some(publisher))
247                };
248                let mut publish_status = |status| {
249                    if let Some(replay) = &replay {
250                        replay.set_status(status);
251                    }
252                    if let Some(publisher) = &mut publisher {
253                        publisher.set_status(status);
254                    }
255                };
256                let mut archive = None;
257                let result = (|| {
258                    let _ = ready_tx.send(Ok(()));
259                    let mut rx = self.rx;
260                    let mut feedback_tx = self.feedback_tx;
261                    let feedback_clock = cu29_clock::RobotClock::new();
262                    let receiver_id = crate::new_session_id();
263                    let mut reporter = None;
264                    let mut feedback_packet = [0; crate::feedback::FEEDBACK_BUFFER_BYTES];
265                    let mut packet = [0; 1200];
266                    while !stopping.load(Ordering::Acquire) {
267                        if let Some(len) = rx
268                            .try_recv(&mut packet)
269                            .map_err(|e| CuError::from(format!("Twin receive: {e:?}")))?
270                        {
271                            let bytes = packet.get(..len).ok_or_else(|| {
272                                CuError::from("Transport returned an invalid packet length")
273                            })?;
274                            status.last_packet = Some(Instant::now());
275                            router
276                                .receive_datagram(bytes, |event| {
277                                    // Recovery objects can arrive before the manifest.
278                                    // The router retains them and emits VerifiedRecoveryPoint
279                                    // once their manifest/keyframe references agree.
280                                    if matches!(&event, SessionEvent::Object { record, .. } if record.decoded().kind != crate::RecordKind::StructuredLog) {
281                                        return Ok(());
282                                    }
283                                    if let SessionEvent::Manifest(manifest) = &event {
284                                        if manifest.manifest().context.mission_index != A::MISSION_INDEX {
285                                            return Err(crate::Error::InvalidConfig("twin requires the matching sender mission"));
286                                        }
287                                        status.identity = Some(manifest.manifest().identity);
288                                        if feedback_tx.is_some() {
289                                            reporter = crate::feedback::FeedbackReporter::new(
290                                                manifest.manifest(),
291                                                receiver_id,
292                                                feedback_clock.now(),
293                                            );
294                                        }
295                                        let mut builder = UnifiedLoggerBuilder::new()
296                                            .file_base_name(&path)
297                                            .preallocated_size(self.log_rollover.unwrap_or(self.slab_bytes))
298                                            .write(true)
299                                            .create(true);
300                                        if let Some(bytes) = self.log_rollover {
301                                            builder = builder.rollover(bytes);
302                                        }
303                                        let logger = builder.build()
304                                            .map_err(|e| crate::Error::Codec(e.to_string()))?;
305                                        let UnifiedLogger::Write(mut logger) = logger else {
306                                            unreachable!()
307                                        };
308                                        A::seal_archive_metadata(&mut logger)
309                                            .map_err(|e| crate::Error::Codec(e.to_string()))?;
310                                        archive = Some(CaptureArchive::<A::DataSet>::new_with_logger(
311                                            manifest, self.section_bytes, logger,
312                                        )?);
313                                    }
314                                    let writer =
315                                        archive.as_mut().ok_or(crate::Error::InvalidConfig(
316                                            "capture arrived before its manifest",
317                                        ))?;
318                                    let capture = writer.accept(&event)?;
319                                    if let Some(entry) = writer.take_structured_log() {
320                                        status.structured_logs += 1;
321                                        if let SessionEvent::Object { identity, record } = &event {
322                                            log_publisher.publish(ReceivedStructuredLog {
323                                                identity: *identity, sequence: record.decoded().object_id, entry,
324                                            });
325                                        }
326                                    }
327                                    if let Some(capture) = &capture {
328                                        status.latest = Some(capture.copperlist.id);
329                                        status.archived += 1;
330                                        status.state = CuTwinRecordingState::Recording;
331                                    }
332                                    if let Some(replay) = &replay {
333                                        replay.accept(&event, capture)?;
334                                    }
335                                    match event {
336                                        SessionEvent::Gap { .. } => status.gaps += 1,
337                                        SessionEvent::VerifiedRecoveryPoint {
338                                            recovery_point,
339                                            ..
340                                        } => {
341                                            status.recovery_point =
342                                                Some(recovery_point.copperlist_id)
343                                        }
344                                        _ => {}
345                                    }
346                                    Ok::<(), crate::Error>(())
347                                })
348                                .map_err(|e| CuError::from(format!("Twin receive: {e}")))?;
349                            status.packets = router.stats().datagrams_seen;
350                            publish_status(status);
351                        } else {
352                            thread::park_timeout(POLL_INTERVAL);
353                        }
354                        if let (Some(reporter), Some(tx), Some(identity)) =
355                            (&mut reporter, &mut feedback_tx, status.identity)
356                            && let Some(mut counters) = router.feedback_counters(identity)
357                        {
358                            counters.latest_copperlist = status.latest;
359                            if let Some(report) = reporter.report(feedback_clock.now(), counters) {
360                                let len = report
361                                    .encode_into(&mut feedback_packet)
362                                    .map_err(stream_error)?;
363                                match tx.try_send_feedback(&feedback_packet[..len]) {
364                                    Ok(()) => status.feedback_reports_sent += 1,
365                                    Err(crate::CuStreamTxError::WouldBlock) => {
366                                        status.feedback_reports_dropped += 1
367                                    }
368                                    Err(crate::CuStreamTxError::Failed(_)) => {
369                                        feedback_tx = None;
370                                        status.feedback_failed = true;
371                                    }
372                                }
373                                publish_status(status);
374                            }
375                        }
376                    }
377                    if let Some(archive) = archive.take() {
378                        archive.finish().map_err(stream_error)?;
379                    }
380                    Ok(())
381                })();
382                status.state = if result.is_ok() {
383                    CuTwinRecordingState::Closed
384                } else {
385                    CuTwinRecordingState::Failed
386                };
387                publish_status(status);
388                // Drain admitted replay before returning the final receiver status.
389                if let Some(replay) = replay {
390                    status.twin = replay.finish()?;
391                }
392                result.map(|()| status)
393            })
394            .map_err(|e| CuError::new_with_cause("Start Copper twin", e))?;
395        if let Err(error) = ready_rx
396            .recv()
397            .map_err(|_| CuError::from("Copper twin initialization failed"))
398            .and_then(|ready| ready)
399        {
400            stop.store(true, Ordering::Release);
401            let _ = worker.join();
402            return Err(error);
403        }
404        Ok((
405            CuTwin {
406                stop,
407                worker: Some(worker),
408                final_status: None,
409                log_reader: Some(log_reader),
410                app: PhantomData,
411            },
412            reader,
413        ))
414    }
415}
416
417fn stream_error(error: crate::Error) -> CuError {
418    CuError::from(error.to_string())
419}
420
421/// Reader for a generated graph's reconstructed frames and recording status.
422pub type CuTwinReader<A> = TelemetryReader<TwinFrame<<A as LiveReplay>::DataSet>, CuTwinStatus>;
423
424/// One original robot log entry, published only after native archival succeeds.
425#[derive(Debug)]
426pub struct ReceivedStructuredLog {
427    pub identity: StreamIdentity,
428    pub sequence: u64,
429    pub entry: cu29_log::CuLogEntry,
430}
431
432/// Independent bounded log reader; a stalled reader never delays native recording.
433pub type CuTwinLogReader = TelemetryReader<ReceivedStructuredLog, ()>;
434
435fn validate_sizes(
436    slab_bytes: usize,
437    section_bytes: usize,
438    limits: SessionRouterLimits,
439) -> CuResult<()> {
440    let header = usize::from(cu29_unifiedlog::SECTION_HEADER_COMPACT_SIZE);
441    if slab_bytes < 1024 || !slab_bytes.is_multiple_of(header) {
442        return Err(CuError::from(
443            "Twin slab size must be at least 1024 bytes and a multiple of 512",
444        ));
445    }
446    if section_bytes <= header
447        || !section_bytes.is_multiple_of(header)
448        || section_bytes > slab_bytes
449    {
450        return Err(CuError::from(
451            "Twin section size must exceed 512 bytes, be a multiple of 512, and fit in its slab",
452        ));
453    }
454    if section_bytes - header > u32::MAX as usize {
455        return Err(CuError::from("Twin section payload size exceeds u32"));
456    }
457    if limits.max_sessions != 1 {
458        return Err(CuError::from(
459            "Twin receiver limits must allow exactly one sender session",
460        ));
461    }
462    if limits.equation_capacity > 64 || usize::from(limits.finite_objects.max_symbol_size) > 1128 {
463        return Err(CuError::from(
464            "Twin receiver limits exceed the 1128-byte-symbol, 64-equation FEC profile",
465        ));
466    }
467    crate::object::validate_limits(limits.finite_objects).map_err(stream_error)?;
468    if limits.max_record_bytes > u32::MAX as usize {
469        return Err(CuError::from("Twin maximum record size exceeds u32"));
470    }
471    let object_bytes = usize::try_from(limits.finite_objects.max_object_bytes)
472        .map_err(|_| CuError::from("Twin maximum object size exceeds usize"))?;
473    let required = limits
474        .max_record_bytes
475        .max(object_bytes)
476        .checked_add(CONTINUITY_OVERHEAD)
477        .and_then(|bytes| bytes.checked_add(header))
478        .ok_or_else(|| CuError::from("Twin section size requirement overflow"))?;
479    if section_bytes < required {
480        return Err(CuError::from(format!(
481            "Twin section size is {section_bytes} bytes; configured record/object bounds require at least {required} bytes including header and continuity envelope"
482        )));
483    }
484    Ok(())
485}
486
487/// Running Copper twin. The separately owned reader can be paused or dropped
488/// without affecting recording. `stop()` closes the archive and drains admitted
489/// replay; dropping the handle also stops and joins Copper's workers.
490pub struct CuTwin<A: LiveReplay> {
491    stop: Arc<AtomicBool>,
492    worker: Option<JoinHandle<CuResult<CuTwinStatus>>>,
493    final_status: Option<CuTwinStatus>,
494    log_reader: Option<CuTwinLogReader>,
495    app: PhantomData<fn() -> A>,
496}
497
498impl<A: LiveReplay> CuTwin<A> {
499    pub fn builder<R: CuStreamRx + 'static>(rx: R) -> CuTwinBuilder<A, R> {
500        CuTwinBuilder {
501            rx,
502            feedback_tx: None,
503            log_base: None,
504            log_rollover: None,
505            frame_capacity: FRAME_CAPACITY,
506            log_capacity: FRAME_CAPACITY,
507            replay_capacity: REPLAY_CAPACITY,
508            slab_bytes: SLAB_BYTES,
509            section_bytes: SECTION_BYTES,
510            receiver_limits: SessionRouterLimits::default(),
511            reconstruct: true,
512            app: PhantomData,
513        }
514    }
515
516    /// Take the independent bounded log view once (64 entries by default).
517    /// Entries retain sender string IDs; use the producing application's string index to render text.
518    pub fn take_log_reader(&mut self) -> Option<CuTwinLogReader> {
519        self.log_reader.take()
520    }
521
522    pub fn stop(&mut self) -> CuResult<CuTwinStatus> {
523        self.stop.store(true, Ordering::Release);
524        if let Some(worker) = self.worker.take() {
525            worker.thread().unpark();
526            self.final_status = Some(
527                worker
528                    .join()
529                    .map_err(|_| CuError::from("Copper twin receiver panicked"))??,
530            );
531        }
532        self.final_status
533            .ok_or_else(|| CuError::from("Copper twin stopped after a worker failure"))
534    }
535}
536
537impl<A: LiveReplay> Drop for CuTwin<A> {
538    fn drop(&mut self) {
539        let _ = self.stop();
540    }
541}
cu29_logstream/twin_session.rs

cu29_logstream/
twin_session.rs
