cu29_logstream/
structured.rs

1//! Static structured-log fan-out at the local encoder's byte boundary.
2
3use bincode::{
4    Encode,
5    enc::{Encoder, EncoderImpl, write::Writer},
6    error::EncodeError,
7};
8use cu29_log::CuLogEntry;
9use cu29_traits::{CuResult, WriteStream};
10use cu29_unifiedlog::{LogStream, SectionStorage, UnifiedLogWrite};
11use std::{cell::RefCell, fmt::Debug};
12
13/// Statically composed destinations for one local serialization attempt.
14/// A local section rollover starts a fresh attempt; only successful local writes
15/// are committed. Destination overflow must not fail the local write.
16pub trait StructuredLogOutput: Debug + Send + Sync {
17    fn begin(&mut self);
18    fn write(&mut self, bytes: &[u8]);
19    fn finish(&mut self, success: bool);
20}
21
22impl StructuredLogOutput for () {
23    fn begin(&mut self) {}
24    fn write(&mut self, _: &[u8]) {}
25    fn finish(&mut self, _: bool) {}
26}
27impl<O: StructuredLogOutput> StructuredLogOutput for Option<O> {
28    fn begin(&mut self) {
29        if let Some(output) = self {
30            output.begin();
31        }
32    }
33    fn write(&mut self, bytes: &[u8]) {
34        if let Some(output) = self {
35            output.write(bytes);
36        }
37    }
38    fn finish(&mut self, success: bool) {
39        if let Some(output) = self {
40            output.finish(success);
41        }
42    }
43}
44impl<A: StructuredLogOutput, B: StructuredLogOutput> StructuredLogOutput for (A, B) {
45    fn begin(&mut self) {
46        self.0.begin();
47        self.1.begin();
48    }
49    fn write(&mut self, bytes: &[u8]) {
50        self.0.write(bytes);
51        self.1.write(bytes);
52    }
53    fn finish(&mut self, success: bool) {
54        self.0.finish(success);
55        self.1.finish(success);
56    }
57}
58
59/// Copies locally encoded bytes into bounded destination buffers during the
60/// existing serialization pass. Framing, hashing and FEC run on sender workers.
61/// Local errors propagate; streamed overflow leaves the local record intact.
62pub struct StructuredLogStream<S: SectionStorage, L: UnifiedLogWrite<S>, O> {
63    local: LogStream<S, L>,
64    outputs: O,
65}
66impl<S: SectionStorage, L: UnifiedLogWrite<S>, O: StructuredLogOutput>
67    StructuredLogStream<S, L, O>
68{
69    pub fn new(local: LogStream<S, L>, outputs: O) -> Self {
70        Self { local, outputs }
71    }
72}
73impl<S: SectionStorage, L: UnifiedLogWrite<S>, O: Debug> Debug for StructuredLogStream<S, L, O> {
74    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
75        f.debug_struct("StructuredLogStream")
76            .field("local", &self.local)
77            .field("outputs", &self.outputs)
78            .finish()
79    }
80}
81impl<S: SectionStorage, L: UnifiedLogWrite<S>, O: StructuredLogOutput> WriteStream<CuLogEntry>
82    for StructuredLogStream<S, L, O>
83{
84    fn log(&mut self, entry: &CuLogEntry) -> CuResult<()> {
85        let captured = CapturedEntry {
86            entry,
87            outputs: RefCell::new(&mut self.outputs),
88        };
89        let result = self.local.log(&captured);
90        self.outputs.finish(result.is_ok());
91        result
92    }
93    fn flush(&mut self) -> CuResult<()> {
94        <LogStream<S, L> as WriteStream<CuLogEntry>>::flush(&mut self.local)
95    }
96    fn last_log_bytes(&self) -> Option<usize> {
97        <LogStream<S, L> as WriteStream<CuLogEntry>>::last_log_bytes(&self.local)
98    }
99}
100
101struct CapturedEntry<'a, O> {
102    entry: &'a CuLogEntry,
103    outputs: RefCell<&'a mut O>,
104}
105impl<O: StructuredLogOutput> Encode for CapturedEntry<'_, O> {
106    fn encode<E: Encoder>(&self, encoder: &mut E) -> Result<(), EncodeError> {
107        let mut outputs = self.outputs.borrow_mut();
108        outputs.begin();
109        let config = *encoder.config();
110        let writer = CapturingWriter {
111            local: encoder.writer(),
112            outputs: &mut **outputs,
113        };
114        self.entry.encode(&mut EncoderImpl::new(writer, config))
115    }
116}
117struct CapturingWriter<'a, W, O> {
118    local: &'a mut W,
119    outputs: &'a mut O,
120}
121impl<W: Writer, O: StructuredLogOutput> Writer for CapturingWriter<'_, W, O> {
122    fn write(&mut self, bytes: &[u8]) -> Result<(), EncodeError> {
123        self.local.write(bytes)?;
124        self.outputs.write(bytes);
125        Ok(())
126    }
127}
128
129#[cfg(test)]
130mod tests {
131    use super::*;
132    use cu29_log::CuLogLevel;
133    use cu29_traits::UnifiedLogType;
134    use cu29_unifiedlog::{NoopLogger, NoopSectionStorage};
135    use std::sync::{Arc, Mutex};
136
137    #[derive(Debug, Default)]
138    struct Observation {
139        attempts: usize,
140        commits: usize,
141        bytes: Vec<u8>,
142    }
143    #[derive(Debug)]
144    struct Probe(Arc<Mutex<Observation>>);
145    impl StructuredLogOutput for Probe {
146        fn begin(&mut self) {
147            let mut seen = self.0.lock().unwrap();
148            seen.attempts += 1;
149            seen.bytes.clear();
150        }
151        fn write(&mut self, bytes: &[u8]) {
152            self.0.lock().unwrap().bytes.extend_from_slice(bytes);
153        }
154        fn finish(&mut self, success: bool) {
155            self.0.lock().unwrap().commits += usize::from(success);
156        }
157    }
158
159    #[test]
160    fn static_fanout_observes_one_native_serialization_and_preserves_byte_accounting() {
161        let first = Arc::new(Mutex::new(Observation::default()));
162        let second = Arc::new(Mutex::new(Observation::default()));
163        let local = LogStream::<NoopSectionStorage, _>::new(
164            UnifiedLogType::StructuredLogLine,
165            Arc::new(Mutex::new(NoopLogger::new())),
166            1024,
167        )
168        .unwrap();
169        let mut sink =
170            StructuredLogStream::new(local, (Probe(first.clone()), Some(Probe(second.clone()))));
171        let mut entry = CuLogEntry::new(47, CuLogLevel::Info);
172        entry.paramname_indexes.push(0);
173        entry.params.push(cu29_value::Value::U64(42));
174        let canonical = bincode::encode_to_vec(&entry, bincode::config::standard()).unwrap();
175        sink.log(&entry).unwrap();
176        assert_eq!(sink.last_log_bytes(), Some(canonical.len()));
177        for observation in [first, second] {
178            let seen = observation.lock().unwrap();
179            assert_eq!(seen.attempts, 1);
180            assert_eq!(seen.commits, 1);
181            assert_eq!(seen.bytes, canonical);
182        }
183    }
184    #[test]
185    fn section_rollover_commits_only_the_complete_retry() {
186        use cu29_unifiedlog::{UnifiedLogger, UnifiedLoggerBuilder};
187        let seen = Arc::new(Mutex::new(Observation::default()));
188        let logs = std::path::Path::new(env!("CARGO_MANIFEST_DIR"))
189            .join("../../examples/cu_logstream_demo/logs");
190        std::fs::create_dir_all(&logs).unwrap();
191        let path = logs.join(format!("structured-rollover-{}.copper", std::process::id()));
192        let UnifiedLogger::Write(logger) = UnifiedLoggerBuilder::new()
193            .file_base_name(&path)
194            .preallocated_size(1024 * 1024)
195            .write(true)
196            .create(true)
197            .build()
198            .unwrap()
199        else {
200            panic!("writer")
201        };
202        let local = LogStream::new(
203            UnifiedLogType::StructuredLogLine,
204            Arc::new(Mutex::new(logger)),
205            1024,
206        )
207        .unwrap();
208        let mut sink = StructuredLogStream::new(local, Probe(seen.clone()));
209        for id in 1..=100 {
210            let entry = CuLogEntry::new(id, CuLogLevel::Info);
211            sink.log(&entry).unwrap();
212            assert_eq!(
213                seen.lock().unwrap().bytes,
214                bincode::encode_to_vec(&entry, bincode::config::standard()).unwrap()
215            );
216        }
217        let seen = seen.lock().unwrap();
218        assert_eq!(seen.commits, 100);
219        assert!(seen.attempts > 100, "local sections never rolled over");
220    }
221}
cu29_logstream/structured.rs

cu29_logstream/
structured.rs
