launcher/src/web/logs.rs

//! Log window attachments — tracked SSH `docker logs -f` streams with a
//! ring-buffer history per (cluster, node, recipe, rank). `POST /api/logs`
//! attaches (idempotent per key), `DELETE /api/logs` detaches, `GET
//! /api/logs` returns ring history for a fresh window. Lines stream to the
//! SSE bus as `Event::Log`.

use std::collections::VecDeque;
use std::sync::atomic::{AtomicBool, Ordering};
use std::sync::{Arc, Mutex};

use tokio::task::JoinHandle;

use crate::ssh::shq;
use crate::web::events::Event;
use crate::web::ssh_stream::stream_lines;

/// Max concurrent log streams (each holds an SSH connection).
pub const MAX_ATTACHES: usize = 8;
/// Ring-buffer depth per attach (lines).
pub const RING_LINES: usize = 1000;
/// Lines of history a fresh window starts with (docker `--tail`).
pub const TAIL_LINES: usize = 200;

/// One live (or finished) log stream.
pub struct LogAttach {
    pub id: String,
    pub cluster: String,
    /// Node name (SSH target).
    pub node: String,
    pub recipe: String,
    pub rank: u32,
    /// Bounded history for `GET /api/logs` and window re-opens.
    pub history: Arc<Mutex<VecDeque<String>>>,
    /// Set on detach; the stream task ends at its next line.
    pub stop: Arc<AtomicBool>,
    /// True once the stream ended on its own (container stopped, SSH drop).
    pub done: Arc<AtomicBool>,
    /// The streaming task (aborted on detach).
    pub task: JoinHandle<()>,
}

type AttachKey = (String, String, String, u32);

impl LogAttach {
    fn key(&self) -> AttachKey {
        (
            self.cluster.clone(),
            self.node.clone(),
            self.recipe.clone(),
            self.rank,
        )
    }

    /// Last `n` history lines, oldest first.
    pub fn history_last(&self, n: usize) -> Vec<String> {
        let ring = self.history.lock().unwrap();
        ring.iter()
            .rev()
            .take(n)
            .map(String::clone)
            .collect::<Vec<_>>()
            .into_iter()
            .rev()
            .collect()
    }
}

/// `docker logs -f --tail=N --timestamps` for one recipe+rank's container(s)
/// on the node — seeds the window with history, then follows.
pub fn logs_stream_cmd(recipe: &str, rank: u32, tail: usize) -> String {
    format!(
        "docker ps -q --filter {} --filter label=sm121.rank={} \
         | xargs -r docker logs -f --tail={tail} --timestamps 2>&1",
        shq(&format!("label=sm121.recipe={}", recipe)),
        rank
    )
}

/// Append a line to a ring, evicting the oldest at capacity.
pub(crate) fn push_line(ring: &Mutex<VecDeque<String>>, line: &str) {
    let mut r = ring.lock().unwrap();
    if r.len() >= RING_LINES {
        r.pop_front();
    }
    r.push_back(line.to_string());
}

