1use std::collections::{BTreeMap, HashMap, VecDeque};
33use std::fmt::Write as _;
34use std::sync::atomic::{AtomicU8, Ordering};
35use std::sync::{Arc, Mutex, OnceLock};
36
37use serde::Serialize;
38use tracing::field::{Field, Visit};
39use tracing::{Event, Subscriber};
40use tracing_subscriber::Layer;
41use tracing_subscriber::layer::Context;
42use tracing_subscriber::registry::LookupSpan;
43
44pub const DEFAULT_CAPACITY: usize = 5_000;
46pub const DEFAULT_MAX_BYTES: usize = 8 * 1024 * 1024;
48pub const MAX_MESSAGE_BYTES: usize = 8 * 1024;
50pub const MAX_FIELD_BYTES: usize = 2 * 1024;
52pub const MAX_BACKTRACE_BYTES: usize = 16 * 1024;
54
55#[derive(
59 Debug, Clone, Copy, PartialEq, Eq, PartialOrd, Ord, Hash, Serialize, serde::Deserialize,
60)]
61#[serde(rename_all = "lowercase")]
62pub enum LogLevel {
63 Trace,
65 Debug,
67 Info,
69 Warn,
71 Error,
73}
74
75impl LogLevel {
76 #[must_use]
78 pub fn parse(s: &str) -> Option<Self> {
79 match s.trim().to_ascii_lowercase().as_str() {
80 "trace" => Some(Self::Trace),
81 "debug" => Some(Self::Debug),
82 "info" => Some(Self::Info),
83 "warn" | "warning" => Some(Self::Warn),
84 "error" | "err" | "fatal" | "panic" => Some(Self::Error),
85 _ => None,
86 }
87 }
88
89 #[must_use]
91 pub const fn as_str(self) -> &'static str {
92 match self {
93 Self::Trace => "trace",
94 Self::Debug => "debug",
95 Self::Info => "info",
96 Self::Warn => "warn",
97 Self::Error => "error",
98 }
99 }
100
101 const fn from_tracing(level: tracing::Level) -> Self {
102 match level {
103 tracing::Level::TRACE => Self::Trace,
104 tracing::Level::DEBUG => Self::Debug,
105 tracing::Level::INFO => Self::Info,
106 tracing::Level::WARN => Self::Warn,
107 tracing::Level::ERROR => Self::Error,
108 }
109 }
110
111 const fn to_level_filter(self) -> tracing::level_filters::LevelFilter {
112 match self {
113 Self::Trace => tracing::level_filters::LevelFilter::TRACE,
114 Self::Debug => tracing::level_filters::LevelFilter::DEBUG,
115 Self::Info => tracing::level_filters::LevelFilter::INFO,
116 Self::Warn => tracing::level_filters::LevelFilter::WARN,
117 Self::Error => tracing::level_filters::LevelFilter::ERROR,
118 }
119 }
120
121 const fn from_log(level: log::Level) -> Self {
122 match level {
123 log::Level::Trace => Self::Trace,
124 log::Level::Debug => Self::Debug,
125 log::Level::Info => Self::Info,
126 log::Level::Warn => Self::Warn,
127 log::Level::Error => Self::Error,
128 }
129 }
130
131 const fn index(self) -> usize {
132 match self {
133 Self::Trace => 0,
134 Self::Debug => 1,
135 Self::Info => 2,
136 Self::Warn => 3,
137 Self::Error => 4,
138 }
139 }
140}
141
142#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize)]
144#[serde(rename_all = "snake_case")]
145pub enum LogSource {
146 Tracing,
148 Log,
150 Panic,
152}
153
154#[derive(Debug, Clone, Serialize)]
158#[non_exhaustive]
159pub struct BackendLogEntry {
160 pub seq: u64,
162 pub ts_ms: u64,
164 pub level: LogLevel,
166 pub target: String,
168 pub message: String,
170 #[serde(skip_serializing_if = "serde_json::Map::is_empty")]
172 pub fields: serde_json::Map<String, serde_json::Value>,
173 #[serde(skip_serializing_if = "Vec::is_empty")]
175 pub spans: Vec<String>,
176 #[serde(skip_serializing_if = "Option::is_none")]
178 pub thread: Option<String>,
179 #[serde(skip_serializing_if = "Option::is_none")]
181 pub location: Option<String>,
182 pub source: LogSource,
184}
185
186impl BackendLogEntry {
187 #[cfg(test)]
189 pub(crate) fn for_tests(target: &str, message: &str) -> Self {
190 Self {
191 seq: 0,
192 ts_ms: now_ms(),
193 level: LogLevel::Info,
194 target: target.to_string(),
195 message: message.to_string(),
196 fields: serde_json::Map::new(),
197 spans: Vec::new(),
198 thread: None,
199 location: None,
200 source: LogSource::Tracing,
201 }
202 }
203
204 fn approx_bytes(&self) -> usize {
205 let fields: usize = self
206 .fields
207 .iter()
208 .map(|(k, v)| k.len() + value_len(v))
209 .sum();
210 let spans: usize = self.spans.iter().map(String::len).sum();
211 96 + self.target.len() + self.message.len() + fields + spans
212 }
213
214 #[must_use]
216 pub fn render(&self) -> String {
217 let secs = self.ts_ms / 1000;
218 let ms = self.ts_ms % 1000;
219 let (h, m, s) = ((secs / 3600) % 24, (secs / 60) % 60, secs % 60);
220 let mut out = format!(
221 "{h:02}:{m:02}:{s:02}.{ms:03}Z {:<5} {}: {}",
222 self.level.as_str().to_ascii_uppercase(),
223 self.target,
224 self.message
225 );
226 for (k, v) in &self.fields {
227 match v {
228 serde_json::Value::String(s) => {
229 let _ = write!(out, " {k}={s:?}");
230 }
231 other => {
232 let _ = write!(out, " {k}={other}");
233 }
234 }
235 }
236 out
237 }
238}
239
240fn value_len(v: &serde_json::Value) -> usize {
241 match v {
242 serde_json::Value::String(s) => s.len(),
243 _ => 16,
244 }
245}
246
247#[derive(Default)]
250struct Ring {
251 entries: VecDeque<BackendLogEntry>,
252 next_seq: u64,
253 bytes: usize,
254 evicted: u64,
256 cleared_before: u64,
259 totals: [u64; 5],
261}
262
263impl Ring {
264 fn index_of(&self, seq: u64) -> usize {
267 self.entries.front().map_or(0, |front| {
268 usize::try_from(seq.saturating_sub(front.seq))
269 .unwrap_or(usize::MAX)
270 .min(self.entries.len())
271 })
272 }
273
274 fn evicted_since(&self, from: u64) -> u64 {
276 let oldest = self.entries.front().map_or(self.next_seq, |e| e.seq);
277 oldest.saturating_sub(from.max(self.cleared_before))
278 }
279}
280
281thread_local! {
282 static HOLDS_RING: std::cell::Cell<bool> = const { std::cell::Cell::new(false) };
284}
285
286struct RingGuard<'a>(std::sync::MutexGuard<'a, Ring>);
288
289impl Drop for RingGuard<'_> {
290 fn drop(&mut self) {
291 HOLDS_RING.with(|h| h.set(false));
292 }
293}
294
295impl std::ops::Deref for RingGuard<'_> {
296 type Target = Ring;
297 fn deref(&self) -> &Ring {
298 &self.0
299 }
300}
301
302impl std::ops::DerefMut for RingGuard<'_> {
303 fn deref_mut(&mut self) -> &mut Ring {
304 &mut self.0
305 }
306}
307
308fn debug_bounded(value: &dyn std::fmt::Debug, max: usize) -> String {
312 struct Bounded {
313 out: String,
314 max: usize,
315 cut: bool,
316 }
317 impl std::fmt::Write for Bounded {
318 fn write_str(&mut self, s: &str) -> std::fmt::Result {
319 let room = self.max.saturating_sub(self.out.len());
320 if s.len() <= room {
321 self.out.push_str(s);
322 return Ok(());
323 }
324 let mut end = room;
325 while end > 0 && !s.is_char_boundary(end) {
326 end -= 1;
327 }
328 self.out.push_str(&s[..end]);
329 self.cut = true;
330 Err(std::fmt::Error)
332 }
333 }
334 let mut b = Bounded {
335 out: String::new(),
336 max,
337 cut: false,
338 };
339 let _ = write!(b, "{value:?}");
340 if b.cut {
341 b.out.push_str("…[truncated]");
342 }
343 b.out
344}
345
346thread_local! {
347 static IN_NOTIFY: std::cell::Cell<bool> = const { std::cell::Cell::new(false) };
355}
356
357struct NotifyScope(bool);
359
360impl NotifyScope {
361 fn enter() -> Self {
362 Self(IN_NOTIFY.with(|f| f.replace(true)))
363 }
364}
365
366impl Drop for NotifyScope {
367 fn drop(&mut self) {
368 let previous = self.0;
369 IN_NOTIFY.with(|f| f.set(previous));
370 }
371}
372
373struct PollInNotifyScope<F>(F);
375
376impl<F: std::future::Future + Unpin> std::future::Future for PollInNotifyScope<F> {
377 type Output = F::Output;
378 fn poll(
379 mut self: std::pin::Pin<&mut Self>,
380 cx: &mut std::task::Context<'_>,
381 ) -> std::task::Poll<Self::Output> {
382 let _scope = NotifyScope::enter();
383 std::pin::Pin::new(&mut self.0).poll(cx)
384 }
385}
386
387pub struct LogBuffer {
392 ring: Mutex<Ring>,
393 notify: tokio::sync::Notify,
394 capacity: usize,
395 max_bytes: usize,
396}
397
398impl std::fmt::Debug for LogBuffer {
399 fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
400 f.debug_struct("LogBuffer")
401 .field("capacity", &self.capacity)
402 .field("max_bytes", &self.max_bytes)
403 .finish_non_exhaustive()
404 }
405}
406
407impl Default for LogBuffer {
408 fn default() -> Self {
409 Self::new(DEFAULT_CAPACITY, DEFAULT_MAX_BYTES)
410 }
411}
412
413impl LogBuffer {
414 #[must_use]
417 pub fn new(capacity: usize, max_bytes: usize) -> Self {
418 Self {
419 ring: Mutex::new(Ring::default()),
420 notify: tokio::sync::Notify::new(),
421 capacity: capacity.max(1),
422 max_bytes: max_bytes.max(1),
423 }
424 }
425
426 fn lock(&self) -> RingGuard<'_> {
427 let guard = self
428 .ring
429 .lock()
430 .unwrap_or_else(std::sync::PoisonError::into_inner);
431 HOLDS_RING.with(|h| h.set(true));
432 RingGuard(guard)
433 }
434
435 pub fn push(&self, entry: BackendLogEntry) {
439 if HOLDS_RING.with(std::cell::Cell::get) {
440 return;
441 }
442 let mut ring = self.lock();
443 Self::push_locked(&mut ring, entry, self.capacity, self.max_bytes);
444 drop(ring);
445 self.wake_waiters();
446 }
447
448 fn try_push(&self, entry: BackendLogEntry) -> bool {
454 if HOLDS_RING.with(std::cell::Cell::get) {
455 return false;
456 }
457 let mut ring = self.lock();
458 Self::push_locked(&mut ring, entry, self.capacity, self.max_bytes);
459 drop(ring);
460 self.wake_waiters();
461 true
462 }
463
464 fn push_locked(ring: &mut Ring, mut entry: BackendLogEntry, capacity: usize, max_bytes: usize) {
465 entry.seq = ring.next_seq;
466 ring.next_seq += 1;
467 ring.totals[entry.level.index()] += 1;
468 ring.bytes += entry.approx_bytes();
469 ring.entries.push_back(entry);
470 while ring.entries.len() > capacity || (ring.bytes > max_bytes && ring.entries.len() > 1) {
471 if let Some(old) = ring.entries.pop_front() {
472 ring.bytes = ring.bytes.saturating_sub(old.approx_bytes());
473 ring.evicted += 1;
474 }
475 }
476 }
477
478 fn wake_waiters(&self) {
481 if IN_NOTIFY.with(std::cell::Cell::get) {
482 return;
483 }
484 let _scope = NotifyScope::enter();
485 self.notify.notify_waiters();
486 }
487
488 #[must_use]
490 pub fn next_seq(&self) -> u64 {
491 self.lock().next_seq
492 }
493
494 #[must_use]
497 pub fn seq_at_or_after(&self, ts_ms: u64) -> u64 {
498 let ring = self.lock();
499 ring.entries
500 .iter()
501 .find(|e| e.ts_ms >= ts_ms)
502 .map_or(ring.next_seq, |e| e.seq)
503 }
504
505 #[must_use]
507 pub fn query(&self, q: &LogQuery) -> LogPage {
508 let ring = self.lock();
509 let oldest_seq = ring.entries.front().map_or(ring.next_seq, |e| e.seq);
510 let matcher = q.matcher();
511 let mut matched: Vec<BackendLogEntry> = Vec::new();
512 let mut total_matched = 0usize;
513 if let Some(since) = q.since_seq {
514 for e in ring.entries.range(ring.index_of(since)..) {
518 if matcher.matches(e) {
519 total_matched += 1;
520 if matched.len() < q.limit {
521 matched.push(e.clone());
522 }
523 }
524 }
525 } else {
526 for e in ring.entries.iter().rev() {
528 if !matcher.matches(e) {
529 continue;
530 }
531 total_matched += 1;
532 if matched.len() < q.limit {
533 matched.push(e.clone());
534 }
535 }
536 matched.reverse();
537 }
538 let truncated = total_matched > matched.len();
539 let gap = q.since_seq.is_some_and(|s| ring.evicted_since(s) > 0);
540 LogPage {
541 next_seq: page_next_seq(q.since_seq, truncated, &matched, ring.next_seq),
542 truncated,
543 total_matched,
544 entries: matched,
545 oldest_seq,
546 evicted: ring.evicted,
547 gap,
548 }
549 }
550
551 #[must_use]
554 pub fn range(
555 &self,
556 from: u64,
557 to: u64,
558 min_level: LogLevel,
559 limit: usize,
560 ) -> (Vec<BackendLogEntry>, usize) {
561 let ring = self.lock();
562 let all: Vec<&BackendLogEntry> = ring
563 .entries
564 .range(ring.index_of(from)..)
565 .take_while(|e| e.seq < to)
566 .filter(|e| e.level >= min_level)
567 .collect();
568 let total = all.len();
569 let start = total.saturating_sub(limit);
570 (all[start..].iter().map(|e| (*e).clone()).collect(), total)
571 }
572
573 #[must_use]
577 pub fn digest(&self, recent: usize, top: usize) -> LogDigest {
578 let ring = self.lock();
579 let mut in_buffer = [0u64; 5];
580 let mut targets: HashMap<&str, u64> = HashMap::new();
581 let mut templates: HashMap<(LogLevel, String, String), TemplateStat> = HashMap::new();
582 let mut recent_problems: VecDeque<BackendLogEntry> = VecDeque::new();
583 let mut panics: Vec<BackendLogEntry> = Vec::new();
584 for e in &ring.entries {
585 in_buffer[e.level.index()] += 1;
586 *targets.entry(e.target.as_str()).or_default() += 1;
587 let key = (e.level, e.target.clone(), message_template(&e.message));
588 let stat = templates.entry(key).or_insert_with(|| TemplateStat {
589 count: 0,
590 first_ts_ms: e.ts_ms,
591 last_ts_ms: e.ts_ms,
592 last_seq: e.seq,
593 example: e.message.clone(),
594 });
595 stat.count += 1;
596 stat.last_ts_ms = e.ts_ms;
597 stat.last_seq = e.seq;
598 if e.level >= LogLevel::Warn {
599 recent_problems.push_back(e.clone());
600 if recent_problems.len() > recent {
601 recent_problems.pop_front();
602 }
603 }
604 if e.source == LogSource::Panic {
605 panics.push(e.clone());
606 }
607 }
608 let mut top_targets: Vec<(String, u64)> = targets
609 .into_iter()
610 .map(|(k, v)| (k.to_string(), v))
611 .collect();
612 top_targets.sort_by(|a, b| b.1.cmp(&a.1).then_with(|| a.0.cmp(&b.0)));
613 top_targets.truncate(top);
614 let mut repeated: Vec<RepeatedMessage> = templates
615 .into_iter()
616 .filter(|(_, s)| s.count > 1)
617 .map(|((level, target, template), s)| RepeatedMessage {
618 level,
619 target,
620 template,
621 count: s.count,
622 first_ts_ms: s.first_ts_ms,
623 last_ts_ms: s.last_ts_ms,
624 last_seq: s.last_seq,
625 example: s.example,
626 })
627 .collect();
628 repeated.sort_by(|a, b| {
629 b.level
630 .cmp(&a.level)
631 .then(b.count.cmp(&a.count))
632 .then_with(|| a.template.cmp(&b.template))
633 });
634 repeated.truncate(top);
635 LogDigest {
636 buffered: ring.entries.len(),
637 next_seq: ring.next_seq,
638 evicted: ring.evicted,
639 oldest_ts_ms: ring.entries.front().map(|e| e.ts_ms),
640 newest_ts_ms: ring.entries.back().map(|e| e.ts_ms),
641 lifetime_counts: level_map(&ring.totals),
642 buffered_counts: level_map(&in_buffer),
643 top_targets: top_targets
644 .into_iter()
645 .map(|(target, count)| TargetCount { target, count })
646 .collect(),
647 repeated,
648 recent_problems: recent_problems.into_iter().collect(),
649 panics,
650 }
651 }
652
653 pub async fn wait_for<F>(
656 &self,
657 from_seq: u64,
658 timeout: std::time::Duration,
659 pred: F,
660 ) -> (Option<BackendLogEntry>, u64)
661 where
662 F: Fn(&BackendLogEntry) -> bool,
663 {
664 let deadline = tokio::time::Instant::now() + timeout;
665 let mut scanned_to = from_seq;
666 let mut missed = 0;
669 loop {
670 let notified = self.notify.notified();
673 tokio::pin!(notified);
674 {
675 let _scope = NotifyScope::enter();
676 notified.as_mut().enable();
677 }
678 {
679 let ring = self.lock();
680 missed += ring.evicted_since(scanned_to);
681 for e in ring.entries.range(ring.index_of(scanned_to)..) {
684 if pred(e) {
685 return (Some(e.clone()), missed);
686 }
687 }
688 scanned_to = scanned_to.max(ring.next_seq);
690 }
691 if tokio::time::timeout_at(deadline, PollInNotifyScope(notified.as_mut()))
692 .await
693 .is_err()
694 {
695 return (None, missed);
696 }
697 }
698 }
699
700 pub fn clear(&self) {
702 let mut ring = self.lock();
703 ring.entries.clear();
704 ring.bytes = 0;
705 ring.cleared_before = ring.next_seq;
706 }
707}
708
709struct TemplateStat {
710 count: u64,
711 first_ts_ms: u64,
712 last_ts_ms: u64,
713 last_seq: u64,
714 example: String,
715}
716
717fn level_map(counts: &[u64; 5]) -> BTreeMap<&'static str, u64> {
718 [
719 LogLevel::Trace,
720 LogLevel::Debug,
721 LogLevel::Info,
722 LogLevel::Warn,
723 LogLevel::Error,
724 ]
725 .into_iter()
726 .map(|l| (l.as_str(), counts[l.index()]))
727 .collect()
728}
729
730#[must_use]
733pub fn message_template(message: &str) -> String {
734 let mut out = String::with_capacity(message.len().min(256));
735 let mut chars = message.chars().peekable();
736 let mut last_was_placeholder = false;
737 while let Some(c) = chars.next() {
738 if c.is_ascii_digit() {
739 while let Some(&n) = chars.peek() {
741 if n.is_ascii_alphanumeric() || n == '.' || n == '-' || n == '_' || n == ':' {
742 chars.next();
743 } else {
744 break;
745 }
746 }
747 if !last_was_placeholder {
748 out.push('#');
749 }
750 last_was_placeholder = true;
751 } else if c == '"' {
752 for n in chars.by_ref() {
753 if n == '"' {
754 break;
755 }
756 }
757 out.push_str("\"…\"");
758 last_was_placeholder = true;
759 } else {
760 out.push(c);
761 last_was_placeholder = false;
762 }
763 if out.len() >= 240 {
764 out.push('…');
765 break;
766 }
767 }
768 out
769}
770
771#[derive(Debug, Clone)]
775#[non_exhaustive]
776pub struct LogQuery {
777 pub min_level: LogLevel,
779 pub targets: Vec<String>,
781 pub contains: Option<String>,
783 pub since_seq: Option<u64>,
785 pub since_ms: Option<u64>,
787 pub fields: Vec<(String, serde_json::Value)>,
789 pub limit: usize,
791}
792
793impl Default for LogQuery {
794 fn default() -> Self {
795 Self {
796 min_level: LogLevel::Trace,
797 targets: Vec::new(),
798 contains: None,
799 since_seq: None,
800 since_ms: None,
801 fields: Vec::new(),
802 limit: 100,
803 }
804 }
805}
806
807impl LogQuery {
808 fn matcher(&self) -> Matcher<'_> {
809 Matcher {
810 q: self,
811 needle: self.contains.as_ref().map(|s| s.to_lowercase()),
812 }
813 }
814}
815
816struct Matcher<'a> {
817 q: &'a LogQuery,
818 needle: Option<String>,
819}
820
821impl Matcher<'_> {
822 fn matches(&self, e: &BackendLogEntry) -> bool {
823 if e.level < self.q.min_level {
824 return false;
825 }
826 if let Some(ms) = self.q.since_ms
827 && e.ts_ms < ms
828 {
829 return false;
830 }
831 if !self.q.targets.is_empty()
832 && !self
833 .q
834 .targets
835 .iter()
836 .any(|t| e.target.starts_with(t.as_str()))
837 {
838 return false;
839 }
840 if !entry_fields_match(e, &self.q.fields) {
841 return false;
842 }
843 if let Some(needle) = &self.needle {
844 return entry_contains(e, needle);
845 }
846 true
847 }
848}
849
850#[must_use]
857pub fn field_matches(entry: &BackendLogEntry, key: &str, expected: &serde_json::Value) -> bool {
858 fn text(v: &serde_json::Value) -> String {
859 match v {
860 serde_json::Value::String(s) => s.trim_matches('"').to_lowercase(),
861 other => other.to_string().to_lowercase(),
862 }
863 }
864 entry
865 .fields
866 .get(key)
867 .is_some_and(|actual| text(actual) == text(expected))
868}
869
870#[must_use]
872pub fn entry_fields_match(entry: &BackendLogEntry, fields: &[(String, serde_json::Value)]) -> bool {
873 fields.iter().all(|(k, v)| field_matches(entry, k, v))
874}
875
876#[must_use]
879pub fn entry_contains(e: &BackendLogEntry, needle: &str) -> bool {
880 if e.message.to_lowercase().contains(needle) || e.target.to_lowercase().contains(needle) {
881 return true;
882 }
883 if e.spans.iter().any(|s| s.to_lowercase().contains(needle)) {
884 return true;
885 }
886 e.fields.iter().any(|(k, v)| {
887 k.to_lowercase().contains(needle)
888 || match v {
889 serde_json::Value::String(s) => s.to_lowercase().contains(needle),
890 other => other.to_string().to_lowercase().contains(needle),
891 }
892 })
893}
894
895#[must_use]
898pub fn page_next_seq(
899 since_seq: Option<u64>,
900 truncated: bool,
901 entries: &[BackendLogEntry],
902 buffer_next_seq: u64,
903) -> u64 {
904 match entries.last() {
905 Some(last) if since_seq.is_some() && truncated => last.seq + 1,
906 _ => buffer_next_seq,
907 }
908}
909
910#[derive(Debug, Clone, Serialize)]
912#[non_exhaustive]
913pub struct LogPage {
914 pub entries: Vec<BackendLogEntry>,
916 pub total_matched: usize,
918 pub truncated: bool,
921 pub next_seq: u64,
923 pub oldest_seq: u64,
925 pub evicted: u64,
927 pub gap: bool,
930}
931
932#[derive(Debug, Clone, Serialize)]
934#[non_exhaustive]
935pub struct LogDigest {
936 pub buffered: usize,
938 pub next_seq: u64,
940 pub evicted: u64,
942 pub oldest_ts_ms: Option<u64>,
944 pub newest_ts_ms: Option<u64>,
946 pub lifetime_counts: BTreeMap<&'static str, u64>,
948 pub buffered_counts: BTreeMap<&'static str, u64>,
950 pub top_targets: Vec<TargetCount>,
952 pub repeated: Vec<RepeatedMessage>,
954 pub recent_problems: Vec<BackendLogEntry>,
956 pub panics: Vec<BackendLogEntry>,
958}
959
960#[derive(Debug, Clone, Serialize)]
962pub struct TargetCount {
963 pub target: String,
965 pub count: u64,
967}
968
969#[derive(Debug, Clone, Serialize)]
971pub struct RepeatedMessage {
972 pub level: LogLevel,
974 pub target: String,
976 pub template: String,
978 pub count: u64,
980 pub first_ts_ms: u64,
982 pub last_ts_ms: u64,
984 pub last_seq: u64,
986 pub example: String,
988}
989
990static GLOBAL: OnceLock<Arc<LogBuffer>> = OnceLock::new();
993static SOURCES: AtomicU8 = AtomicU8::new(0);
994
995const SRC_TRACING: u8 = 1;
996const SRC_LOG: u8 = 1 << 1;
997const SRC_PANIC: u8 = 1 << 2;
998const SRC_STRUCTURED_INSTALLED: u8 = 1 << 3;
1002
1003#[must_use]
1006pub fn global() -> Arc<LogBuffer> {
1007 GLOBAL
1008 .get_or_init(|| {
1009 let capacity = std::env::var("VICTAURI_LOG_CAPACITY")
1010 .ok()
1011 .and_then(|v| v.trim().parse::<usize>().ok())
1012 .filter(|n| (1..=100_000).contains(n))
1013 .unwrap_or(DEFAULT_CAPACITY);
1014 Arc::new(LogBuffer::new(
1015 capacity,
1016 DEFAULT_MAX_BYTES.max(capacity * 512),
1017 ))
1018 })
1019 .clone()
1020}
1021
1022#[derive(Debug, Clone, Copy, Serialize)]
1024#[allow(clippy::struct_excessive_bools)]
1025#[non_exhaustive]
1026pub struct ActiveSources {
1027 pub tracing: bool,
1029 pub log: bool,
1031 pub panic_hook: bool,
1033 pub stdout_capture: bool,
1035 pub structured_installed: bool,
1039}
1040
1041impl ActiveSources {
1042 #[must_use]
1044 pub const fn any_structured(self) -> bool {
1045 self.tracing || self.log
1046 }
1047}
1048
1049#[must_use]
1051pub fn active_sources() -> ActiveSources {
1052 let bits = SOURCES.load(Ordering::Relaxed);
1053 ActiveSources {
1054 tracing: bits & SRC_TRACING != 0,
1055 log: bits & SRC_LOG != 0,
1056 panic_hook: bits & SRC_PANIC != 0,
1057 stdout_capture: console_capture_path().is_some(),
1058 structured_installed: bits & SRC_STRUCTURED_INSTALLED != 0,
1059 }
1060}
1061
1062fn mark_source(bit: u8) {
1063 SOURCES.fetch_or(bit, Ordering::Relaxed);
1064}
1065
1066fn capture_disabled() -> bool {
1068 if cfg!(not(debug_assertions)) {
1069 return true;
1070 }
1071 std::env::var("VICTAURI_DISABLE").is_ok_and(|v| {
1072 matches!(
1073 v.trim().to_ascii_lowercase().as_str(),
1074 "1" | "true" | "yes" | "on"
1075 )
1076 })
1077}
1078
1079#[must_use]
1084pub fn is_internal_target(target: &str) -> bool {
1085 static KEEP_INTERNAL: OnceLock<bool> = OnceLock::new();
1086 let keep = *KEEP_INTERNAL.get_or_init(|| {
1087 std::env::var("VICTAURI_LOG_INTERNAL").is_ok_and(|v| {
1088 matches!(
1089 v.trim().to_ascii_lowercase().as_str(),
1090 "1" | "true" | "yes" | "on"
1091 )
1092 })
1093 });
1094 if keep {
1095 return false;
1096 }
1097 target.starts_with("victauri_")
1098 || target == "victauri"
1099 || target == "rmcp"
1100 || target.starts_with("rmcp::")
1101}
1102
1103#[must_use]
1108pub fn is_runtime_internal_target(target: &str) -> bool {
1109 target == "tokio"
1110 || target.starts_with("tokio::")
1111 || target.starts_with("runtime::")
1112 || target.starts_with("runtime.")
1113}
1114
1115#[must_use]
1119pub fn capture_level() -> LogLevel {
1120 static LEVEL: OnceLock<LogLevel> = OnceLock::new();
1121 *LEVEL.get_or_init(|| {
1122 std::env::var("VICTAURI_LOG_LEVEL")
1123 .ok()
1124 .and_then(|v| LogLevel::parse(&v))
1125 .unwrap_or(LogLevel::Debug)
1126 })
1127}
1128
1129fn now_ms() -> u64 {
1130 std::time::SystemTime::now()
1131 .duration_since(std::time::UNIX_EPOCH)
1132 .map_or(0, |d| u64::try_from(d.as_millis()).unwrap_or(u64::MAX))
1133}
1134
1135fn truncate_string(mut s: String, max: usize) -> String {
1136 if s.len() > max {
1137 let mut cut = max;
1138 while cut > 0 && !s.is_char_boundary(cut) {
1139 cut -= 1;
1140 }
1141 let extra = s.len() - cut;
1142 s.truncate(cut);
1143 let _ = write!(s, "…[+{extra} bytes]");
1144 }
1145 s
1146}
1147
1148fn current_thread_name() -> Option<String> {
1149 std::thread::current().name().map(str::to_string)
1150}
1151
1152#[derive(Clone)]
1157#[cfg(debug_assertions)]
1158pub struct BackendLogLayer {
1159 buffer: Arc<LogBuffer>,
1160}
1161
1162#[cfg(debug_assertions)]
1163impl std::fmt::Debug for BackendLogLayer {
1164 fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
1165 f.debug_struct("BackendLogLayer").finish_non_exhaustive()
1166 }
1167}
1168
1169#[cfg(debug_assertions)]
1170impl BackendLogLayer {
1171 #[must_use]
1173 pub const fn with_buffer(buffer: Arc<LogBuffer>) -> Self {
1174 Self { buffer }
1175 }
1176}
1177
1178#[cfg(not(debug_assertions))]
1183#[derive(Debug, Clone, Copy)]
1184pub struct BackendLogLayer {
1185 _private: (),
1186}
1187
1188#[cfg(not(debug_assertions))]
1189impl BackendLogLayer {
1190 #[must_use]
1192 pub fn with_buffer(_buffer: Arc<LogBuffer>) -> Self {
1193 Self { _private: () }
1194 }
1195}
1196
1197#[cfg(not(debug_assertions))]
1198impl<S: Subscriber> Layer<S> for BackendLogLayer {}
1199
1200#[must_use]
1225pub fn log_layer() -> Option<BackendLogLayer> {
1226 if capture_disabled() {
1227 return None;
1228 }
1229 mark_source(SRC_STRUCTURED_INSTALLED);
1230 Some(BackendLogLayer::with_buffer(global()))
1231}
1232
1233#[cfg(debug_assertions)]
1235struct SpanFields(String);
1236
1237#[cfg(debug_assertions)]
1238impl<S> Layer<S> for BackendLogLayer
1239where
1240 S: Subscriber + for<'a> LookupSpan<'a>,
1241{
1242 fn max_level_hint(&self) -> Option<tracing::level_filters::LevelFilter> {
1247 Some(capture_level().to_level_filter())
1248 }
1249
1250 fn on_new_span(
1251 &self,
1252 attrs: &tracing::span::Attributes<'_>,
1253 id: &tracing::span::Id,
1254 ctx: Context<'_, S>,
1255 ) {
1256 let mut v = SpanVisitor::default();
1257 attrs.record(&mut v);
1258 if let Some(span) = ctx.span(id) {
1259 span.extensions_mut().insert(SpanFields(v.0));
1260 }
1261 }
1262
1263 fn on_record(
1264 &self,
1265 id: &tracing::span::Id,
1266 values: &tracing::span::Record<'_>,
1267 ctx: Context<'_, S>,
1268 ) {
1269 if let Some(span) = ctx.span(id) {
1270 let mut ext = span.extensions_mut();
1271 let mut v = SpanVisitor(
1272 ext.get_mut::<SpanFields>()
1273 .map(|f| std::mem::take(&mut f.0))
1274 .unwrap_or_default(),
1275 );
1276 values.record(&mut v);
1277 if let Some(f) = ext.get_mut::<SpanFields>() {
1278 f.0 = v.0;
1279 } else {
1280 ext.insert(SpanFields(v.0));
1281 }
1282 }
1283 }
1284
1285 fn on_event(&self, event: &Event<'_>, ctx: Context<'_, S>) {
1286 mark_source(SRC_TRACING);
1287 let meta = event.metadata();
1288 if is_internal_target(meta.target())
1289 || is_runtime_internal_target(meta.target())
1290 || LogLevel::from_tracing(*meta.level()) < capture_level()
1291 {
1292 return;
1293 }
1294 let mut v = EventVisitor::default();
1295 event.record(&mut v);
1296
1297 let mut target = meta.target().to_string();
1300 let mut location = meta
1301 .file()
1302 .map(|f| format!("{f}:{}", meta.line().unwrap_or(0)));
1303 if let Some(serde_json::Value::String(t)) = v.fields.remove("log.target") {
1304 target = t;
1305 }
1306 let log_file = v.fields.remove("log.file");
1307 let log_line = v.fields.remove("log.line");
1308 v.fields.remove("log.module_path");
1309 if let Some(serde_json::Value::String(f)) = log_file {
1310 location = Some(format!(
1311 "{f}:{}",
1312 log_line.as_ref().map_or(0, |l| l.as_u64().unwrap_or(0))
1313 ));
1314 }
1315
1316 let spans = ctx
1317 .event_scope(event)
1318 .map(|scope| {
1319 scope
1320 .from_root()
1321 .map(|span| {
1322 let ext = span.extensions();
1323 match ext.get::<SpanFields>() {
1324 Some(f) if !f.0.is_empty() => format!("{}{{{}}}", span.name(), f.0),
1325 _ => span.name().to_string(),
1326 }
1327 })
1328 .collect()
1329 })
1330 .unwrap_or_default();
1331
1332 self.buffer.push(BackendLogEntry {
1333 seq: 0,
1334 ts_ms: now_ms(),
1335 level: LogLevel::from_tracing(*meta.level()),
1336 target,
1337 message: truncate_string(v.message, MAX_MESSAGE_BYTES),
1338 fields: v.fields,
1339 spans,
1340 thread: current_thread_name(),
1341 location,
1342 source: LogSource::Tracing,
1343 });
1344 }
1345}
1346
1347#[derive(Default)]
1348struct EventVisitor {
1349 message: String,
1350 fields: serde_json::Map<String, serde_json::Value>,
1351}
1352
1353impl EventVisitor {
1354 fn put(&mut self, field: &Field, value: serde_json::Value) {
1355 if self.fields.len() < 64 {
1356 self.fields.insert(field.name().to_string(), value);
1357 }
1358 }
1359}
1360
1361impl Visit for EventVisitor {
1362 fn record_debug(&mut self, field: &Field, value: &dyn std::fmt::Debug) {
1363 if field.name() == "message" {
1364 self.message = debug_bounded(value, MAX_MESSAGE_BYTES);
1365 } else {
1366 self.put(
1367 field,
1368 serde_json::Value::String(debug_bounded(value, MAX_FIELD_BYTES)),
1369 );
1370 }
1371 }
1372
1373 fn record_str(&mut self, field: &Field, value: &str) {
1374 if field.name() == "message" {
1375 value.clone_into(&mut self.message);
1376 } else {
1377 self.put(
1378 field,
1379 serde_json::Value::String(truncate_string(value.to_string(), MAX_FIELD_BYTES)),
1380 );
1381 }
1382 }
1383
1384 fn record_i64(&mut self, field: &Field, value: i64) {
1385 self.put(field, value.into());
1386 }
1387
1388 fn record_u64(&mut self, field: &Field, value: u64) {
1389 self.put(field, value.into());
1390 }
1391
1392 fn record_f64(&mut self, field: &Field, value: f64) {
1393 self.put(field, value.into());
1394 }
1395
1396 fn record_bool(&mut self, field: &Field, value: bool) {
1397 self.put(field, value.into());
1398 }
1399
1400 fn record_error(&mut self, field: &Field, value: &(dyn std::error::Error + 'static)) {
1401 let mut text = value.to_string();
1402 let mut source = value.source();
1403 let mut depth = 0;
1404 while let Some(s) = source {
1405 if depth >= 8 {
1406 break;
1407 }
1408 let _ = write!(text, ": {s}");
1409 source = s.source();
1410 depth += 1;
1411 }
1412 self.put(
1413 field,
1414 serde_json::Value::String(truncate_string(text, MAX_FIELD_BYTES)),
1415 );
1416 }
1417}
1418
1419#[derive(Default)]
1420struct SpanVisitor(String);
1421
1422impl Visit for SpanVisitor {
1423 fn record_debug(&mut self, field: &Field, value: &dyn std::fmt::Debug) {
1424 if self.0.len() > 512 {
1425 return;
1426 }
1427 if !self.0.is_empty() {
1428 self.0.push(' ');
1429 }
1430 let value = debug_bounded(value, MAX_FIELD_BYTES);
1434 if field.name() == "message" {
1435 self.0.push_str(&value);
1436 } else {
1437 let _ = write!(self.0, "{}={value}", field.name());
1438 }
1439 }
1440}
1441
1442pub struct VictauriLogger {
1447 inner: Option<Box<dyn log::Log>>,
1448 buffer: Arc<LogBuffer>,
1449}
1450
1451impl std::fmt::Debug for VictauriLogger {
1452 fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
1453 f.debug_struct("VictauriLogger")
1454 .field("forwards", &self.inner.is_some())
1455 .finish_non_exhaustive()
1456 }
1457}
1458
1459impl VictauriLogger {
1460 #[must_use]
1462 pub fn with_buffer(buffer: Arc<LogBuffer>, inner: Option<Box<dyn log::Log>>) -> Self {
1463 Self { inner, buffer }
1464 }
1465}
1466
1467impl log::Log for VictauriLogger {
1468 fn enabled(&self, metadata: &log::Metadata<'_>) -> bool {
1469 match &self.inner {
1473 Some(inner) => inner.enabled(metadata),
1474 None => metadata.level() <= log::max_level(),
1475 }
1476 }
1477
1478 fn log(&self, record: &log::Record<'_>) {
1479 mark_source(SRC_LOG);
1480 if is_internal_target(record.target())
1481 || is_runtime_internal_target(record.target())
1482 || LogLevel::from_log(record.level()) < capture_level()
1483 {
1484 if let Some(inner) = &self.inner
1485 && inner.enabled(record.metadata())
1486 {
1487 inner.log(record);
1488 }
1489 return;
1490 }
1491 let mut fields = serde_json::Map::new();
1492 if let Some(mp) = record.module_path()
1493 && mp != record.target()
1494 {
1495 fields.insert("module_path".into(), mp.into());
1496 }
1497 self.buffer.push(BackendLogEntry {
1498 seq: 0,
1499 ts_ms: now_ms(),
1500 level: LogLevel::from_log(record.level()),
1501 target: record.target().to_string(),
1502 message: truncate_string(record.args().to_string(), MAX_MESSAGE_BYTES),
1503 fields,
1504 spans: Vec::new(),
1505 thread: current_thread_name(),
1506 location: record
1507 .file()
1508 .map(|f| format!("{f}:{}", record.line().unwrap_or(0))),
1509 source: LogSource::Log,
1510 });
1511 if let Some(inner) = &self.inner
1512 && inner.enabled(record.metadata())
1513 {
1514 inner.log(record);
1515 }
1516 }
1517
1518 fn flush(&self) {
1519 if let Some(inner) = &self.inner {
1520 inner.flush();
1521 }
1522 }
1523}
1524
1525#[must_use]
1530pub fn log_logger() -> Box<dyn log::Log> {
1531 if capture_disabled() {
1532 return Box::new(NullLogger);
1533 }
1534 mark_source(SRC_STRUCTURED_INSTALLED);
1535 Box::new(VictauriLogger::with_buffer(global(), None))
1536}
1537
1538#[must_use]
1551pub fn wrap_logger(inner: Box<dyn log::Log>) -> Box<dyn log::Log> {
1552 if capture_disabled() {
1553 return inner;
1554 }
1555 mark_source(SRC_STRUCTURED_INSTALLED);
1556 Box::new(VictauriLogger::with_buffer(global(), Some(inner)))
1557}
1558
1559struct NullLogger;
1560
1561impl log::Log for NullLogger {
1562 fn enabled(&self, _: &log::Metadata<'_>) -> bool {
1563 false
1564 }
1565 fn log(&self, _: &log::Record<'_>) {}
1566 fn flush(&self) {}
1567}
1568
1569pub fn install_panic_hook() {
1575 static INSTALLED: OnceLock<()> = OnceLock::new();
1576 if capture_disabled() {
1577 return;
1578 }
1579 INSTALLED.get_or_init(|| {
1580 let buffer = global();
1581 let previous = std::panic::take_hook();
1582 std::panic::set_hook(Box::new(move |info| {
1583 previous(info);
1586 capture_panic(&buffer, info);
1587 }));
1588 mark_source(SRC_PANIC);
1589 });
1590}
1591
1592struct PendingBacktrace {
1597 backtrace: std::backtrace::Backtrace,
1598 resolved: OnceLock<String>,
1599}
1600
1601fn pending_backtraces() -> &'static Mutex<HashMap<u32, Arc<PendingBacktrace>>> {
1602 static PENDING: OnceLock<Mutex<HashMap<u32, Arc<PendingBacktrace>>>> = OnceLock::new();
1603 PENDING.get_or_init(|| Mutex::new(HashMap::new()))
1604}
1605
1606pub fn resolve_backtraces(value: &mut serde_json::Value) {
1610 match value {
1611 serde_json::Value::Object(map) => {
1612 if let Some(id) = map
1613 .get("backtrace_id")
1614 .and_then(serde_json::Value::as_u64)
1615 .and_then(|id| u32::try_from(id).ok())
1616 {
1617 let slot = pending_backtraces()
1618 .lock()
1619 .unwrap_or_else(std::sync::PoisonError::into_inner)
1620 .get(&id)
1621 .cloned();
1622 if let Some(slot) = slot {
1623 let text = slot.resolved.get_or_init(|| {
1624 truncate_string(
1625 trim_backtrace(&slot.backtrace.to_string()),
1626 MAX_BACKTRACE_BYTES,
1627 )
1628 });
1629 map.insert("backtrace".into(), text.clone().into());
1630 }
1631 map.remove("backtrace_id");
1632 }
1633 for v in map.values_mut() {
1634 resolve_backtraces(v);
1635 }
1636 }
1637 serde_json::Value::Array(items) => items.iter_mut().for_each(resolve_backtraces),
1638 _ => {}
1639 }
1640}
1641
1642fn capture_panic(buffer: &LogBuffer, info: &std::panic::PanicHookInfo<'_>) {
1643 static BACKTRACES_TAKEN: std::sync::atomic::AtomicU32 = std::sync::atomic::AtomicU32::new(0);
1646 let payload = info.payload();
1647 let message = payload
1648 .downcast_ref::<&str>()
1649 .map(|s| (*s).to_string())
1650 .or_else(|| payload.downcast_ref::<String>().cloned())
1651 .unwrap_or_else(|| "<non-string panic payload>".to_string());
1652 let location = info
1653 .location()
1654 .map(|l| format!("{}:{}:{}", l.file(), l.line(), l.column()));
1655 let mut fields = serde_json::Map::new();
1656 let n = BACKTRACES_TAKEN.fetch_add(1, Ordering::Relaxed);
1657 if n < MAX_PANIC_BACKTRACES {
1658 let slot = Arc::new(PendingBacktrace {
1659 backtrace: std::backtrace::Backtrace::force_capture(),
1660 resolved: OnceLock::new(),
1661 });
1662 if let Ok(mut pending) = pending_backtraces().try_lock() {
1665 pending.insert(n, slot);
1666 fields.insert("backtrace_id".into(), n.into());
1667 fields.insert("backtrace".into(), "(symbolised when read)".into());
1668 }
1669 } else {
1670 fields.insert(
1671 "backtrace".into(),
1672 format!("omitted — only the first {MAX_PANIC_BACKTRACES} panics carry a backtrace")
1673 .into(),
1674 );
1675 }
1676 let thread = std::thread::current()
1677 .name()
1678 .map_or_else(|| "<unnamed>".to_string(), str::to_string);
1679 let _ = buffer.try_push(BackendLogEntry {
1680 seq: 0,
1681 ts_ms: now_ms(),
1682 level: LogLevel::Error,
1683 target: "panic".to_string(),
1684 message: truncate_string(message, MAX_MESSAGE_BYTES),
1685 fields,
1686 spans: Vec::new(),
1687 thread: Some(thread),
1688 location,
1689 source: LogSource::Panic,
1690 });
1691}
1692
1693pub const MAX_PANIC_BACKTRACES: u32 = 20;
1695pub const MAX_BACKTRACE_FRAMES: usize = 40;
1697
1698#[must_use]
1702pub fn trim_backtrace(raw: &str) -> String {
1703 const MACHINERY: [&str; 9] = [
1704 "std::backtrace",
1705 "std::sys::backtrace",
1706 "backtrace_rs",
1707 "victauri_plugin::backend_logs",
1708 "std::panicking",
1709 "core::panicking",
1710 "rust_begin_unwind",
1711 "alloc::boxed::",
1712 "__rustc",
1713 ];
1714 let mut frames: Vec<Vec<&str>> = Vec::new();
1716 for line in raw.lines() {
1717 let t = line.trim_start();
1718 let is_header = t
1719 .split_once(':')
1720 .is_some_and(|(n, _)| !n.is_empty() && n.chars().all(|c| c.is_ascii_digit()));
1721 if is_header || frames.is_empty() {
1722 frames.push(vec![line]);
1723 } else if let Some(last) = frames.last_mut() {
1724 last.push(line);
1725 }
1726 }
1727 let first_real = frames
1728 .iter()
1729 .position(|f| !MACHINERY.iter().any(|m| f[0].contains(m)))
1730 .unwrap_or(0);
1731 let kept: Vec<String> = frames
1732 .iter()
1733 .skip(first_real)
1734 .take(MAX_BACKTRACE_FRAMES)
1735 .map(|f| f.join("\n"))
1736 .collect();
1737 let dropped = frames.len().saturating_sub(first_real + kept.len());
1738 let mut out = kept.join("\n");
1739 if dropped > 0 {
1740 let _ = write!(out, "\n … {dropped} more frames");
1741 }
1742 out
1743}
1744
1745pub const CONSOLE_CAPTURE_ENV: &str = "VICTAURI_CONSOLE_LOG";
1750
1751#[must_use]
1753pub fn console_capture_path() -> Option<std::path::PathBuf> {
1754 std::env::var_os(CONSOLE_CAPTURE_ENV)
1755 .filter(|v| !v.is_empty())
1756 .map(std::path::PathBuf::from)
1757}
1758
1759#[derive(Debug, Clone, Serialize, serde::Deserialize)]
1761#[non_exhaustive]
1762pub struct ConsoleLine {
1763 #[serde(default)]
1765 pub offset: u64,
1766 #[serde(rename = "t")]
1768 pub ts_ms: u64,
1769 #[serde(rename = "s")]
1771 pub stream: String,
1772 #[serde(rename = "l")]
1774 pub text: String,
1775 #[serde(default, skip_serializing_if = "Option::is_none")]
1777 pub level: Option<LogLevel>,
1778 #[serde(default, skip_serializing_if = "Vec::is_empty")]
1780 pub tail: Vec<String>,
1781}
1782
1783pub fn read_console_capture(
1790 path: &std::path::Path,
1791 from: Option<u64>,
1792 tail_bytes: u64,
1793) -> std::io::Result<(Vec<ConsoleLine>, u64)> {
1794 read_console_capture_window(path, from, tail_bytes, CONSOLE_READ_WINDOW)
1795 .map(|(lines, next, _)| (lines, next))
1796}
1797
1798pub const CONSOLE_READ_WINDOW: u64 = 16 * 1024 * 1024;
1800
1801pub fn read_console_capture_window(
1805 path: &std::path::Path,
1806 from: Option<u64>,
1807 tail_bytes: u64,
1808 window: u64,
1809) -> std::io::Result<(Vec<ConsoleLine>, u64, bool)> {
1810 use std::io::{Read, Seek, SeekFrom};
1811 let mut file = std::fs::File::open(path)?;
1812 let len = file.metadata()?.len();
1813 let mut start = from.unwrap_or_else(|| len.saturating_sub(tail_bytes));
1814 if start > len {
1815 start = 0;
1817 }
1818 file.seek(SeekFrom::Start(start))?;
1819 let mut buf = Vec::new();
1820 file.take(window).read_to_end(&mut buf)?;
1821 let more = start + (buf.len() as u64) < len;
1822 let mut lines = Vec::new();
1823 let mut offset = start;
1824 let mut consumed = 0usize;
1825 let mut first = from.is_none() && start > 0;
1826 for chunk in buf.split_inclusive(|b| *b == b'\n') {
1827 if !chunk.ends_with(b"\n") {
1828 break; }
1830 consumed += chunk.len();
1831 let rec_offset = offset;
1832 offset += chunk.len() as u64;
1833 if first {
1834 first = false;
1836 continue;
1837 }
1838 if let Ok(mut line) = serde_json::from_slice::<ConsoleLine>(chunk) {
1839 line.offset = rec_offset;
1840 if line.level.is_none() {
1841 line.level = infer_level(&line.text);
1842 }
1843 lines.push(line);
1844 }
1845 }
1846 let more = more && consumed > 0;
1849 Ok((lines, start + consumed as u64, more))
1850}
1851
1852#[must_use]
1855pub fn infer_level(text: &str) -> Option<LogLevel> {
1856 if text.contains("panicked at") || text.starts_with("error:") || text.starts_with("error[") {
1857 return Some(LogLevel::Error);
1858 }
1859 if text.starts_with("warning:") {
1860 return Some(LogLevel::Warn);
1861 }
1862 for token in text
1865 .split(|c: char| c.is_whitespace() || c == '[' || c == ']')
1866 .filter(|t| !t.is_empty())
1867 .take(4)
1868 {
1869 match token {
1870 "TRACE" => return Some(LogLevel::Trace),
1871 "DEBUG" => return Some(LogLevel::Debug),
1872 "INFO" => return Some(LogLevel::Info),
1873 "WARN" | "WARNING" => return Some(LogLevel::Warn),
1874 "ERROR" | "FATAL" => return Some(LogLevel::Error),
1875 _ => {}
1876 }
1877 }
1878 None
1879}
1880
1881#[cfg(test)]
1882#[path = "backend_logs_tests.rs"]
1883mod tests;