exfiltrate 0.3.0

// SPDX-License-Identifier: MIT OR Apache-2.0
//! Logwise integration for log capture and retrieval.
//!
//! # What changed, and why it was a bug
//!
//! The capture used to be an unbounded `Vec` that nothing ever removed from, and
//! `logwise_logs` cloned the whole thing, formatted every record, and returned
//! the history from record zero on every call. In a long-running program that is
//! an unbounded memory leak introduced by the debugging tool, and for a caller
//! polling for new output it means being handed the same growing blob over and
//! over.
//!
//! Three changes fix it, and they are the same three the panic buffer needed:
//!
//! * a bounded [`Ring`], so memory use has a ceiling;
//! * a cursor, so `--since` returns only what is new and formatting cost is
//!   proportional to the new slice rather than to the whole history;
//! * server-side `--level` and `--grep`, so a large history is filtered where it
//!   lives instead of being shipped over a socket to be filtered locally.
//!
//! Dropped records are counted and reported. A bounded buffer that quietly
//! returns less than it was asked for is worse than an unbounded one, because
//! the caller cannot tell the difference between "nothing happened" and "it
//! happened and I lost it".

use exfiltrate_internal::args::{ArgKind, ArgSpec, ParsedArgs};
use exfiltrate_internal::command::{Command, CommandContext, FileInfo, Response};
use logwise::{Level, LogRecord};
use std::pin::Pin;
use std::sync::{Arc, LazyLock};
use wasm_lite_std::Mutex;

#[derive(Debug)]
struct ExfiltrateLogger {
    records: Mutex<exfiltrate_internal::ring::Ring<LogRecord>>,
}

impl ExfiltrateLogger {
    fn new(capacity: usize) -> ExfiltrateLogger {
        ExfiltrateLogger {
            records: Mutex::new(exfiltrate_internal::ring::Ring::new(capacity)),
        }
    }
}

static LOGGER: LazyLock<Arc<ExfiltrateLogger>> = LazyLock::new(|| {
    // The capacity `begin_log_capture` was given, or the default if a record
    // arrives before it ran.
    Arc::new(ExfiltrateLogger::new(crate::config_snapshot().log_capacity))
});

impl logwise::Logger for ExfiltrateLogger {
    fn finish_log_record(&self, record: LogRecord) {
        self.records.with_mut_sync(|ring| {
            ring.push(record);
        });
    }

    fn finish_log_record_async<'s>(
        &'s self,
        record: LogRecord,
    ) -> Pin<Box<dyn Future<Output = ()> + Send + 's>> {
        Box::pin(self.records.with_mut_async(|ring| {
            ring.push(record);
        }))
    }

    fn prepare_to_die(&self) {}
}

/// Starts capturing logs from the `logwise` crate.
///
/// Adds a global logger that keeps the most recent `capacity` records in memory.
/// These logs can then be retrieved via the `logwise_logs` command.
pub fn begin_log_capture(capacity: usize) {
    LOGGER
        .records
        .with_mut_sync(|ring| ring.set_capacity(capacity));
    logwise::add_global_logger(LOGGER.clone());
    crate::add_command(LogwiseCapture);
}

/// The `logwise_logs` command.
pub struct LogwiseCapture;

static ARGS: &[ArgSpec] = &[
    ArgSpec::flag(
        "since",
        "only return records with a cursor at or after this value",
        ArgKind::Integer,
    ),
    ArgSpec::flag(
        "tail",
        "return only the last N matching records",
        ArgKind::Integer,
    ),
    ArgSpec::flag(
        "level",
        "only return records at this level or above",
        ArgKind::Enum(&[
            "trace",
            "debuginternal",
            "info",
            "analytics",
            "perfwarn",
            "warning",
            "error",
            "panic",
            "mandatory",
            "profile",
        ]),
    ),
    ArgSpec::flag(
        "grep",
        "only return records whose text contains this substring",
        ArgKind::String,
    ),
    ArgSpec::flag(
        "text",
        "return the records inline as text instead of as a file attachment",
        ArgKind::Bool,
    ),
    ArgSpec::flag(
        "follow",
        "keep streaming new records as they arrive instead of returning",
        ArgKind::Bool,
    ),
];