/// Idempotent per (cluster, node, recipe, rank): a live stream for the same
/// key is reused; a finished one is replaced.
pub fn attach(
    app: &crate::web::AppState,
    attaches: &Mutex<Vec<LogAttach>>,
    cluster: &str,
    node: &str,
    recipe: &str,
    rank: u32,
) -> Result<(String, Vec<String>), String> {
    let config = &app.config;
    let cl = config
        .clusters
        .iter()
        .find(|c| c.name == cluster)
        .ok_or_else(|| format!("unknown cluster '{cluster}'"))?;
    let node_cfg = cl
        .nodes
        .iter()
        .find(|n| n.name == node)
        .ok_or_else(|| format!("unknown node '{node}' in cluster '{cluster}'"))?;
    let key: AttachKey = (
        cluster.to_string(),
        node.to_string(),
        recipe.to_string(),
        rank,
    );

    let mut list = attaches.lock().unwrap();
    if let Some(a) = list.iter().find(|a| a.key() == key) {
        if !a.done.load(Ordering::Relaxed) {
            return Ok((a.id.clone(), a.history_last(TAIL_LINES)));
        }
        // Finished stream — drop it and start fresh below.
        let pos = list.iter().position(|x| x.id == a.id).unwrap();
        list.remove(pos);
    }
    if list.len() >= MAX_ATTACHES {
        return Err(format!(
            "max {MAX_ATTACHES} log attaches — close a window first"
        ));
    }

    let id = format!("l{}", list.len() + 1);
    let history: Arc<Mutex<VecDeque<String>>> =
        Arc::new(Mutex::new(VecDeque::with_capacity(RING_LINES)));
    let stop = Arc::new(AtomicBool::new(false));
    let done = Arc::new(AtomicBool::new(false));
    let cmd = logs_stream_cmd(recipe, rank, TAIL_LINES);
    let (h, s, d) = (history.clone(), stop.clone(), done.clone());
    let (cfg, node_obj, bus) = (config.clone(), node_cfg.clone(), app.bus.clone());
    let (idc, cluster_s, node_s, recipe_s) = (
        id.clone(),
        cl.name.clone(),
        node_cfg.name.clone(),
        recipe.to_string(),
    );
    let appc = app.clone();
    let task = tokio::spawn(async move {
        let _ = stream_lines(&cfg, &node_obj, cmd, s, |line| {
            push_line(&h, line);
            let _ = bus.send(Event::Log {
                attach: idc.clone(),
                cluster: cluster_s.clone(),
                node: node_s.clone(),
                recipe: recipe_s.clone(),
                rank,
                line: line.to_string(),
            });
        })
        .await;
        // Stream errors (node offline, container gone) just end the window —
        // refresh the board so attaches[].done (and fleet changes) reach the
        // UI immediately instead of on the next 30 s tick.
        d.store(true, Ordering::Relaxed);
        crate::web::push_state(&appc).await;
    });

    let attach = LogAttach {
        id: id.clone(),
        cluster: cluster.to_string(),
        node: node.to_string(),
        recipe: recipe.to_string(),
        rank,
        history: history.clone(),
        stop,
        done,
        task,
    };
    list.push(attach);
    Ok((id, vec![]))
}

/// Detach: stop the stream, abort the task, drop the entry.
pub fn detach(attaches: &Mutex<Vec<LogAttach>>, attach_id: &str) -> bool {
    let mut list = attaches.lock().unwrap();
    if let Some(pos) = list.iter().position(|a| a.id == attach_id) {
        list[pos].stop.store(true, Ordering::Relaxed);
        list[pos].task.abort();
        list.remove(pos);
        true
    } else {
        false
    }
}

