cu29_logstream/
archive.rs

1//! Application-typed native archival. All work here is receiver-side.
2
3use crate::capture::CapturedList;
4use crate::{
5    ApplicationSchema, Error, ReceivedManifest, RecordKind, Result, SessionEvent, SessionEventRef,
6    StreamIdentity,
7};
8use bincode::{
9    Encode,
10    enc::{Encoder, write::Writer},
11    error::EncodeError,
12};
13use cu29_runtime::{
14    continuity::{SourceGapReason, StreamContinuityRecord},
15    copperlist::CopperList,
16};
17use cu29_traits::{CopperListTuple, UnifiedLogType, WriteStream};
18use cu29_unifiedlog::{
19    LogStream, UnifiedLogger, UnifiedLoggerBuilder, UnifiedLoggerWrite, memmap::MmapSectionStorage,
20};
21use std::{
22    marker::PhantomData,
23    path::Path,
24    sync::{Arc, Mutex},
25};
26
27type ArchiveDecoder<P> = fn(&[u8]) -> Result<(CapturedList<P>, &[u8])>;
28
29type NativeStream = LogStream<MmapSectionStorage, UnifiedLoggerWrite>;
30
31/// One application-typed archive per `(session id, sender id)`. Pass all ordered
32/// router events for that sender to `accept`. The expected schema comes from
33/// `P`'s generated output specs, independently of the remote manifest.
34/// `CaptureArchive` applies the generated selective codec for a live twin.
35/// A write failure poisons the writer: do not retry a partially committed event.
36pub struct NativeArchive<P: CopperListTuple> {
37    copperlists: NativeStream,
38    keyframes: NativeStream,
39    continuity: NativeStream,
40    structured: NativeStream,
41    last_structured: Option<cu29_log::CuLogEntry>,
42    identity: StreamIdentity,
43    manifest_digest: [u8; 32],
44    manifest_object_id: u64,
45    next_id: u64,
46    poisoned: bool,
47    payload: PhantomData<P>,
48    decode: ArchiveDecoder<P>,
49}
50
51struct CanonicalEntry<'a>(&'a [u8]);
52impl Encode for CanonicalEntry<'_> {
53    fn encode<E: Encoder>(&self, encoder: &mut E) -> core::result::Result<(), EncodeError> {
54        encoder.writer().write(self.0)
55    }
56}
57
58impl<P: CopperListTuple> NativeArchive<P> {
59    /// Creates native message, task state, continuity and structured log sections.
60    /// The section size must fit the largest accepted canonical entry.
61    pub fn new(
62        path: &Path,
63        received: &ReceivedManifest,
64        slab_bytes: usize,
65        section_bytes: usize,
66    ) -> Result<Self> {
67        let expected = ApplicationSchema::from_output_specs(P::get_output_specs());
68        Self::new_checked(
69            path,
70            received,
71            slab_bytes,
72            section_bytes,
73            expected,
74            |bytes| Ok((CapturedList::new(crate::decode_copperlist(bytes)?), bytes)),
75        )
76    }
77
78    fn new_checked(
79        path: &Path,
80        received: &ReceivedManifest,
81        slab_bytes: usize,
82        section_bytes: usize,
83        expected_schema: ApplicationSchema,
84        decode: ArchiveDecoder<P>,
85    ) -> Result<Self> {
86        let manifest = received.manifest();
87        manifest.requirements.validate()?;
88        if manifest.application_schema != expected_schema {
89            return Err(Error::InvalidConfig(
90                "archive requires the matching application schema",
91            ));
92        }
93        if section_bytes == 0 || section_bytes > slab_bytes {
94            return Err(Error::InvalidConfig("archive section must fit in its slab"));
95        }
96        let logger = UnifiedLoggerBuilder::new()
97            .file_base_name(path)
98            .preallocated_size(slab_bytes)
99            .write(true)
100            .create(true)
101            .build()
102            .map_err(io_error)?;
103        let UnifiedLogger::Write(logger) = logger else {
104            unreachable!()
105        };
106        Self::new_with_logger(received, section_bytes, expected_schema, decode, logger)
107    }
108
109    fn new_with_logger(
110        received: &ReceivedManifest,
111        section_bytes: usize,
112        expected_schema: ApplicationSchema,
113        decode: ArchiveDecoder<P>,
114        logger: UnifiedLoggerWrite,
115    ) -> Result<Self> {
116        let manifest = received.manifest();
117        manifest.requirements.validate()?;
118        if manifest.application_schema != expected_schema {
119            return Err(Error::InvalidConfig(
120                "archive requires the matching application schema",
121            ));
122        }
123        let logger = Arc::new(Mutex::new(logger));
124        let mut archive = Self {
125            last_structured: None,
126            copperlists: NativeStream::with_context(
127                UnifiedLogType::CopperList,
128                logger.clone(),
129                section_bytes,
130                manifest.context,
131            )
132            .map_err(io_error)?,
133            keyframes: NativeStream::with_context(
134                UnifiedLogType::FrozenTasks,
135                logger.clone(),
136                section_bytes,
137                manifest.context,
138            )
139            .map_err(io_error)?,
140            structured: NativeStream::with_context(
141                UnifiedLogType::StructuredLogLine,
142                logger.clone(),
143                section_bytes,
144                manifest.context,
145            )
146            .map_err(io_error)?,
147            continuity: NativeStream::with_context(
148                UnifiedLogType::StreamContinuity,
149                logger,
150                section_bytes,
151                manifest.context,
152            )
153            .map_err(io_error)?,
154            identity: manifest.identity,
155            manifest_digest: received.record().decoded().digest,
156            manifest_object_id: received.record().decoded().object_id,
157            next_id: 0,
158            poisoned: false,
159            payload: PhantomData,
160            decode,
161        };
162        archive
163            .continuity
164            .log(&StreamContinuityRecord::Manifest {
165                record: received.record().bytes(),
166            })
167            .map_err(io_error)?;
168        Ok(archive)
169    }
170
171    /// Validates a typed CopperList once, appends its original canonical bytes,
172    /// and returns the typed value for an in-process consumer. Keyframes are
173    /// archived only after the router has verified their recovery point references.
174    pub fn accept(&mut self, event: &SessionEventRef<'_>) -> Result<Option<CopperList<P>>> {
175        self.accept_capture(event)
176            .map(|capture| capture.map(|c| c.copperlist))
177    }
178
179    fn accept_capture(&mut self, event: &SessionEventRef<'_>) -> Result<Option<CapturedList<P>>> {
180        self.last_structured = None;
181        if self.poisoned {
182            return Err(Error::InvalidConfig(
183                "archive failed; close it before continuing",
184            ));
185        }
186        let result = self.accept_inner(event);
187        if result.is_err() {
188            self.poisoned = true;
189        }
190        result
191    }
192
193    fn accept_inner(&mut self, event: &SessionEventRef<'_>) -> Result<Option<CapturedList<P>>> {
194        let identity = match event {
195            SessionEvent::Manifest(manifest) => manifest.manifest().identity,
196            SessionEvent::ContinuousRecord { identity, .. }
197            | SessionEvent::Gap { identity, .. }
198            | SessionEvent::Object { identity, .. }
199            | SessionEvent::VerifiedRecoveryPoint { identity, .. } => *identity,
200        };
201        if identity != self.identity {
202            return Err(Error::InconsistentObject);
203        }
204        match event {
205            SessionEvent::Manifest(manifest) => {
206                if manifest.record().decoded().digest != self.manifest_digest
207                    || manifest.record().decoded().object_id != self.manifest_object_id
208                {
209                    return Err(Error::InconsistentObject);
210                }
211            }
212            SessionEvent::ContinuousRecord { record, .. } => {
213                let decoded = record.decoded();
214                if decoded.kind != RecordKind::CopperList || decoded.object_id != self.next_id {
215                    return Err(Error::InconsistentObject);
216                }
217                let (capture, native) = (self.decode)(decoded.payload)?;
218                let copperlist = &capture.copperlist;
219                if copperlist.id != decoded.object_id {
220                    return Err(Error::InconsistentObject);
221                }
222                let next = self
223                    .next_id
224                    .checked_add(1)
225                    .ok_or(Error::InconsistentObject)?;
226                self.copperlists
227                    .log(&CanonicalEntry(native))
228                    .map_err(io_error)?;
229                self.next_id = next;
230                return Ok(Some(capture));
231            }
232            SessionEvent::Gap { gap, .. } => {
233                if gap.first_id != self.next_id || gap.last_id < gap.first_id {
234                    return Err(Error::InconsistentObject);
235                }
236                let next = gap
237                    .last_id
238                    .checked_add(1)
239                    .ok_or(Error::InconsistentObject)?;
240                let reason = match gap.reason {
241                    crate::GapReason::RlcWindowExpired => SourceGapReason::RlcWindowExpired,
242                    crate::GapReason::SessionEnded => SourceGapReason::SessionEnded,
243                    crate::GapReason::LateJoin => SourceGapReason::LateJoin,
244                    crate::GapReason::RecoveryPoint => SourceGapReason::RecoveryPoint,
245                };
246                self.continuity
247                    .log(&StreamContinuityRecord::<&[u8]>::Gap {
248                        first_id: gap.first_id,
249                        last_id: gap.last_id,
250                        reason,
251                    })
252                    .map_err(io_error)?;
253                self.next_id = next;
254            }
255            SessionEvent::VerifiedRecoveryPoint {
256                recovery_point,
257                recovery_record,
258                keyframe,
259                ..
260            } => {
261                let decoded = keyframe.decoded();
262                if recovery_point.manifest_record_digest != self.manifest_digest
263                    || recovery_point.manifest_object_id != self.manifest_object_id
264                    || !recovery_point.references_keyframe(decoded)
265                    || keyframe.keyframe_id() != Some(recovery_point.copperlist_id)
266                    || recovery_record.decoded().kind != RecordKind::RecoveryPoint
267                    || recovery_record.decoded().object_id != recovery_point.copperlist_id
268                    || crate::decode_recovery_point(recovery_record.decoded().payload)?
269                        != *recovery_point
270                {
271                    return Err(Error::InconsistentObject);
272                }
273                self.keyframes
274                    .log(&CanonicalEntry(decoded.payload))
275                    .map_err(io_error)?;
276                self.continuity
277                    .log(&StreamContinuityRecord::RecoveryPoint {
278                        copperlist_id: recovery_point.copperlist_id,
279                        record: recovery_record.bytes(),
280                    })
281                    .map_err(io_error)?;
282            }
283            SessionEvent::Object { record, .. } => {
284                let decoded = record.decoded();
285                if decoded.kind == RecordKind::StructuredLog {
286                    let (entry, used): (cu29_log::CuLogEntry, usize) = bincode::decode_from_slice(
287                        decoded.payload,
288                        bincode::config::standard().with_limit::<4_194_304>(),
289                    )
290                    .map_err(|_| Error::InvalidConfig("invalid structured log entry"))?;
291                    if used != decoded.payload.len() {
292                        return Err(Error::InvalidConfig("trailing structured log bytes"));
293                    }
294                    self.structured
295                        .log(&CanonicalEntry(decoded.payload))
296                        .map_err(io_error)?;
297                    self.last_structured = Some(entry);
298                }
299            }
300        }
301        Ok(None)
302    }
303
304    /// Move the structured entry decoded during the most recent successful accept.
305    /// Call before accepting another event; publication can reuse this owned value.
306    pub fn take_structured_log(&mut self) -> Option<cu29_log::CuLogEntry> {
307        self.last_structured.take()
308    }
309
310    /// Records the receiver's final known boundary and closes native sections.
311    /// The router must first be finalized through any caller-known sender tail.
312    pub fn finish(mut self) -> Result<()> {
313        if self.poisoned {
314            return Err(Error::InvalidConfig("archive failed before finalization"));
315        }
316        self.continuity
317            .log(&StreamContinuityRecord::<&[u8]>::Finished {
318                next_copperlist_id: self.next_id,
319            })
320            .map_err(io_error)
321    }
322}
323
324fn io_error(error: impl core::fmt::Display) -> Error {
325    Error::Codec(error.to_string())
326}
327
328/// Capture-aware native archive. Synthesized payloads never enter this writer.
329/// Native CopperList metadata retains both original and captured presence planes;
330/// StreamContinuity retains the reconstruction contract and source gaps.
331pub struct CaptureArchive<P: crate::capture::CaptureDataSet>(NativeArchive<P>);
332impl<P: crate::capture::CaptureDataSet> CaptureArchive<P> {
333    /// Open capture streams after the application has sealed static metadata.
334    #[doc(hidden)]
335    pub fn new_with_logger(
336        received: &ReceivedManifest,
337        section_bytes: usize,
338        logger: UnifiedLoggerWrite,
339    ) -> Result<Self> {
340        Ok(Self(NativeArchive::new_with_logger(
341            received,
342            section_bytes,
343            P::stream_schema(),
344            crate::capture::decode_capture,
345            logger,
346        )?))
347    }
348
349    pub fn new(
350        path: &Path,
351        received: &ReceivedManifest,
352        slab_bytes: usize,
353        section_bytes: usize,
354    ) -> Result<Self> {
355        Ok(Self(NativeArchive::new_checked(
356            path,
357            received,
358            slab_bytes,
359            section_bytes,
360            P::stream_schema(),
361            crate::capture::decode_capture,
362        )?))
363    }
364    pub fn accept(
365        &mut self,
366        event: &SessionEventRef<'_>,
367    ) -> Result<Option<crate::capture::CapturedList<P>>> {
368        self.0.accept_capture(event)
369    }
370
371    /// Move the structured entry from the most recent successful accept.
372    pub fn take_structured_log(&mut self) -> Option<cu29_log::CuLogEntry> {
373        self.0.take_structured_log()
374    }
375
376    pub fn finish(self) -> Result<()> {
377        self.0.finish()
378    }
379}
cu29_logstream/archive.rs

cu29_logstream/
archive.rs