impl Command for LogwiseCapture {
    fn name(&self) -> &'static str {
        "logwise_logs"
    }

    fn short_description(&self) -> &'static str {
        "Shows logwise logs.  Use this to stream logs from a running Rust program.  ALWAYS use this to read
        logs on wasm32-unknown-unknown, since other methods are broken."
    }

    fn full_description(&self) -> &'static str {
        "Shows logwise logs.

In some cases, logs may be difficult to access.  For example we may be debugging WASM code, running in a browser, or a remote computer.

Often, on wasm, only the main thread's logs are printed.  So if you are reading stdout, you are missing many logs that are being written by other threads.  So the output from other sources may be HIGHLY misleading.

Using this command ensures you get all the logwise logs from all threads, from the point `exfiltrate::begin` was called onwards.  (Logs prior to this call are not captured; so users are instructed to make this call early in their program).

The buffer is bounded — see `Config::log_capacity` — so old records are eventually
dropped. The output always reports how many were dropped rather than quietly
returning a shorter history than you asked for.

Every call ends with a `cursor=` line. Pass it back as `--since` and you get only
what has arrived since, which is what makes repeated calls cheap: nothing older is
re-formatted or re-sent.

`--level` and `--grep` filter on the server, so a large history is narrowed before it
crosses the wire rather than after.

`--follow` streams new records as they are produced instead of returning. It needs a
client that understands streaming; without one it returns the current contents and
says so.

For more information on using logwise, try building the latest documentation for it.  Alternatively, some resources are
        * https://sealedabstract.com/code/logwise
        * https://docs.rs/logwise/latest/logwise/
"
    }

    fn args(&self) -> &'static [ArgSpec] {
        ARGS
    }

    fn execute(&self, args: Vec<String>) -> Result<Response, Response> {
        self.execute_with(args, &CommandContext::detached())
    }

    fn execute_with(
        &self,
        args: Vec<String>,
        context: &CommandContext,
    ) -> Result<Response, Response> {
        let parsed = ParsedArgs::parse(self.args(), args).map_err(Response::String)?;
        let query = Query::from(&parsed);

        if parsed.boolean("follow") {
            if !context.supports_streaming() {
                let (text, cursor) = collect(&query);
                return Ok(format!(
                    "{text}\n--follow needs a client that understands streaming; \
                     returning the current contents instead. Poll with --since {cursor}.\n"
                )
                .into());
            }
            return follow(query, context);
        }

        let (text, _) = collect(&query);
        if parsed.boolean("text") {
            Ok(text.into())
        } else {
            Ok(Response::Files(vec![FileInfo::new(
                "log".to_string(),
                None,
                text.into_bytes(),
            )]))
        }
    }
}

/// A parsed `logwise_logs` request.
struct Query {
    since: u64,
    tail: Option<usize>,
    level: Option<Level>,
    grep: Option<String>,
}

impl Query {
    fn from(parsed: &ParsedArgs) -> Query {
        Query {
            since: parsed.integer("since").unwrap_or(0).max(0) as u64,
            tail: parsed
                .integer("tail")
                .filter(|tail| *tail > 0)
                .map(|tail| tail as usize),
            level: parsed.get("level").and_then(parse_level),
            grep: parsed.get("grep").map(str::to_string),
        }
    }

    fn matches(&self, record: &LogRecord) -> bool {
        if let Some(minimum) = self.level
            && record.level() < minimum
        {
            return false;
        }
        match &self.grep {
            Some(needle) => record.to_string().contains(needle),
            None => true,
        }
    }
}