/// Find an attach by key (for `GET /api/logs?cluster=&node=&recipe=&rank=`).
pub fn by_key(
    attaches: &Mutex<Vec<LogAttach>>,
    cluster: &str,
    node: &str,
    recipe: &str,
    rank: u32,
    lines: usize,
) -> Option<(String, Vec<String>)> {
    let list = attaches.lock().unwrap();
    list.iter()
        .find(|a| {
            a.key()
                == (
                    cluster.to_string(),
                    node.to_string(),
                    recipe.to_string(),
                    rank,
                )
        })
        .map(|a| (a.id.clone(), a.history_last(lines)))
}

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

    fn ring() -> Arc<Mutex<VecDeque<String>>> {
        Arc::new(Mutex::new(VecDeque::new()))
    }

    #[test]
    fn push_line_trims_to_capacity_keeping_newest() {
        let r = ring();
        for i in 0..=RING_LINES {
            push_line(&r, &format!("line-{i}"));
        }
        let g = r.lock().unwrap();
        assert_eq!(g.len(), RING_LINES, "ring stays at capacity");
        assert_eq!(g.front().unwrap(), &"line-1".to_string(), "oldest evicted");
        assert_eq!(
            g.back().unwrap(),
            &format!("line-{}", RING_LINES),
            "newest kept"
        );
    }

    #[tokio::test]
    async fn history_last_returns_oldest_first() {
        let a = LogAttach {
            id: "l1".into(),
            cluster: "c".into(),
            node: "n".into(),
            recipe: "r".into(),
            rank: 0,
            history: ring(),
            stop: Arc::new(AtomicBool::new(false)),
            done: Arc::new(AtomicBool::new(false)),
            task: tokio::spawn(async {}),
        };
        for i in 0..5 {
            push_line(&a.history, &format!("x{i}"));
        }
        assert_eq!(a.history_last(3), ["x2", "x3", "x4"]);
        assert_eq!(a.history_last(99), ["x0", "x1", "x2", "x3", "x4"]);
    }

    #[test]
    fn stream_cmd_carries_recipe_rank_and_tail() {
        let cmd = logs_stream_cmd("glm52-nf4-lloyd", 1, 200);
        assert!(cmd.contains("label=sm121.recipe=glm52-nf4-lloyd"));
        assert!(cmd.contains("label=sm121.rank=1"));
        assert!(cmd.contains("--tail=200"));
        assert!(cmd.contains("docker logs -f"));
    }

    #[tokio::test]
    async fn attach_is_idempotent_and_bounded() {
        let app = crate::web::tests::app_state(None);
        let attaches: Mutex<Vec<LogAttach>> = Mutex::new(Vec::new());
        // 8 attaches fill the cap; the 9th is refused.
        for i in 0..MAX_ATTACHES {
            let (id, _) = attach(&app, &attaches, "fake", "n0", &format!("r{i}"), 0).unwrap();
            assert_eq!(id, format!("l{}", i + 1));
        }
        let err = attach(&app, &attaches, "fake", "n0", "overflow", 0).unwrap_err();
        assert!(err.contains("max"), "{err}");
        // Unknown cluster / node are 400-class errors.
        assert!(attach(&app, &attaches, "nope", "n0", "r", 0).is_err());
        assert!(attach(&app, &attaches, "fake", "nope", "r", 0).is_err());
        // by_key finds what exists, misses what doesn't.
        assert!(by_key(&attaches, "fake", "n0", "r0", 0, 10).is_some());
        assert!(by_key(&attaches, "fake", "n0", "r0", 7, 10).is_none());
        // Detach removes + frees a slot; re-attach after detach works.
        assert!(detach(&attaches, "l1"));
        assert!(!detach(&attaches, "l1"));
        attach(&app, &attaches, "fake", "n0", "new", 0).unwrap();
        assert_eq!(attaches.lock().unwrap().len(), MAX_ATTACHES);
    }

    /// A fake attach with a controlled done flag (no real SSH involved).
    fn fake_attach(id: &str, done: bool) -> LogAttach {
        LogAttach {
            id: id.into(),
            cluster: "fake".into(),
            node: "n0".into(),
            recipe: "r".into(),
            rank: 0,
            history: Arc::new(Mutex::new(VecDeque::new())),
            stop: Arc::new(AtomicBool::new(false)),
            done: Arc::new(AtomicBool::new(done)),
            // Long-lived task so the attach stays "live" deterministically.
            task: tokio::spawn(async {
                tokio::time::sleep(std::time::Duration::from_secs(60)).await
            }),
        }
    }

    #[tokio::test]
    async fn attach_reuses_live_stream_for_same_key() {
        let app = crate::web::tests::app_state(None);
        let list: Mutex<Vec<LogAttach>> = Mutex::new(Vec::new());
        list.lock().unwrap().push(fake_attach("l1", false));
        // Same key, live stream -> the existing attach is reused, no new task.
        let (id, history) = attach(&app, &list, "fake", "n0", "r", 0).unwrap();
        assert_eq!(id, "l1");
        assert!(history.is_empty());
        assert_eq!(list.lock().unwrap().len(), 1, "no second attach spawned");
    }

    #[tokio::test]
    async fn attach_replaces_finished_stream_for_same_key() {
        let app = crate::web::tests::app_state(None);
        let list: Mutex<Vec<LogAttach>> = Mutex::new(Vec::new());
        list.lock().unwrap().push(fake_attach("l1", true));
        // Same key, finished stream -> dropped and replaced with a new task.
        attach(&app, &list, "fake", "n0", "r", 0).unwrap();
        let g = list.lock().unwrap();
        assert_eq!(g.len(), 1, "old attach dropped");
        assert_eq!(g[0].id, "l1", "slot id may be reused");
        assert!(
            !g[0].done.load(Ordering::Relaxed),
            "fresh stream not yet done"
        );
        g[0].stop.store(true, Ordering::Relaxed); // end the fake task
    }
}