cu29_logstream/
worker.rs

1//! Host driver for the autonomous sender. Only semantic output workers call these sinks.
2
3use crate::feedback::{
4    FEEDBACK_BUFFER_BYTES, FeedbackController, FeedbackSnapshot, ReceiverReport,
5};
6use crate::{
7    CuFeedbackRx, CuStreamRxError, CuStreamTx, LogStreamSenderConfig, OneWay, SenderCore,
8    SenderStats,
9};
10use cu29_clock::{CuDuration, CuTime, RobotClock};
11#[allow(unused_imports)]
12use cu29_log::{ANONYMOUS, CuLogEntry, CuLogLevel};
13#[allow(unused_imports)]
14use cu29_log_runtime::log;
15#[cfg(debug_assertions)]
16#[allow(unused_imports)]
17use cu29_log_runtime::log_debug_mode;
18use cu29_runtime::{copperlist::CopperList, curuntime::KeyFrame};
19use cu29_traits::{CopperListTuple, CuError, CuResult, WriteStream};
20#[allow(unused_imports)]
21use cu29_value::to_value;
22use std::{
23    fmt::{Debug, Formatter},
24    marker::PhantomData,
25    sync::{
26        Arc, Mutex,
27        atomic::{AtomicBool, AtomicU64, Ordering},
28        mpsc::{self, Receiver, SyncSender, TryRecvError},
29    },
30    thread::{JoinHandle, Thread},
31};
32
33const CL_BUFFERS: usize = 4;
34const KF_BUFFERS: usize = 2;
35const STRUCTURED_BUFFERS: usize = 4;
36use crate::pacing::STRUCTURED_RECORD_BYTES;
37
38struct StructuredRecord {
39    bytes: Box<[u8]>,
40    len: usize,
41    queued_at: CuTime,
42}
43
44struct StructuredInbox {
45    free: crossbeam_queue::ArrayQueue<StructuredRecord>,
46    pending: crossbeam_queue::ArrayQueue<StructuredRecord>,
47}
48
49/// A bounded destination for bytes produced by the local structured-log encoder.
50/// It does not own the sender thread: dropping the CL and keyframe sinks still
51/// drains and joins that worker, even while the global logger holds this sink.
52pub struct ScheduledStructuredLogSink {
53    inbox: Arc<StructuredInbox>,
54    status: Arc<Shared>,
55    thread: Thread,
56    clock: RobotClock,
57    current: Option<StructuredRecord>,
58    oversized: bool,
59}
60
61impl Debug for ScheduledStructuredLogSink {
62    fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result {
63        f.debug_struct("ScheduledStructuredLogSink")
64            .finish_non_exhaustive()
65    }
66}
67
68impl crate::structured::StructuredLogOutput for ScheduledStructuredLogSink {
69    fn begin(&mut self) {
70        if self.current.is_none()
71            && !self.status.stop.load(Ordering::Acquire)
72            && !self.status.failed.load(Ordering::Acquire)
73        {
74            self.current = self.inbox.free.pop();
75        }
76        if let Some(record) = &mut self.current {
77            record.len = crate::record::RECORD_HEADER_LEN;
78        }
79        self.oversized = false;
80    }
81
82    fn write(&mut self, bytes: &[u8]) {
83        if let Some(record) = &mut self.current
84            && !self.oversized
85        {
86            if bytes.len() > record.bytes.len() - record.len {
87                self.oversized = true;
88            } else {
89                record.bytes[record.len..record.len + bytes.len()].copy_from_slice(bytes);
90                record.len += bytes.len();
91            }
92        }
93    }
94
95    fn finish(&mut self, success: bool) {
96        let record = self.current.take();
97        if !success
98            || self.oversized
99            || self.status.stop.load(Ordering::Acquire)
100            || self.status.failed.load(Ordering::Acquire)
101        {
102            if let Some(record) = record {
103                let _ = self.inbox.free.push(record);
104            }
105            if success {
106                self.status.inbox_drops.fetch_add(1, Ordering::Relaxed);
107            }
108        } else if let Some(mut record) = record {
109            record.queued_at = self.clock.now();
110            // Every owned buffer has a slot in the bounded pending queue.
111            if let Err(record) = self.inbox.pending.push(record) {
112                let _ = self.inbox.free.push(record);
113                self.status.inbox_drops.fetch_add(1, Ordering::Relaxed);
114            } else {
115                self.thread.unpark();
116            }
117        } else {
118            self.status.inbox_drops.fetch_add(1, Ordering::Relaxed);
119        }
120    }
121}
122
123struct Record {
124    bytes: Box<[u8]>,
125    len: usize,
126    keyframe: bool,
127    queued_at: CuTime,
128}
129
130/// Live worker snapshot. Publication uses try_lock at 10 Hz, never on the RT path.
131#[derive(Clone, Copy, Debug, Default)]
132pub struct SenderSnapshot {
133    pub sampled_at: CuTime,
134    pub stats: SenderStats,
135    pub inbox_drops: u64,
136    pub stopped: bool,
137    pub failed: bool,
138    pub feedback_failed: bool,
139    pub feedback: Option<FeedbackSnapshot>,
140}
141
142/// Final counters remain readable after both output sinks have been dropped.
143#[derive(Clone, Debug)]
144pub struct SenderMonitor(Arc<Shared>);
145
146#[derive(Debug, Default)]
147struct Shared {
148    stop: AtomicBool,
149    failed: AtomicBool,
150    inbox_drops: AtomicU64,
151    final_stats: Mutex<Option<SenderStats>>,
152    snapshot: Mutex<SenderSnapshot>,
153}
154
155impl SenderMonitor {
156    pub fn snapshot(&self) -> SenderSnapshot {
157        let mut snapshot = *self.0.snapshot.lock().expect("sender snapshot poisoned");
158        snapshot.failed |= self.failed();
159        snapshot
160    }
161
162    pub fn inbox_drops(&self) -> u64 {
163        self.0.inbox_drops.load(Ordering::Relaxed)
164    }
165    pub fn failed(&self) -> bool {
166        self.0.failed.load(Ordering::Acquire)
167    }
168    pub fn final_stats(&self) -> Option<SenderStats> {
169        *self.0.final_stats.lock().expect("sender stats poisoned")
170    }
171}
172
173#[derive(Debug)]
174struct Worker {
175    shared: Arc<Shared>,
176    thread: Thread,
177    handle: Option<JoinHandle<()>>,
178}
179
180impl Drop for Worker {
181    fn drop(&mut self) {
182        self.shared.stop.store(true, Ordering::Release);
183        self.thread.unpark();
184        if self
185            .handle
186            .take()
187            .is_some_and(|handle| handle.join().is_err())
188        {
189            self.shared.failed.store(true, Ordering::Release);
190        }
191    }
192}
193
194#[derive(Debug)]
195struct Inbox {
196    clock: RobotClock,
197    worker: Arc<Worker>,
198    pending: SyncSender<Record>,
199    // Mutex supplies Sync for WriteStream; exclusive log() uses get_mut, never lock().
200    free: Mutex<Receiver<Box<[u8]>>>,
201    recycled: SyncSender<Box<[u8]>>,
202}
203
204impl Inbox {
205    fn submit(
206        &mut self,
207        keyframe: bool,
208        encode: impl FnOnce(&mut [u8]) -> CuResult<usize>,
209    ) -> CuResult<()> {
210        if self.worker.shared.failed.load(Ordering::Acquire) {
211            return Err(CuError::from("Logstream sender worker failed"));
212        }
213        let mut bytes = match self.free.get_mut().expect("exclusive inbox").try_recv() {
214            Ok(bytes) => bytes,
215            Err(TryRecvError::Empty) => {
216                self.worker
217                    .shared
218                    .inbox_drops
219                    .fetch_add(1, Ordering::Relaxed);
220                return Ok(());
221            }
222            Err(TryRecvError::Disconnected) => {
223                return Err(CuError::from("Logstream sender worker stopped"));
224            }
225        };
226        let queued_at = self.clock.now();
227        let len = match encode(&mut bytes) {
228            Ok(len) => len,
229            Err(error) => {
230                let _ = self.recycled.try_send(bytes);
231                return Err(error);
232            }
233        };
234        // Capacity equals the total buffer count, so an owned free buffer always
235        // has a corresponding inbox slot. Neither this nor pool exhaustion waits.
236        self.pending
237            .try_send(Record {
238                bytes,
239                len,
240                keyframe,
241                queued_at,
242            })
243            .map_err(|_| CuError::from("Logstream sender inbox disconnected"))?;
244        self.worker.thread.unpark();
245        Ok(())
246    }
247}
248
249/// Encodes directly into a sender-owned buffer on the existing CL output worker.
250pub struct ScheduledCopperListSink<P: CopperListTuple> {
251    inbox: Inbox,
252    structured: Option<ScheduledStructuredLogSink>,
253    encoder: fn(&CopperList<P>, &mut [u8]) -> crate::Result<usize>,
254    _payload: PhantomData<fn() -> P>,
255}
256impl<P: CopperListTuple> ScheduledCopperListSink<P> {
257    /// Take the optional structured-log handoff before installing the CL sink.
258    pub fn take_structured_log_sink(&mut self) -> Option<ScheduledStructuredLogSink> {
259        self.structured.take()
260    }
261
262    /// Select the generated capture encoder before handing the sink to its output worker.
263    pub fn with_encoder(
264        mut self,
265        encoder: fn(&CopperList<P>, &mut [u8]) -> crate::Result<usize>,
266    ) -> Self
267    where
268        P: CopperListTuple,
269    {
270        self.encoder = encoder;
271        self
272    }
273}
274impl<P: CopperListTuple> Debug for ScheduledCopperListSink<P> {
275    fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result {
276        f.debug_struct("ScheduledCopperListSink")
277            .finish_non_exhaustive()
278    }
279}
280impl<P: CopperListTuple + Send + Sync> WriteStream<CopperList<P>> for ScheduledCopperListSink<P> {
281    fn log(&mut self, record: &CopperList<P>) -> CuResult<()> {
282        self.inbox.submit(false, |bytes| {
283            (self.encoder)(record, bytes).map_err(|error| CuError::from(error.to_string()))
284        })
285    }
286}
287
288/// Bounded keyframe encoding; capture objects are released before pacing begins.
289#[derive(Debug)]
290pub struct ScheduledKeyFrameSink {
291    inbox: Inbox,
292    interval: u32,
293}
294impl WriteStream<KeyFrame> for ScheduledKeyFrameSink {
295    fn log(&mut self, keyframe: &KeyFrame) -> CuResult<()> {
296        if !keyframe.culistid.is_multiple_of(u64::from(self.interval)) {
297            return Ok(());
298        }
299        self.inbox.submit(true, |bytes| {
300            if bytes.len() < crate::record::RECORD_HEADER_LEN {
301                return Err(CuError::from("Keyframe buffer too small"));
302            }
303            let (header, payload) = bytes.split_at_mut(crate::record::RECORD_HEADER_LEN);
304            let len = bincode::encode_into_slice(keyframe, payload, bincode::config::standard())
305                .map_err(|error| CuError::from(error.to_string()))?;
306            crate::record::encode_record_header(
307                crate::RecordKind::KeyFrame,
308                keyframe.culistid,
309                &payload[..len],
310                header,
311            )
312            .map_err(|error| CuError::from(error.to_string()))?;
313            Ok(header.len() + len)
314        })
315    }
316}
317
318/// Start one autonomous worker with an explicitly selected local RobotClock.
319/// A real carrier must use a running clock even when application time is mocked.
320/// Drops of both sinks stop repetition, then drain at most max_latency in local
321/// clock time. No application object, serialization, or wait is added to the RT path.
322pub fn scheduled_sinks<P, T>(
323    transport: T,
324    config: LogStreamSenderConfig,
325    clock: RobotClock,
326) -> CuResult<(
327    ScheduledCopperListSink<P>,
328    ScheduledKeyFrameSink,
329    SenderMonitor,
330)>
331where
332    P: CopperListTuple + Send + Sync,
333    T: CuStreamTx + 'static,
334{
335    scheduled_worker(OneWay::new(transport), config, clock, false)
336}
337
338/// Explicitly enable the advisory receive direction; the manifest must advertise it.
339pub fn scheduled_feedback_sinks<P, T>(
340    transport: T,
341    config: LogStreamSenderConfig,
342    clock: RobotClock,
343) -> CuResult<(
344    ScheduledCopperListSink<P>,
345    ScheduledKeyFrameSink,
346    SenderMonitor,
347)>
348where
349    P: CopperListTuple + Send + Sync,
350    T: CuStreamTx + CuFeedbackRx + 'static,
351{
352    scheduled_worker(transport, config, clock, true)
353}
354
355fn scheduled_worker<P, T>(
356    transport: T,
357    config: LogStreamSenderConfig,
358    clock: RobotClock,
359    feedback_enabled: bool,
360) -> CuResult<(
361    ScheduledCopperListSink<P>,
362    ScheduledKeyFrameSink,
363    SenderMonitor,
364)>
365where
366    P: CopperListTuple + Send + Sync,
367    T: CuStreamTx + CuFeedbackRx + 'static,
368{
369    let manifest = crate::SessionManifest::decode_record(&config.recovery.manifest_record);
370    let policy = config.feedback;
371    let advertised = manifest
372        .as_ref()
373        .ok()
374        .and_then(|manifest| manifest.requirements.feedback);
375    if policy.is_some() != feedback_enabled || advertised.is_some() != feedback_enabled {
376        return Err(CuError::from(
377            "Feedback transport wiring must match manifest capability",
378        ));
379    }
380    if let (Some(policy), Some(advertised)) = (policy, advertised)
381        && (manifest
382            .as_ref()
383            .expect("validated feedback manifest")
384            .identity
385            != config.continuous.identity
386            || advertised.report_interval_ms != policy.report_interval_ms)
387    {
388        return Err(CuError::from(
389            "Feedback manifest identity and cadence must match the sender",
390        ));
391    }
392    let mut feedback = policy
393        .map(|policy| {
394            let baseline = u16::try_from(config.continuous.repair_every_source_symbols)
395                .map_err(|_| crate::Error::InvalidConfig("feedback baseline exceeds u16"))?;
396            FeedbackController::new(
397                policy,
398                config.continuous.identity,
399                advertised
400                    .expect("validated feedback capability")
401                    .destination,
402                baseline,
403            )
404        })
405        .transpose()
406        .map_err(|error| CuError::from(error.to_string()))?;
407    let cl_bytes = config.continuous.max_record_bytes;
408    let kf_bytes = usize::try_from(config.recovery.finite.max_object_bytes)
409        .map_err(|_| CuError::from("Keyframe limit exceeds usize"))?;
410    let structured_bytes = kf_bytes.min(STRUCTURED_RECORD_BYTES);
411    let reserved = cl_bytes
412        .checked_mul(CL_BUFFERS)
413        .and_then(|v| {
414            kf_bytes
415                .checked_mul(KF_BUFFERS)
416                .and_then(|k| v.checked_add(k))
417        })
418        .and_then(|v| {
419            v.checked_add(
420                STRUCTURED_BUFFERS * (structured_bytes + 2 * size_of::<StructuredRecord>()),
421            )
422        })
423        .ok_or_else(|| CuError::from("Sender pool size overflow"))?;
424    let shutdown_clock = if clock.is_mock() {
425        RobotClock::new()
426    } else {
427        clock.clone()
428    };
429    let interval = config.recovery.recovery_interval;
430    let drain_time = config.pacing.max_latency;
431    let mut core = SenderCore::new(config, clock.now(), reserved)
432        .map_err(|error| CuError::from(error.to_string()))?;
433    let (pending, records) = mpsc::sync_channel::<Record>(CL_BUFFERS + KF_BUFFERS);
434    let (cl_return, cl_free) = mpsc::sync_channel(CL_BUFFERS);
435    let (kf_return, kf_free) = mpsc::sync_channel(KF_BUFFERS);
436    for _ in 0..CL_BUFFERS {
437        cl_return
438            .try_send(vec![0; cl_bytes].into_boxed_slice())
439            .expect("new pool");
440    }
441    for _ in 0..KF_BUFFERS {
442        kf_return
443            .try_send(vec![0; kf_bytes].into_boxed_slice())
444            .expect("new pool");
445    }
446    let structured = Arc::new(StructuredInbox {
447        free: crossbeam_queue::ArrayQueue::new(STRUCTURED_BUFFERS),
448        pending: crossbeam_queue::ArrayQueue::new(STRUCTURED_BUFFERS),
449    });
450    for _ in 0..STRUCTURED_BUFFERS {
451        let _ = structured.free.push(StructuredRecord {
452            bytes: vec![0; structured_bytes].into_boxed_slice(),
453            len: 0,
454            queued_at: CuTime::default(),
455        });
456    }
457    let structured_worker = structured.clone();
458    let structured_clock = clock.clone();
459    let cl_clock = clock.clone();
460    let kf_clock = clock.clone();
461    let shared = Arc::new(Shared::default());
462    *shared.snapshot.lock().expect("new snapshot") = SenderSnapshot {
463        sampled_at: clock.now(),
464        feedback: feedback.as_ref().map(FeedbackController::snapshot),
465        ..Default::default()
466    };
467    let status = shared.clone();
468    let cl_recycled = cl_return.clone();
469    let kf_recycled = kf_return.clone();
470    let handle = std::thread::Builder::new()
471        .name("cu-logstream".into())
472        .spawn(move || {
473            let mut transport = transport;
474            let mut feedback_packet = [0; FEEDBACK_BUFFER_BYTES];
475            let mut feedback_failed = false;
476            let mut next_snapshot = clock.now();
477            let mut stop_at = None;
478            let mut structured_id = 0u64;
479            // Teardown remains finite even for deliberately frozen scheduler test clocks.
480            let result = (|| -> crate::Result<()> {
481                loop {
482                    if status.stop.load(Ordering::Acquire) && stop_at.is_none() {
483                        core.begin_shutdown();
484                        stop_at = Some(shutdown_clock.now() + drain_time);
485                    }
486                    if let Some(controller) = &mut feedback {
487                        for _ in 0..8 {
488                            if feedback_failed { break; }
489                            match transport.try_recv_feedback(&mut feedback_packet) {
490                                Ok(Some(len)) => match feedback_packet.get(..len).and_then(|p| ReceiverReport::decode(p).ok()) {
491                                    Some(report) => { if controller.receive(report, clock.now()) { core.request_recovery(); } }
492                                    None => controller.invalid_report(),
493                                },
494                                Ok(None) => break,
495                                Err(CuStreamRxError::BufferTooSmall { .. }) => controller.invalid_report(),
496                                Err(CuStreamRxError::Failed(_)) => { feedback_failed = true; }
497                            }
498                        }
499                        controller.tick(clock.now());
500                        core.set_repair_interval(controller.snapshot().effective_repair_every_source_symbols)?;
501                    }
502                    let mut received = 0;
503                    for _ in 0..CL_BUFFERS + KF_BUFFERS {
504                        let Ok(record) = records.try_recv() else {
505                            break;
506                        };
507                        let result =
508                            core.accept_record(&record.bytes[..record.len], record.queued_at);
509                        let returned = if record.keyframe {
510                            &kf_return
511                        } else {
512                            &cl_return
513                        };
514                        let _ = returned.try_send(record.bytes);
515                        result?;
516                        received += 1;
517                    }
518                    for _ in 0..STRUCTURED_BUFFERS {
519                        let Some(mut record) = structured_worker.pending.pop() else { break; };
520                        let (header, payload) = record.bytes[..record.len].split_at_mut(crate::record::RECORD_HEADER_LEN);
521                        let result = crate::record::encode_record_header(
522                            crate::RecordKind::StructuredLog, structured_id, payload, header,
523                        ).and_then(|()| core.accept_record(&record.bytes[..record.len], record.queued_at));
524                        let _ = structured_worker.free.push(record);
525                        result?;
526                        structured_id = structured_id.checked_add(1)
527                            .ok_or(crate::Error::InvalidConfig("structured log sequence exhausted"))?;
528                        received += 1;
529                    }
530                    let next = core.poll(clock.now(), &mut transport)?;
531                    let now = clock.now();
532                    if now >= next_snapshot {
533                        if let Ok(mut snapshot) = status.snapshot.try_lock() {
534                            *snapshot = SenderSnapshot { sampled_at: now, stats: core.stats(),
535                                inbox_drops: status.inbox_drops.load(Ordering::Relaxed),
536                                feedback: feedback.as_ref().map(FeedbackController::snapshot), feedback_failed,
537                                ..Default::default() };
538                        }
539                        next_snapshot = now + CuDuration::from_millis(100);
540                    }
541                    if stop_at.is_some_and(|deadline| shutdown_clock.now() >= deadline)
542                        || (stop_at.is_some() && received == 0 && core.is_idle())
543                    {
544                        break;
545                    }
546                    if received > 0 {
547                        continue;
548                    }
549                    if let Some(deadline) = next {
550                        let now = clock.now();
551                        if deadline > now {
552                            let duration = (deadline - now).min(CuDuration::from_millis(50));
553                            std::thread::park_timeout(std::time::Duration::from_nanos(
554                                duration.as_nanos(),
555                            ));
556                        }
557                    } else {
558                        std::thread::park_timeout(std::time::Duration::from_millis(50));
559                    }
560                }
561                Ok(())
562            })();
563            status.failed.store(result.is_err(), Ordering::Release);
564            core.discard_pending();
565            while let Some(record) = structured_worker.pending.pop() {
566                status.inbox_drops.fetch_add(1, Ordering::Relaxed);
567                let _ = structured_worker.free.push(record);
568            }
569            let stats = core.stats();
570            *status.final_stats.lock().expect("sender stats") = Some(stats);
571            *status.snapshot.lock().expect("sender snapshot") = SenderSnapshot {
572                sampled_at: clock.now(), stats, inbox_drops: status.inbox_drops.load(Ordering::Relaxed),
573                stopped: true, failed: result.is_err(), feedback_failed,
574                feedback: feedback.as_ref().map(FeedbackController::snapshot),
575            };
576            cu29_log_derive::info!("Logstream sender stopped: sent={} bytes={} queue_drops={} expired={} transport_drops={} inbox_drops={} shutdown_drops={}",
577                stats.packets_sent, stats.bytes_sent, stats.queue_drops, stats.expired_packets, stats.transport_drops,
578                status.inbox_drops.load(Ordering::Relaxed), stats.shutdown_drops);
579            if let Err(error) = result {
580                status.failed.store(true, Ordering::Release);
581                cu29_log_derive::error!("Logstream sender failed: {}", error.to_string());
582            }
583        })
584        .map_err(|error| CuError::new_with_cause("Start logstream sender", error))?;
585    let worker = Arc::new(Worker {
586        shared: shared.clone(),
587        thread: handle.thread().clone(),
588        handle: Some(handle),
589    });
590    Ok((
591        ScheduledCopperListSink {
592            structured: Some(ScheduledStructuredLogSink {
593                inbox: structured,
594                status: shared.clone(),
595                thread: worker.thread.clone(),
596                clock: structured_clock,
597                current: None,
598                oversized: false,
599            }),
600            encoder: crate::encode_copperlist_record_into,
601            inbox: Inbox {
602                clock: cl_clock,
603                worker: worker.clone(),
604                pending: pending.clone(),
605                free: Mutex::new(cl_free),
606                recycled: cl_recycled,
607            },
608            _payload: PhantomData,
609        },
610        ScheduledKeyFrameSink {
611            inbox: Inbox {
612                clock: kf_clock,
613                worker,
614                pending,
615                free: Mutex::new(kf_free),
616                recycled: kf_recycled,
617            },
618            interval,
619        },
620        SenderMonitor(shared),
621    ))
622}
623
624impl SenderMonitor {
625    /// Transfer a read-only worker handle into generated monitoring metadata.
626    pub fn into_runtime_monitor(
627        self,
628        destination: &str,
629        bitrate_bps: u64,
630        baseline: usize,
631    ) -> cu29_runtime::monitoring::LogStreamMonitor {
632        cu29_runtime::monitoring::LogStreamMonitor::new(destination, bitrate_bps, baseline, self)
633    }
634}
635impl cu29_runtime::monitoring::LogStreamStatsSource for SenderMonitor {
636    fn snapshot(&self) -> cu29_runtime::monitoring::LogStreamStats {
637        use cu29_runtime::monitoring::{
638            LogStreamFeedbackState, LogStreamFeedbackStats, LogStreamStats,
639        };
640        let snapshot = SenderMonitor::snapshot(self);
641        let stats = snapshot.stats;
642        LogStreamStats {
643            sampled_at: snapshot.sampled_at,
644            packets_sent: stats.packets_sent,
645            bytes_sent: stats.bytes_sent,
646            queue_drops: stats.queue_drops,
647            expired_packets: stats.expired_packets,
648            transport_drops: stats.transport_drops,
649            inbox_drops: snapshot.inbox_drops,
650            shutdown_drops: stats.shutdown_drops,
651            recovery_rounds: stats.recovery_rounds,
652            recovery_superseded: stats.recovery_superseded,
653            queue_peak: stats.queue_peak,
654            stopped: snapshot.stopped,
655            failed: snapshot.failed,
656            feedback: snapshot.feedback.map(|feedback| {
657                let report = feedback.report.unwrap_or_default();
658                LogStreamFeedbackStats {
659                    state: match feedback.state {
660                        crate::feedback::FeedbackState::Waiting => LogStreamFeedbackState::Waiting,
661                        crate::feedback::FeedbackState::Active => LogStreamFeedbackState::Active,
662                        crate::feedback::FeedbackState::Stale => LogStreamFeedbackState::Stale,
663                    },
664                    age: feedback.last_received.map(|last| {
665                        CuDuration::from_nanos(
666                            snapshot
667                                .sampled_at
668                                .as_nanos()
669                                .saturating_sub(last.as_nanos()),
670                        )
671                    }),
672                    failed: snapshot.feedback_failed,
673                    reports: feedback.accepted_reports,
674                    rejected_reports: feedback.rejected_reports,
675                    invalid_reports: feedback.invalid_reports,
676                    rates_available: feedback.receiver_rates_available,
677                    source_metrics_available: feedback.source_metrics_available,
678                    bytes_per_second: feedback.receiver_bytes_per_second,
679                    packets_per_second: feedback.receiver_packets_per_second,
680                    finalized_symbols: report.sources.finalized,
681                    loss_basis_points: feedback.source_loss_basis_points,
682                    recovery_basis_points: feedback.source_recovery_basis_points,
683                    buffered_records: report.buffered_records,
684                    record_capacity: report.record_capacity,
685                    latest_copperlist: report.latest_copperlist,
686                    effective_repair_every_source_symbols: feedback
687                        .effective_repair_every_source_symbols,
688                    invalid_packets: report.invalid_packets,
689                    duplicate_packets: report.duplicate_packets,
690                    expired_records: report.expired_records,
691                }
692            }),
693        }
694    }
695}
cu29_logstream/worker.rs

cu29_logstream/
worker.rs