/// Maps the `--level` flag onto a `logwise::Level`.
///
/// The flag is an `ArgKind::Enum`, so an unrecognised value never reaches here;
/// the `None` arm exists because `Level` is `#[non_exhaustive]` upstream and a
/// future variant should not become a compile error in this crate.
fn parse_level(name: &str) -> Option<Level> {
    match name {
        "trace" => Some(Level::Trace),
        "debuginternal" => Some(Level::DebugInternal),
        "info" => Some(Level::Info),
        "analytics" => Some(Level::Analytics),
        "perfwarn" => Some(Level::PerfWarn),
        "warning" => Some(Level::Warning),
        "error" => Some(Level::Error),
        "panic" => Some(Level::Panic),
        "mandatory" => Some(Level::Mandatory),
        "profile" => Some(Level::Profile),
        _ => None,
    }
}

/// Formats the records matching `query`, returning the text and the next cursor.
fn collect(query: &Query) -> (String, u64) {
    let (lines, next_cursor, missed, dropped_total, returned) = LOGGER.records.with_sync(|ring| {
        let slice = ring.since(query.since, query.tail, |record| query.matches(record));
        let mut lines = String::new();
        for record in &slice.records {
            lines.push_str(&record.to_string());
            lines.push('\n');
        }
        (
            lines,
            slice.next_cursor,
            slice.missed,
            slice.dropped_total,
            slice.records.len(),
        )
    });

    let mut out = lines;
    if missed > 0 {
        out.push_str(&format!(
            "\n{missed} record(s) were dropped before this call could read them.\n"
        ));
    }
    out.push_str(&format!(
        "\ncursor={next_cursor} returned={returned} dropped_total={dropped_total}\n"
    ));
    (out, next_cursor)
}

/// Streams matching records until the client cancels.
fn follow(mut query: Query, context: &CommandContext) -> Result<Response, Response> {
    // `--tail` selects the last N of the backlog; after that it must not keep
    // re-applying, or a steady stream would be silently thinned.
    let mut first = true;
    loop {
        context.check_cancelled()?;
        let (batch, next_cursor) = LOGGER.records.with_sync(|ring| {
            let slice = ring.since(
                query.since,
                if first { query.tail } else { None },
                |record| query.matches(record),
            );
            let mut lines = String::new();
            for record in &slice.records {
                lines.push_str(&record.to_string());
                lines.push('\n');
            }
            (lines, slice.next_cursor)
        });
        first = false;
        query.since = next_cursor;
        if !batch.is_empty() && context.emit(batch).is_err() {
            // The client is gone. Ending quietly is right: there is nobody left
            // to report an error to.
            return Ok(Response::String(String::new()));
        }
        wasm_lite_std::sleep(exfiltrate_internal::wire::BACKOFF_DURATION);
    }
}

#[cfg(test)]
mod tests {
    use super::*;

    fn record(level: Level, message: &str) -> LogRecord {
        let mut record = LogRecord::new(level);
        record.log(message);
        record
    }

    fn push(level: Level, message: &str) -> u64 {
        LOGGER
            .records
            .with_mut_sync(|ring| ring.push(record(level, message)))
    }

    fn query(since: u64) -> Query {
        Query {
            since,
            tail: None,
            level: None,
            grep: None,
        }
    }

    #[test]
    fn a_cursor_returns_only_what_arrived_since_the_last_call() {
        let start = LOGGER.records.with_sync(|ring| ring.next_cursor());
        push(Level::Warning, "logwise_test_first");
        let (text, cursor) = collect(&query(start));
        assert!(text.contains("logwise_test_first"), "{text}");

        push(Level::Warning, "logwise_test_second");
        let (text, _) = collect(&query(cursor));
        assert!(!text.contains("logwise_test_first"), "{text}");
        assert!(text.contains("logwise_test_second"), "{text}");
        assert!(text.contains("returned=1"), "{text}");
    }

