1use scope::metrics::{Metric, Sample};
38use std::collections::VecDeque;
39use std::fmt;
40use std::io::{self, Write};
41use std::sync::atomic::{AtomicBool, Ordering};
42use std::sync::{Arc, Mutex};
43use std::time::Instant;
44
45type Sink = Arc<Mutex<dyn Write + Send>>;
47
48#[derive(Debug, Clone, Default)]
53pub struct RunProgress {
54 pub run: u64,
56 pub total: Option<u64>,
58 pub next_at: Option<Instant>,
60}
61
62pub type ProgressHook = Arc<Mutex<RunProgress>>;
64
65#[derive(Debug, Clone)]
67pub struct RunRecord {
68 pub run: u64,
70 pub at: chrono::DateTime<chrono::Utc>,
72 pub sample: Sample,
74 pub snapshot: String,
76 pub view: Option<Arc<SummaryView>>,
78}
79
80#[derive(Debug, Clone)]
82pub struct SummaryView {
83 pub summary: scope::market::MarketSummary,
84 pub venue: String,
86 pub tokens: Vec<scope::market::assets::TokenAddress>,
88}
89
90pub type ViewHook = Arc<Mutex<Option<SummaryView>>>;
92
93#[derive(Debug, Default)]
95pub struct Series {
96 pub metrics: Vec<Metric>,
98 pub runs: VecDeque<RunRecord>,
100 pub dropped: u64,
102}
103
104pub type SeriesHook = Arc<Mutex<Series>>;
106
107pub struct Ask {
110 pub prompt: String,
111 pub options: Vec<String>,
112 pub reply: tokio::sync::oneshot::Sender<Option<usize>>,
113}
114
115pub type Asker = Arc<Mutex<Option<Ask>>>;
118
119pub const MAX_SERIES_RUNS: usize = 500;
121
122#[derive(Clone)]
126pub struct Output {
127 out: Sink,
128 err: Sink,
129 progress: bool,
130 hook: Option<ProgressHook>,
131 series: Option<SeriesHook>,
132 tui: bool,
133 prompts: bool,
135 asker: Option<Asker>,
137 view: Option<ViewHook>,
139}
140
141impl Output {
142 pub fn stdio() -> Self {
145 Self {
146 out: Arc::new(Mutex::new(io::stdout())),
147 err: Arc::new(Mutex::new(io::stderr())),
148 progress: true,
149 hook: None,
150 series: None,
151 tui: false,
152 prompts: true,
153 asker: None,
154 view: None,
155 }
156 }
157
158 pub fn capture() -> (Self, Captured) {
163 let out = Arc::new(Mutex::new(Vec::new()));
164 let err = Arc::new(Mutex::new(Vec::new()));
165 let captured = Captured {
166 data: out.clone(),
167 out: out.clone(),
168 err: err.clone(),
169 trimmed: Arc::default(),
170 };
171 let output = Self {
172 out,
173 err,
174 progress: false,
175 hook: None,
176 series: None,
177 tui: false,
178 prompts: false,
179 asker: None,
180 view: None,
181 };
182 (output, captured)
183 }
184
185 pub fn capture_merged() -> (Self, Captured) {
193 let buf = Arc::new(Mutex::new(Vec::new()));
194 let data = Arc::new(Mutex::new(Vec::new()));
195 let captured = Captured {
196 out: buf.clone(),
197 err: buf.clone(),
198 data: data.clone(),
199 trimmed: Arc::default(),
200 };
201 let output = Self {
202 out: Arc::new(Mutex::new(Tee {
203 merged: buf.clone(),
204 data,
205 })),
206 err: buf,
207 progress: false,
208 hook: Some(ProgressHook::default()),
209 series: Some(SeriesHook::default()),
210 tui: true,
211 prompts: false,
212 asker: None,
213 view: Some(ViewHook::default()),
214 };
215 (output, captured)
216 }
217
218 pub fn teed(&self, limit: usize) -> (Self, Captured) {
232 let buf = Arc::new(Mutex::new(Vec::new()));
233 let data = Arc::new(Mutex::new(Vec::new()));
234 let trimmed = Arc::new(AtomicBool::new(false));
235 let captured = Captured {
236 out: buf.clone(),
237 err: buf.clone(),
238 data: data.clone(),
239 trimmed: trimmed.clone(),
240 };
241 let copy = |real: &Sink, data: Option<Arc<Mutex<Vec<u8>>>>| -> Sink {
242 Arc::new(Mutex::new(TeeSink {
243 real: real.clone(),
244 merged: buf.clone(),
245 data,
246 limit,
247 trimmed: trimmed.clone(),
248 }))
249 };
250 let output = Self {
251 out: copy(&self.out, Some(data.clone())),
252 err: copy(&self.err, None),
253 progress: self.progress,
254 hook: None,
255 series: None,
256 tui: true,
257 prompts: self.prompts,
258 asker: None,
259 view: None,
260 };
261 (output, captured)
262 }
263
264 #[cfg(test)]
266 pub(crate) fn from_writers(
267 out: impl Write + Send + 'static,
268 err: impl Write + Send + 'static,
269 ) -> Self {
270 Self {
271 out: Arc::new(Mutex::new(out)),
272 err: Arc::new(Mutex::new(err)),
273 progress: false,
274 hook: None,
275 series: None,
276 tui: false,
277 prompts: true,
278 asker: None,
279 view: None,
280 }
281 }
282
283 pub fn progress_enabled(&self) -> bool {
285 self.progress
286 }
287
288 pub fn without_prompts(mut self) -> Self {
291 self.prompts = false;
292 self
293 }
294
295 pub fn with_asker(mut self, asker: Asker) -> Self {
298 self.asker = Some(asker);
299 self
300 }
301
302 pub fn asker(&self) -> Option<Asker> {
304 self.asker.clone()
305 }
306
307 pub fn prompts_allowed(&self) -> bool {
309 self.prompts
310 }
311
312 pub fn is_tui(&self) -> bool {
315 self.tui
316 }
317
318 pub fn progress_hook(&self) -> Option<ProgressHook> {
321 self.hook.clone()
322 }
323
324 pub fn series_hook(&self) -> Option<SeriesHook> {
327 self.series.clone()
328 }
329
330 pub fn view_hook(&self) -> Option<ViewHook> {
332 self.view.clone()
333 }
334
335 pub fn record_view(&self, view: SummaryView) -> bool {
338 let Some(slot) = &self.view else {
339 return false;
340 };
341 *slot.lock().unwrap_or_else(|e| e.into_inner()) = Some(view);
342 true
343 }
344
345 pub fn record_run(&self, metrics: &[Metric], record: RunRecord) -> bool {
348 let Some(series) = &self.series else {
349 return false;
350 };
351 let mut series = lock(series);
352 if series.metrics.is_empty() {
353 series.metrics = metrics.to_vec();
354 }
355 series.runs.push_back(record);
356 while series.runs.len() > MAX_SERIES_RUNS {
357 series.runs.pop_front();
358 series.dropped += 1;
359 }
360 true
361 }
362
363 pub fn update_progress(&self, change: impl FnOnce(&mut RunProgress)) {
365 if let Some(hook) = &self.hook {
366 change(&mut lock(hook));
367 }
368 }
369
370 pub fn write_out(&self, args: fmt::Arguments<'_>) -> io::Result<()> {
377 emit(&self.out, args)
378 }
379
380 pub fn write_err(&self, args: fmt::Arguments<'_>) -> io::Result<()> {
386 emit(&self.err, args)
387 }
388
389 pub fn note(&self, args: fmt::Arguments<'_>) {
393 let _ = emit(&self.err, args);
394 }
395
396 pub fn note_saved(&self, what: &str, path: impl fmt::Display) {
398 crate::note!(self, "{what} saved to: {path}");
399 }
400}
401
402#[derive(Clone)]
404pub struct Captured {
405 data: Arc<Mutex<Vec<u8>>>,
407 out: Arc<Mutex<Vec<u8>>>,
408 err: Arc<Mutex<Vec<u8>>>,
409 trimmed: Arc<AtomicBool>,
411}
412
413impl Captured {
414 pub fn trimmed(&self) -> bool {
417 self.trimmed.load(Ordering::Relaxed)
418 }
419
420 pub fn out(&self) -> String {
422 String::from_utf8_lossy(&lock(&self.out)).into_owned()
423 }
424
425 pub fn data(&self) -> String {
428 String::from_utf8_lossy(&lock(&self.data)).into_owned()
429 }
430
431 pub fn err(&self) -> String {
433 String::from_utf8_lossy(&lock(&self.err)).into_owned()
434 }
435
436 pub fn out_sink(&self) -> scope::diag::Sink {
439 self.out.clone()
440 }
441}
442
443struct Tee {
446 merged: Arc<Mutex<Vec<u8>>>,
447 data: Arc<Mutex<Vec<u8>>>,
448}
449
450impl Write for Tee {
451 fn write(&mut self, buf: &[u8]) -> io::Result<usize> {
452 lock(&self.merged).extend_from_slice(buf);
453 lock(&self.data).extend_from_slice(buf);
454 Ok(buf.len())
455 }
456 fn flush(&mut self) -> io::Result<()> {
457 Ok(())
458 }
459}
460
461struct TeeSink {
464 real: Sink,
465 merged: Arc<Mutex<Vec<u8>>>,
466 data: Option<Arc<Mutex<Vec<u8>>>>,
468 limit: usize,
469 trimmed: Arc<AtomicBool>,
470}
471
472impl Write for TeeSink {
473 fn write(&mut self, buf: &[u8]) -> io::Result<usize> {
476 for copy in std::iter::once(&self.merged).chain(self.data.as_ref()) {
477 let mut copy = lock(copy);
478 copy.extend_from_slice(buf);
479 if keep_tail(&mut copy, self.limit) {
480 self.trimmed.store(true, Ordering::Relaxed);
481 }
482 }
483 lock(&self.real).write_all(buf)?;
484 Ok(buf.len())
485 }
486 fn flush(&mut self) -> io::Result<()> {
487 lock(&self.real).flush()
488 }
489}
490
491fn keep_tail(buf: &mut Vec<u8>, limit: usize) -> bool {
497 if buf.len() <= limit {
498 return false;
499 }
500 let cut = buf.len() - limit / 2;
501 let start = buf[cut..]
502 .iter()
503 .position(|&b| b == b'\n')
504 .map(|i| cut + i + 1)
505 .filter(|&start| start < buf.len())
506 .unwrap_or_else(|| whole_cut(buf, cut));
507 buf.drain(..start);
508 true
509}
510
511fn whole_cut(buf: &[u8], cut: usize) -> usize {
515 let mut i = 0;
516 while i < cut {
517 i = crate::cli::ansi::escape_end(buf, i).unwrap_or(i + 1);
518 }
519 let continuation = |b: u8| b & 0b1100_0000 == 0b1000_0000;
520 buf[i.min(buf.len())..]
521 .iter()
522 .position(|&b| !continuation(b))
523 .map_or(buf.len(), |n| i + n)
524}
525
526fn lock<T: ?Sized>(m: &Mutex<T>) -> std::sync::MutexGuard<'_, T> {
529 m.lock().unwrap_or_else(|poisoned| poisoned.into_inner())
530}
531
532fn emit(sink: &Mutex<dyn Write + Send>, args: fmt::Arguments<'_>) -> io::Result<()> {
537 lock(sink).write_fmt(args)
538}
539
540#[macro_export]
542macro_rules! outln {
543 ($o:expr) => {
544 $o.write_out(format_args!("\n"))
545 };
546 ($o:expr, $($arg:tt)*) => {
547 $o.write_out(format_args!("{}\n", format_args!($($arg)*)))
548 };
549}
550
551#[macro_export]
553macro_rules! out {
554 ($o:expr, $($arg:tt)*) => {
555 $o.write_out(format_args!($($arg)*))
556 };
557}
558
559#[macro_export]
561macro_rules! errln {
562 ($o:expr) => {
563 $o.write_err(format_args!("\n"))
564 };
565 ($o:expr, $($arg:tt)*) => {
566 $o.write_err(format_args!("{}\n", format_args!($($arg)*)))
567 };
568}
569
570#[macro_export]
573macro_rules! note {
574 ($o:expr) => {
575 $o.note(format_args!("\n"))
576 };
577 ($o:expr, $($arg:tt)*) => {
578 $o.note(format_args!("{}\n", format_args!($($arg)*)))
579 };
580}
581
582#[macro_export]
584macro_rules! err {
585 ($o:expr, $($arg:tt)*) => {
586 $o.write_err(format_args!($($arg)*))
587 };
588}
589
590#[cfg(test)]
595pub(crate) mod tests {
596 use super::*;
597
598 #[test]
599 fn test_capture_separates_data_from_diagnostics() {
600 let (o, cap) = Output::capture();
602 outln!(o, "{{\"ok\":{}}}", true).unwrap();
603 errln!(o, "Warning: {}", "partial data").unwrap();
604 assert_eq!(cap.out(), "{\"ok\":true}\n");
605 assert_eq!(cap.err(), "Warning: partial data\n");
606 }
607
608 #[test]
609 fn test_a_note_goes_to_the_diagnostic_channel_and_never_fails() {
610 let (o, cap) = Output::capture();
613 note!(o, "Fetching {}...", "USDC");
614 assert_eq!(cap.err(), "Fetching USDC...\n");
615 assert_eq!(cap.out(), "");
616 let closed = Output::from_writers(Vec::new(), ClosedPipe);
617 note!(closed, "dropped");
618 }
619
620 #[test]
621 fn test_note_saved_is_one_line_on_the_diagnostic_channel() {
622 let (o, cap) = Output::capture();
623 o.note_saved("Report", "r.md");
624 assert_eq!(cap.err(), "Report saved to: r.md\n");
625 assert_eq!(cap.out(), "");
626 }
627
628 #[test]
629 fn test_clones_share_writers() {
630 let (o, cap) = Output::capture();
632 let o2 = o.clone();
633 out!(o, "a").unwrap();
634 out!(o2, "b").unwrap();
635 outln!(o).unwrap();
636 assert_eq!(cap.out(), "ab\n");
637 }
638
639 pub(crate) struct ClosedPipe;
641
642 impl Write for ClosedPipe {
643 fn write(&mut self, _: &[u8]) -> io::Result<usize> {
644 Err(io::Error::new(io::ErrorKind::BrokenPipe, "closed"))
645 }
646 fn flush(&mut self) -> io::Result<()> {
647 Ok(())
648 }
649 }
650
651 #[test]
652 fn test_merged_capture_keeps_data_apart_from_diagnostics() {
653 let (o, cap) = Output::capture_merged();
654 outln!(o, "data").unwrap();
655 errln!(o, "warn").unwrap();
656 assert_eq!(cap.data(), "data\n");
657 assert_eq!(cap.out(), "data\nwarn\n");
658 }
659
660 #[test]
661 fn test_failed_write_is_returned_not_dropped() {
662 let o = Output::from_writers(ClosedPipe, ClosedPipe);
664 let e = outln!(o, "data").unwrap_err();
665 assert_eq!(e.kind(), io::ErrorKind::BrokenPipe);
666 assert_eq!(
667 errln!(o, "warn").unwrap_err().kind(),
668 io::ErrorKind::BrokenPipe
669 );
670 }
671
672 #[test]
673 fn test_only_the_tui_capture_is_tui() {
674 assert!(Output::capture_merged().0.is_tui());
675 assert!(!Output::capture().0.is_tui());
676 assert!(!Output::stdio().is_tui());
677 assert!(Output::stdio().teed(1).0.is_tui());
678 }
679
680 #[test]
681 fn test_capture_merged_keeps_write_order() {
682 let (o, cap) = Output::capture_merged();
685 outln!(o, "row 1").unwrap();
686 errln!(o, "warning").unwrap();
687 outln!(o, "row 2").unwrap();
688 assert_eq!(cap.out(), "row 1\nwarning\nrow 2\n");
689 }
690
691 #[test]
692 fn test_teed_output_reaches_the_terminal_and_keeps_a_copy() {
693 let (terminal, screen) = Output::capture();
696 let (o, copy) = terminal.teed(1024);
697 outln!(o, "row").unwrap();
698 errln!(o, "warn").unwrap();
699 assert_eq!(screen.out(), "row\n");
700 assert_eq!(screen.err(), "warn\n");
701 assert_eq!(copy.out(), "row\nwarn\n");
702 assert_eq!(copy.data(), "row\n");
703 assert!(!copy.trimmed());
704 assert!(o.is_tui());
707 assert_eq!(o.progress_enabled(), terminal.progress_enabled());
708 assert!(Output::stdio().teed(1).0.progress_enabled());
709 assert!(o.progress_hook().is_none());
710 }
711
712 #[test]
713 fn test_teed_copy_keeps_whole_lines_from_the_end_within_its_limit() {
714 let (terminal, screen) = Output::capture();
716 let (o, copy) = terminal.teed(100);
717 for i in 0..1000 {
718 outln!(o, "line {i:04}").unwrap();
719 }
720 assert!(copy.trimmed());
721 let kept = copy.out();
722 assert!(kept.len() <= 100, "{}", kept.len());
723 assert!(kept.starts_with("line "), "whole lines: {kept:?}");
724 assert!(kept.ends_with("line 0999\n"), "{kept:?}");
725 assert!(copy.data().ends_with("line 0999\n"));
726 assert_eq!(screen.out().lines().count(), 1000, "the terminal got all");
727 }
728
729 #[test]
730 fn test_teed_copy_keeps_the_tail_of_one_long_line() {
731 let (terminal, _screen) = Output::capture();
734 let (o, copy) = terminal.teed(100);
735 let blob: String = (0..300)
736 .map(|i| char::from(b'a' + (i % 26) as u8))
737 .collect();
738 out!(o, "{blob}").unwrap();
739 assert!(copy.trimmed());
740 let kept = copy.out();
741 assert!(!kept.is_empty() && kept.len() <= 100, "{}", kept.len());
742 assert!(blob.ends_with(&kept), "the end of the line: {kept:?}");
743 }
744
745 #[test]
746 fn test_trim_of_one_line_splits_no_character_and_no_escape() {
747 let mut chars = "é".repeat(60).into_bytes(); assert!(keep_tail(&mut chars, 102)); assert!(String::from_utf8(chars.clone()).is_ok(), "{chars:?}");
752 assert_eq!(chars.len(), 50);
753 let mut colored = b"x".repeat(60);
754 colored.extend_from_slice(b"\x1b[38;5;196mred");
755 assert!(keep_tail(&mut colored, 20));
757 assert_eq!(colored, b"red");
758 let mut link = b"y".repeat(60);
759 link.extend_from_slice(b"\x1b]8;;https://example.com/a/long/path\x1b\\text");
760 let cut_at = link.len() - 40 / 2;
762 assert!(cut_at > 60 && cut_at < 98, "{cut_at}");
763 assert!(keep_tail(&mut link, 40));
764 assert_eq!(link, b"text");
765 }
766
767 #[test]
768 fn test_teed_copy_keeps_text_the_terminal_refused() {
769 let (o, copy) = Output::from_writers(ClosedPipe, ClosedPipe).teed(64);
771 assert!(outln!(o, "data").is_err());
772 assert_eq!(copy.out(), "data\n");
773 }
774
775 #[test]
776 fn test_capture_disables_progress() {
777 let (o, _) = Output::capture();
779 assert!(!o.progress_enabled());
780 assert!(Output::stdio().progress_enabled());
781 }
782}