    #[test]
    fn the_level_filter_drops_anything_quieter() {
        let start = LOGGER.records.with_sync(|ring| ring.next_cursor());
        push(Level::Info, "logwise_test_info_line");
        push(Level::Error, "logwise_test_error_line");
        let (text, _) = collect(&Query {
            level: Some(Level::Error),
            ..query(start)
        });
        assert!(!text.contains("logwise_test_info_line"), "{text}");
        assert!(text.contains("logwise_test_error_line"), "{text}");
    }

    #[test]
    fn grep_matches_the_rendered_text() {
        let start = LOGGER.records.with_sync(|ring| ring.next_cursor());
        push(Level::Warning, "logwise_test_needle_here");
        push(Level::Warning, "logwise_test_something_else");
        let (text, _) = collect(&Query {
            grep: Some("needle".to_string()),
            ..query(start)
        });
        assert!(text.contains("logwise_test_needle_here"), "{text}");
        assert!(!text.contains("logwise_test_something_else"), "{text}");
    }

    #[test]
    fn tail_takes_the_last_matching_records_not_the_last_records() {
        let start = LOGGER.records.with_sync(|ring| ring.next_cursor());
        push(Level::Error, "logwise_test_tail_a");
        push(Level::Info, "logwise_test_tail_noise");
        push(Level::Error, "logwise_test_tail_b");
        push(Level::Info, "logwise_test_tail_noise");
        let (text, _) = collect(&Query {
            tail: Some(2),
            level: Some(Level::Error),
            ..query(start)
        });
        assert!(text.contains("logwise_test_tail_a"), "{text}");
        assert!(text.contains("logwise_test_tail_b"), "{text}");
        assert!(!text.contains("noise"), "{text}");
    }

    #[test]
    fn the_output_always_ends_with_a_cursor_line() {
        let (text, cursor) = collect(&query(u64::MAX));
        assert!(text.contains(&format!("cursor={cursor}")), "{text}");
        assert!(text.contains("dropped_total="), "{text}");
    }

    #[test]
    fn the_ring_bounds_memory_and_says_what_it_dropped() {
        let previous = LOGGER.records.with_sync(|ring| ring.capacity());
        LOGGER.records.with_mut_sync(|ring| ring.set_capacity(4));
        let start = LOGGER.records.with_sync(|ring| ring.next_cursor());
        for index in 0..20 {
            push(Level::Warning, &format!("logwise_test_bounded_{index}"));
        }
        assert_eq!(LOGGER.records.with_sync(|ring| ring.len()), 4);

        let (text, _) = collect(&query(start));
        assert!(text.contains("logwise_test_bounded_19"), "{text}");
        assert!(!text.contains("logwise_test_bounded_0\n"), "{text}");
        assert!(text.contains("record(s) were dropped"), "{text}");

        LOGGER
            .records
            .with_mut_sync(|ring| ring.set_capacity(previous));
    }

    #[test]
    fn follow_without_a_streaming_client_returns_the_backlog_and_explains() {
        let response = LogwiseCapture
            .execute(vec!["--follow".to_string()])
            .unwrap()
            .into_string();
        assert!(
            response.contains("needs a client that understands streaming"),
            "{response}"
        );
        assert!(response.contains("--since"), "{response}");
    }

    #[test]
    fn every_level_the_flag_offers_maps_to_a_real_level() {
        for name in [
            "trace",
            "debuginternal",
            "info",
            "analytics",
            "perfwarn",
            "warning",
            "error",
            "panic",
            "mandatory",
            "profile",
        ] {
            assert!(parse_level(name).is_some(), "{name} did not map");
        }
    }

    #[test]
    fn text_mode_returns_a_string_and_the_default_returns_a_file() {
        let text = LogwiseCapture.execute(vec!["--text".to_string()]).unwrap();
        assert!(matches!(text, Response::String(_)));
        let file = LogwiseCapture.execute(Vec::new()).unwrap();
        assert!(matches!(file, Response::Files(_)));
    }
}