diff --git a/apps/desktop/DESIGN.md b/apps/desktop/DESIGN.md index 019be7501..d501dc9af 100644 --- a/apps/desktop/DESIGN.md +++ b/apps/desktop/DESIGN.md @@ -118,15 +118,16 @@ Source: `hack logs --json --path --tail 500` (ndjson, follows by default) | State / control | Behaviour | |---|---| | Follow (default) | Newest at bottom; auto-scroll while at the bottom. | -| Pause | Stops auto-scroll; keeps reading into a ring buffer bounded at 10,000 lines **and** 8 MB; "N new lines" chip. Scrolling up pauses. | +| Pause | Stops auto-scroll; keeps reading into a ring buffer bounded at 10,000 lines **and** 8 MB; "N new lines" chip. Scrolling up pauses; Resume returns to the bottom. No pause timeout or automatic resume at the bottom in this preview. | | Search | Filters the buffer in place (plain text, case-insensitive), with match count. | | Service filter | One service at a time: restarts the stream as `hack logs `. `--services` is Loki-only and native rejects it (and `--compose`), so the app never passes either. Multi-select needs R4. | | Very long lines | Cut at 4 KB on screen with "… (+N KB)", click to expand; `entry.truncated` shown as "truncated by hack". | | Secrets warning | Fixed note: "Logs are shown as your services print them. Hack does not hide secrets in logs." No copy-all; buffer never written to disk; "Open in terminal" offered. | | Finished | `end` event on a `--no-follow` read: "End of log" (normal). | | Follow ended | while following, an `end` event (`eof` from Compose, `closed` from Loki) and exit 0: "Log stream ended" + Reconnect. Not an error. | -| Stream error | `error` event or exit ≠ 0: banner with reason and Reconnect. Docker down: "Docker not reachable". | -| Closed by the app | tab change, project change or pause timeout: the app kills the stream; nothing is shown. | +| Stream error | `error` event or exit ≠ 0: banner with reason and Reconnect. CLI failure envelopes and bounded diagnostic tails provide the reason. Docker failures retain the CLI's wording; a normalized "Docker not reachable" label is deferred. | +| Closed by the app | tab change or project change: the app kills the stream; nothing is shown. | +| Busy stream | The reader continuously consumes stdout. A full bounded event queue discards incoming events and reports a visible skipped-event count; it never waits for UI drain. This bounds app retention, not arbitrary producer-side CLI buffering. | | Stopped project | Last 200 lines (`--no-follow`), marked "project stopped". | ### Env tab (names only, never values) diff --git a/apps/desktop/README.md b/apps/desktop/README.md index 57203fb21..7cd78c2d6 100644 --- a/apps/desktop/README.md +++ b/apps/desktop/README.md @@ -60,7 +60,7 @@ each project's detail as JSON. It runs no action, opens no window and writes nothing. An accepted refresh that changes the selected project root or backend clears the -previous log tail, including when Follow is off. +previous log tail, including when Follow is off, and replaces any open Logs stream. Live reads refresh every two seconds with a matching daemon and retry every ten seconds when unavailable. Every request has an order token: a read started before @@ -85,11 +85,34 @@ engine-down, error / cli-missing, many-projects, stopped / starting, running, degraded / stopping, failed-start, needs-attention): [light](screenshots/fixtures-light.png), [dark](screenshots/fixtures-dark.png). -## What v0 does not do yet - -Logs and Activity tabs, CPU and memory, Doctor view, onboarding, always-on, -daemon actions and upgrades come in later units. The design and the CLI/daemon -requests they need are in `DESIGN.md` (PR #800). +## Logs tab + +`hack logs --json` for the selected project, following by default: +- Pause (reading continues, with an "N new" count), search, and a filter for one service (`hack logs `). +- Click any line to show it in full below the list (long or width-clipped lines); lines over 4 KB show how much is hidden. +- "Log stream ended" and failures are shown distinctly, with Reconnect. +- CLI stderr is drained concurrently with an 8 KB retained tail. A separate 8 KB tail retains non-event stdout, including pretty JSON failure envelopes. Failed exits show the CLI error message or diagnostic tail and a human-readable exit status. +- The buffer holds at most 10,000 lines and 8 MB (service and timestamp included). Messages in valid records are capped at 64 KB; raw records over 1 MiB are replaced by a visible dropped-record marker instead. Memory only: nothing is written to disk and there is no copy-all. +- Scrolling up pauses following; Resume returns to the newest line. Each event's text fields are capped before queueing (log message 64 KB, notice 4 KB, other fields 256 bytes). The 1,024-event queue retains at most about 70 MiB of text. When full, the reader discards incoming events and keeps consuming stdout; the tab shows a cumulative skipped-event count. The CLI does not honor pipe backpressure, so blocking the reader would allow its own output buffer to grow. This bounds app retention, not the CLI's memory under an arbitrarily fast producer. +- Stderr output uses a muted style; only explicit error-level entries use the danger color. Services-tab tail polling stops while Logs is open. +- The app's limits do not bound allocations before the CLI emits a record: the current Compose reader accumulates a complete line before JSON formatting. A single oversized or unterminated service line can therefore grow CLI memory. Bounding that upstream chunk reader is a separate CLI fix; this preview does not claim end-to-end memory safety against arbitrary service output. +- "Copy command" copies the equivalent `hack logs --path … [service]` for a terminal. +- Native projects show the terminal command instead, because native logs are a one-service snapshot. + +Fixtures: `--logs-fixture ` (`--list-fixtures` lists them). Boards: +[light](screenshots/logs-light.png), [dark](screenshots/logs-dark.png). Row by row: +following / paused with search, long lines / service filter, follow ended / +failed, native / stopped. +The paused-search fixture shows its active query in the search field. Finite +reads for observed stopped projects include that state alongside the stream status. +The paused fixture also demonstrates the overflow count. Pause has no timeout; +following resumes through Resume, not automatically when scrolling to the bottom. + +## Not yet + +The Activity tab, CPU and memory, onboarding, always-on, daemon actions and +upgrades come in later units. The design and the CLI/daemon requests they +need are in `DESIGN.md`. ## Env tab (names only) diff --git a/apps/desktop/app/src/logs_view.rs b/apps/desktop/app/src/logs_view.rs new file mode 100644 index 000000000..3ca5adb66 --- /dev/null +++ b/apps/desktop/app/src/logs_view.rs @@ -0,0 +1,517 @@ +//! The Logs tab (DESIGN.md, "Logs tab"). +//! +//! One `hack logs --json` stream per open tab. The stream is opened on the +//! background executor and drained on a short timer; dropping the feed kills +//! the CLI's process group. Changing project, tab or service filter closes +//! the stream (no banner) and opens a new one. Nothing is written to disk and +//! there is no copy-all. + +use std::time::Duration; + +use gpui_kit::component::{ + ActiveTheme, Sizable as _, + button::{Button, ButtonVariants as _}, + h_flex, + input::{Input, InputEvent, InputState}, + v_flex, +}; +use gpui_kit::{ + AnyElement, Context, Entity, FontWeight, IntoElement, ParentElement, ScrollStrategy, + SharedString, Styled, UniformListScrollHandle, Window, div, prelude::*, px, uniform_list, +}; +use hack_desktop_core::logs::{LogsModel, StreamState}; +use hack_desktop_core::source::LogFeed; + +use crate::view::HackApp; + +const DRAIN_EVERY: Duration = Duration::from_millis(150); +/// Events applied per tick. The reader discards overflow without blocking +/// stdout, because the CLI does not await pipe backpressure. +const BATCH: usize = 500; + +/// Logs tab state owned by the window. +pub struct LogsPane { + pub model: LogsModel, + feed: Option>, + search: Entity, + scroll: UniformListScrollHandle, + /// Bumped on every (re)open so a late open result is discarded. + generation: u64, + /// Visible buffer indices, recomputed on change. + visible: Vec, +} + +impl LogsPane { + pub fn new(window: &mut Window, cx: &mut Context) -> Self { + let search = cx.new(|cx| InputState::new(window, cx).placeholder("Search logs")); + cx.subscribe(&search, |this: &mut HackApp, state, ev: &InputEvent, cx| { + if matches!(ev, InputEvent::Change) { + this.logs.model.search = state.read(cx).value().to_string(); + this.logs.refresh_visible(); + cx.notify(); + } + }) + .detach(); + Self { + model: LogsModel::default(), + feed: None, + search, + scroll: UniformListScrollHandle::new(), + generation: 0, + visible: Vec::new(), + } + } + + /// Show a fixture state (screenshots and `--logs-fixture`). + pub fn set_model(&mut self, model: LogsModel, window: &mut Window, cx: &mut Context) { + self.feed = None; + self.search.update(cx, |input, cx| { + input.set_value(model.search.clone(), window, cx); + }); + self.model = model; + self.refresh_visible(); + if self.should_autoscroll() { + self.scroll + .scroll_to_item(self.visible.len() - 1, ScrollStrategy::Bottom); + } + } + + /// Auto-scroll only while following a live stream and not paused. + fn should_autoscroll(&self) -> bool { + self.model.follow + && self.model.state == StreamState::Following + && !self.model.paused + && !self.visible.is_empty() + } + + fn refresh_visible(&mut self) { + self.visible = self.model.visible_indices(); + } + + /// The app closes the stream: no banner. + pub fn close(&mut self) { + self.generation += 1; + self.feed = None; + self.model.close(); + } +} + +impl HackApp { + /// The equivalent terminal command for the current project and filter, + /// shell-quoted. Following, so it keeps streaming in the terminal. + fn logs_command(&self) -> Option { + let (root, native) = self.selected_root()?; + if native { + return None; + } + let quote = |s: &str| format!("'{}'", s.replace('\'', "'\\''")); + let mut cmd = format!("hack logs --path {}", quote(&root.to_string_lossy())); + if let Some(svc) = &self.logs.model.service { + cmd.push(' '); + cmd.push_str("e(svc)); + } + Some(cmd) + } + + /// (Re)open the stream for the selected project and current filter. + pub(crate) fn open_logs(&mut self, cx: &mut Context) { + self.logs.close(); + let Some((root, native)) = self.selected_root() else { + // No folder to read from: never leave another project's lines. + self.logs + .model + .unavailable("This project has no folder to read logs from."); + self.logs.refresh_visible(); + cx.notify(); + return; + }; + if native { + self.logs + .model + .unavailable(hack_desktop_core::source::NATIVE_LOGS_NOTE); + self.logs.refresh_visible(); + cx.notify(); + return; + } + let running = self.model.detail().is_some_and(|d| d.state.is_active()); + let args = self.logs.model.cli_args(running); + self.logs.model.start(running); + self.logs.refresh_visible(); + let generation = self.logs.generation; + let source = self.source.clone(); + cx.spawn(async move |this, cx| { + let opened = cx + .background_executor() + .spawn(async move { source.open_logs(&root, false, &args) }) + .await; + let _ = this.update(cx, |this, cx| { + if this.logs.generation != generation { + return; // closed or reopened meanwhile; drop the feed + } + match opened { + Ok(feed) => this.logs.feed = Some(feed), + Err(reason) => { + this.logs.model.state = StreamState::Failed { reason }; + } + } + cx.notify(); + }); + }) + .detach(); + } + + /// Drain the open stream on a timer while the window exists. + pub(crate) fn start_log_drain(&mut self, cx: &mut Context) { + cx.spawn(async move |this, cx| { + loop { + cx.background_executor().timer(DRAIN_EVERY).await; + if this.update(cx, |this, cx| this.drain_logs(cx)).is_err() { + return; + } + } + }) + .detach(); + } + + fn drain_logs(&mut self, cx: &mut Context) { + let Some(feed) = self.logs.feed.as_mut() else { + return; + }; + let hack_desktop_core::runner::StreamPoll { + events, + exit, + skipped, + } = feed.poll(BATCH); + self.logs.model.skipped = self.logs.model.skipped.saturating_add(skipped); + if events.is_empty() && exit.is_none() && skipped == 0 { + return; + } + // While paused, remember which line is at the top so evictions at + // the front do not move the text being read. + let anchor = if self.logs.model.paused { + // Same logic as gpui's test-only `logical_scroll_top_index`, + // through the public production API. + let top = { + let state = self.logs.scroll.0.borrow(); + state.deferred_scroll_to_item.as_ref().map_or_else( + || state.base_handle.logical_scroll_top().0, + |d| d.item_index, + ) + }; + self.logs + .visible + .get(top) + .and_then(|&i| self.logs.model.line(i)) + .map(|l| l.seq) + } else { + None + }; + for e in &events { + self.logs.model.apply(e); + } + if let Some(code) = exit { + self.logs.feed = None; + self.logs.model.exited(code); + } + self.logs.refresh_visible(); + if self.logs.should_autoscroll() { + self.logs + .scroll + .scroll_to_item(self.logs.visible.len() - 1, ScrollStrategy::Bottom); + } else if let Some(seq) = anchor { + let pos = self.logs.model.position_of_seq(&self.logs.visible, seq); + self.logs.scroll.scroll_to_item(pos, ScrollStrategy::Top); + } + cx.notify(); + } + + pub(crate) fn render_logs(&self, services: &[String], cx: &mut Context) -> AnyElement { + let t = cx.theme().clone(); + let m = &self.logs.model; + + let pause_label: SharedString = if m.paused { + if m.unseen > 0 { + format!("Resume · {} new", m.unseen).into() + } else { + "Resume".into() + } + } else { + "Pause".into() + }; + let unavailable = matches!(m.state, StreamState::Unavailable { .. }); + // Pause only makes sense while following; a finite read has ended. + let mut toolbar = h_flex() + .gap_2() + .px_5() + .py_2() + .when(m.follow, |d| { + d.child( + Button::new("logs-pause") + .label(pause_label) + .small() + .outline() + .on_click(cx.listener(|this, _, _, cx| { + let paused = !this.logs.model.paused; + this.logs.model.set_paused(paused); + if !paused && !this.logs.visible.is_empty() { + this.logs.scroll.scroll_to_item( + this.logs.visible.len() - 1, + ScrollStrategy::Bottom, + ); + } + cx.notify(); + })), + ) + }) + .child( + div() + .w(px(220.)) + .child(Input::new(&self.logs.search).small()), + ); + // Service filter: one at a time (positional `hack logs `). + let chip = |id: SharedString, label: String, active: bool| { + let b = Button::new(id).label(label).small(); + if active { b.primary() } else { b.ghost() } + }; + toolbar = toolbar.child( + chip("svc-all".into(), "All".into(), m.service.is_none()).on_click(cx.listener( + |this, _, _, cx| { + if this.logs.model.set_service(None) { + this.open_logs(cx); + } + }, + )), + ); + for s in services { + let name = s.clone(); + toolbar = toolbar.child( + chip( + format!("svc-{s}").into(), + s.clone(), + m.service.as_deref() == Some(s.as_str()), + ) + .on_click(cx.listener(move |this, _, _, cx| { + if this.logs.model.set_service(Some(name.clone())) { + this.open_logs(cx); + } + })), + ); + } + toolbar = toolbar.child(div().flex_1()); + if !m.search.trim().is_empty() { + toolbar = toolbar.child( + div() + .text_xs() + .text_color(t.muted_foreground) + .child(format!("{} matches", self.logs.visible.len())), + ); + } + if let Some(cmd) = self.logs_command() { + toolbar = toolbar.child( + Button::new("logs-copy-cmd") + .label("Copy command") + .small() + .ghost() + .tooltip("Copy the hack logs command to run in a terminal") + .on_click(move |_, _, cx| { + cx.write_to_clipboard(gpui_kit::ClipboardItem::new_string(cmd.clone())); + }), + ); + } + if m.can_reconnect() { + toolbar = toolbar.child( + Button::new("logs-reconnect") + .label("Reconnect") + .small() + .primary() + .on_click(cx.listener(|this, _, _, cx| this.open_logs(cx))), + ); + } + + let banner = m.banner().map(|text| { + let stopped = self + .model + .detail() + .is_some_and(|d| d.state == hack_desktop_core::state::ProjectState::Stopped); + let text = if !m.follow && stopped && !unavailable { + format!("{text} · project stopped") + } else { + text + }; + let failed = matches!(m.state, StreamState::Failed { .. }); + div() + .mx_5() + .mb_2() + .px_3() + .py_1() + .rounded_md() + .text_xs() + .bg(if failed { + t.danger.opacity(0.12) + } else { + t.muted + }) + .text_color(if failed { t.danger } else { t.muted_foreground }) + .child(text) + }); + + let list = uniform_list( + "log-lines", + self.logs.visible.len(), + cx.processor(|this, range: std::ops::Range, _window, cx| { + let t = cx.theme().clone(); + range + .filter_map(|ix| { + let buf_ix = *this.logs.visible.get(ix)?; + let line = this.logs.model.line(buf_ix)?; + let (text, shown_hidden) = line.display(false); + let hidden = shown_hidden + line.cut; + let seq = line.seq; + let color = if line.error_level { + t.danger + } else if line.stderr { + t.muted_foreground + } else { + t.foreground + }; + let mut row = h_flex() + .id(("log-line", seq as usize)) + .w_full() + .overflow_hidden() + .h(px(20.)) + .px_3() + .gap_2() + .whitespace_nowrap() + .child( + div() + .w(px(72.)) + .flex_none() + .text_color(t.muted_foreground) + .child(line.service.clone().unwrap_or_default()), + ) + .child( + div() + .flex_1() + .min_w_0() + .truncate() + .text_color(color) + .child(text.to_owned()), + ); + if hidden > 0 { + row = row.child( + div() + .flex_none() + .text_color(t.muted_foreground) + .child(format!("… (+{} KB)", hidden.div_ceil(1024))), + ); + } + // Any row expands on click: a line can be clipped by + // the row width long before the 4 KB cut applies. + let expanded = this.logs.model.is_expanded(seq); + row = row + .cursor_pointer() + .when(expanded, |r| r.bg(t.muted)) + .on_click(cx.listener(move |this, _, _, cx| { + this.logs.model.toggle_expanded(seq); + cx.notify(); + })); + if line.truncated_by_hack { + row = row.child( + div() + .flex_none() + .text_color(t.muted_foreground) + .child("truncated by hack"), + ); + } + Some(row) + }) + .collect() + }), + ) + .track_scroll(&self.logs.scroll) + .flex_1() + .font_family(t.mono_font_family.clone()) + .text_xs(); + + let empty = self.logs.visible.is_empty() + && !matches!( + m.state, + StreamState::Unavailable { .. } | StreamState::Failed { .. } + ); + let body = v_flex() + .id("log-body") + // Scrolling up to read older lines pauses following. + .on_scroll(cx.listener(|this, ev: &gpui_kit::ScrollEvent, _, cx| { + let up = match ev.delta { + gpui_kit::ScrollDelta::Pixels(p) => p.y > px(0.), + gpui_kit::ScrollDelta::Lines(l) => l.y > 0., + }; + if up && this.logs.model.follow && !this.logs.model.paused { + this.logs.model.set_paused(true); + cx.notify(); + } + })) + .mx_5() + .flex_1() + .min_h(px(160.)) + .rounded_md() + .border_1() + .border_color(t.border) + .bg(t.muted.opacity(0.4)) + .py_1() + .child(list) + .when(empty, |d| { + d.child(div().px_3().text_xs().text_color(t.muted_foreground).child( + if m.search.trim().is_empty() { + "No log lines yet" + } else { + "No lines match the search" + }, + )) + }); + + let expanded = m.expanded_lines(); + let detail = (!expanded.is_empty()).then(|| { + let mut col = v_flex() + .id("log-expanded") + .mx_5() + .mt_2() + .max_h(px(160.)) + .overflow_y_scroll() + .rounded_md() + .border_1() + .border_color(t.border) + .p_2() + .gap_1() + .text_xs() + .font_family(t.mono_font_family.clone()); + for l in expanded { + col = col.child(l.message.clone()); + } + col + }); + + v_flex() + .flex_1() + .min_h_0() + .when(!unavailable, |d| d.child(toolbar)) + .when(m.skipped > 0, |d| d.child(div().mx_5().mb_2().text_xs() + .text_color(t.muted_foreground) + .child(format!("{} events skipped while the log stream outran the display.", m.skipped)))) + .children(banner) + .child(body) + .children(detail) + .child( + div() + .px_5() + .py_2() + .text_xs() + .text_color(t.muted_foreground) + .font_weight(FontWeight::NORMAL) + .child( + "Logs are shown as your services print them. Hack does not hide secrets in logs. Nothing here is saved to disk.", + ), + ) + .into_any_element() + } +} diff --git a/apps/desktop/app/src/main.rs b/apps/desktop/app/src/main.rs index 0536023fd..84553a28b 100644 --- a/apps/desktop/app/src/main.rs +++ b/apps/desktop/app/src/main.rs @@ -9,6 +9,7 @@ //! fixture, light and dark, offscreen //! ``` +mod logs_view; #[cfg(feature = "screenshots")] mod screenshots; mod view; @@ -22,13 +23,14 @@ use hack_desktop_core::source::{FixtureSource, LiveSource, Source}; pub(crate) struct Args { pub fixture: Option, + pub logs_fixture: Option, pub theme: Option, pub screenshots: Option, } fn usage() -> ! { eprintln!( - "usage: hack-desktop [--fixtures ] [--theme light|dark] [--list-fixtures] [--screenshots ]" + "usage: hack-desktop [--fixtures | --logs-fixture ] [--theme light|dark] [--list-fixtures] [--screenshots ]" ); std::process::exit(2) } @@ -36,6 +38,7 @@ fn usage() -> ! { fn parse_args() -> Args { let mut a = Args { fixture: None, + logs_fixture: None, theme: None, screenshots: None, }; @@ -49,10 +52,24 @@ fn parse_args() -> Args { std::process::exit(2) })); } + "--logs-fixture" => { + let name = it.next().unwrap_or_else(|| usage()); + a.logs_fixture = Some( + hack_desktop_core::logs::fixtures::LogsFixture::parse(&name).unwrap_or_else( + || { + eprintln!("unknown logs fixture {name:?}; see --list-fixtures"); + std::process::exit(2) + }, + ), + ); + } "--list-fixtures" => { for f in Fixture::ALL { println!("{}", f.name()); } + for f in hack_desktop_core::logs::fixtures::LogsFixture::ALL { + println!("--logs-fixture {}", f.name()); + } std::process::exit(0) } "--theme" => { @@ -102,7 +119,14 @@ fn main() { } gpui_kit::application().run(move |cx| { gpui_kit::init(cx); - let fixture = args.fixture; + let fixture = args.fixture.or(args.logs_fixture.map(|l| { + if l == hack_desktop_core::logs::fixtures::LogsFixture::Stopped { + Fixture::Stopped + } else { + Fixture::Running + } + })); + let logs_fixture = args.logs_fixture; let theme = args.theme; let source = source_for(fixture); gpui_kit::open_window(window_options(), cx, move |window, cx| { @@ -111,7 +135,7 @@ fn main() { None => Theme::sync_system_appearance(Some(window), cx), } cx.new(|cx| { - let mut app = view::HackApp::new(source, fixture.is_none(), cx); + let mut app = view::HackApp::new(source, fixture.is_none(), window, cx); if let Some(f) = fixture { app.select(fixtures::selected(f)); if f.name().starts_with("env-") { @@ -124,6 +148,9 @@ fn main() { if let Some(a) = fixture.and_then(|f| fixtures::local(f).in_flight) { app.simulate_in_flight(a); } + if let Some(l) = logs_fixture { + app.show_logs_fixture(l, window, cx); + } app }) }) diff --git a/apps/desktop/app/src/screenshots.rs b/apps/desktop/app/src/screenshots.rs index e9ea9680c..88f81675a 100644 --- a/apps/desktop/app/src/screenshots.rs +++ b/apps/desktop/app/src/screenshots.rs @@ -11,6 +11,7 @@ use gpui_kit::component::{Theme, ThemeMode}; use gpui_kit::test::TestWindowExt as _; use gpui_kit::{AppContext as _, HeadlessAppContext, WindowOptions}; use hack_desktop_core::fixtures::{self, Fixture}; +use hack_desktop_core::logs::fixtures::LogsFixture; use crate::view::HackApp; @@ -22,10 +23,21 @@ pub fn run(dir: &Path, fixture: Option) { gpui_kit::platform::current_headless_renderer, ); cx.update(gpui_kit::init); + // Every app fixture, then every Logs tab fixture on its base project state. + let mut shots: Vec<(&'static str, Fixture, Option)> = + Fixture::ALL.iter().map(|f| (f.name(), *f, None)).collect(); + for l in LogsFixture::ALL { + let base = if l == LogsFixture::Stopped { + Fixture::Stopped + } else { + Fixture::Running + }; + shots.push((l.name(), base, Some(l))); + } for mode in [ThemeMode::Light, ThemeMode::Dark] { - for f in Fixture::ALL - .into_iter() - .filter(|f| fixture.is_none_or(|selected| selected == *f)) + for &(name, f, logs) in shots + .iter() + .filter(|(_, f, logs)| fixture.is_none_or(|selected| selected == *f && logs.is_none())) { cx.update(|cx| Theme::change(mode, None, cx)); let opts = WindowOptions { @@ -35,9 +47,10 @@ pub fn run(dir: &Path, fixture: Option) { }; let (handle, _) = cx .update(|cx| { - gpui_kit::open_window(opts, cx, |_, cx| { + gpui_kit::open_window(opts, cx, |window, cx| { cx.new(|cx| { - let mut app = HackApp::new(crate::source_for(Some(f)), false, cx); + let mut app = + HackApp::new(crate::source_for(Some(f)), false, window, cx); app.select(fixtures::selected(f)); if f.name().starts_with("env-") { app.show_env_fixture(); @@ -48,6 +61,9 @@ pub fn run(dir: &Path, fixture: Option) { if let Some(a) = fixtures::local(f).in_flight { app.simulate_in_flight(a); } + if let Some(l) = logs { + app.show_logs_fixture(l, window, cx); + } app }) }) @@ -63,7 +79,7 @@ pub fn run(dir: &Path, fixture: Option) { } else { "light" }; - let path = dir.join(format!("{}-{suffix}.png", f.name())); + let path = dir.join(format!("{name}-{suffix}.png")); img.save(&path).expect("write png"); println!("{}", path.display()); cx.update_window(handle, |_, window, _| window.remove_window()) diff --git a/apps/desktop/app/src/view.rs b/apps/desktop/app/src/view.rs index 77e6707cb..72032cc3a 100644 --- a/apps/desktop/app/src/view.rs +++ b/apps/desktop/app/src/view.rs @@ -24,13 +24,16 @@ use gpui_kit::{ use hack_desktop_core::models::LogEntry; use hack_desktop_core::source::{ActionOutcome, DaemonRecovery, Source}; use hack_desktop_core::state::{Action, ProjectState, ServiceView, Tone}; -use hack_desktop_core::viewmodel::{AppModel, DetailView, Event, ListView, Phase, Row}; +use hack_desktop_core::viewmodel::{ + AppModel, DetailView, Event, ListView, Phase, RefreshToken, Row, +}; const TAIL_LINES: u32 = 20; pub struct HackApp { - model: AppModel, - source: Arc, + pub(crate) model: AppModel, + pub(crate) source: Arc, + pub(crate) logs: crate::logs_view::LogsPane, tab: usize, env: hack_desktop_core::env::EnvModel, tail: Result, String>, @@ -45,10 +48,17 @@ pub struct HackApp { impl HackApp { /// `live`: poll the source. Fixtures load once and never poll. - pub fn new(source: Arc, live: bool, cx: &mut Context) -> Self { + pub fn new( + source: Arc, + live: bool, + window: &mut Window, + cx: &mut Context, + ) -> Self { + let logs = crate::logs_view::LogsPane::new(window, cx); let mut this = Self { model: AppModel::new(), source, + logs, tab: 0, env: Default::default(), tail: Ok(Vec::new()), @@ -63,9 +73,25 @@ impl HackApp { } else { this.refresh_now(); } + this.start_log_drain(cx); this } + /// Fixtures: show the Logs tab in a designed state. + pub fn show_logs_fixture( + &mut self, + f: hack_desktop_core::logs::fixtures::LogsFixture, + window: &mut Window, + cx: &mut Context, + ) { + use hack_desktop_core::logs::fixtures::{LogsFixture, model}; + if f == LogsFixture::Native { + self.select("docs-site"); + } + self.tab = 1; + self.logs.set_model(model(f), window, cx); + } + /// Synchronous refresh, for fixtures and screenshots. pub fn refresh_now(&mut self) { let snap = self.source.snapshot(); @@ -76,6 +102,45 @@ impl HackApp { ); } + /// Apply a fresh snapshot. If it changed the selected project (the old + /// one disappeared), the Logs tab's stream follows the new selection. + fn apply_refresh( + &mut self, + token: RefreshToken, + snap: Result, + cx: &mut Context, + ) -> bool { + let before = self.model.selected_row().map(|r| r.name.clone()); + let previous_root = self.selected_root(); + if !self.model.finish_refresh(token, snap) { + return false; + } + let after = self.model.selected_row().map(|r| r.name.clone()); + let selection_changed = before != after || previous_root != self.selected_root(); + if selection_changed { + // Only accepted snapshots may invalidate a tail or replace a + // stream. Root/backend changes also change stream ownership. + self.tail = Ok(Vec::new()); + self.logs.model.set_service(None); + self.env.begin(); + if self.tab == 2 { + self.read_env(cx); + } + } + if self.tab == 1 && selection_changed { + self.open_logs(cx); + } else if self.tab == 1 { + // A stopped project's finite read becomes a followed stream once + // the project runs (started here or elsewhere). + let running = self.model.detail().is_some_and(|d| d.state.is_active()); + if self.logs.model.should_follow_now(running) { + self.open_logs(cx); + } + } + cx.notify(); + true + } + /// Fixtures: select a project by name. pub fn select(&mut self, name: &str) { self.model.apply(Event::Select(name.to_owned())); @@ -145,7 +210,7 @@ impl HackApp { self.tail = Err("Project status could not be read".into()); } - fn selected_root(&self) -> Option<(PathBuf, bool)> { + pub(crate) fn selected_root(&self) -> Option<(PathBuf, bool)> { let r = self.model.selected_row()?; Some((PathBuf::from(r.repo_root.as_ref()?), r.is_native())) } @@ -167,31 +232,24 @@ impl HackApp { .await; let root = this .update(cx, |this, cx| { - let previous = this.selected_root(); - if !this.model.finish_refresh(token, snap) { + if !this.apply_refresh(token, snap, cx) { return None; } - if previous != this.selected_root() { - this.tail = Ok(Vec::new()); - this.env.begin(); - if this.tab == 2 { - this.read_env(cx); - } - } - cx.notify(); - this.follow.then(|| this.selected_root()).flatten() + (this.follow && this.tab == 0) + .then(|| this.selected_root()) + .flatten() }) .ok() .flatten(); if let Some((root, native)) = root { - let requested = root.clone(); + let requested = (root.clone(), native); let tail = cx .background_executor() .spawn(async move { source.log_tail(&root, native, TAIL_LINES) }) .await; let _ = this.update(cx, |this, cx| { - // Drop a tail fetched for a project no longer selected. - if this.selected_root().is_some_and(|(r, _)| r == requested) { + // Root and backend both identify the requested tail. + if this.selected_root().as_ref() == Some(&requested) { this.tail = tail; cx.notify(); } @@ -263,19 +321,7 @@ impl HackApp { .background_executor() .spawn(async move { source.snapshot() }) .await; - let _ = this.update(cx, |this, cx| { - let previous = this.selected_root(); - if this.model.finish_refresh(token, snap) { - if previous != this.selected_root() { - this.tail = Ok(Vec::new()); - this.env.begin(); - if this.tab == 2 { - this.read_env(cx); - } - } - cx.notify(); - } - }); + let _ = this.update(cx, |this, cx| this.apply_refresh(token, snap, cx)); }) .detach(); } @@ -466,6 +512,12 @@ impl HackApp { if !this.live { this.refresh_now(); } + // A service filter belongs to one project: reset it on every + // project change, whichever tab is open. + this.logs.model.set_service(None); + if this.tab == 1 { + this.open_logs(cx); + } cx.notify(); })) .into_any_element() @@ -590,13 +642,19 @@ impl HackApp { .small() .selected_index(self.tab) .child(Tab::new().label("Services")) - .child(Tab::new().label("Logs").disabled(true)) + .child(Tab::new().label("Logs")) .child(Tab::new().label("Env")) .child(Tab::new().label("Activity").disabled(true)) .on_click(cx.listener(|this, ix: &usize, _, cx| { + let (from, to) = (this.tab, *ix); + this.tab = to; this.env.begin(); - this.tab = *ix; - if *ix == 2 { + if to == 1 && from != 1 { + this.open_logs(cx); + } else if from == 1 && to != 1 { + this.logs.close(); + } + if to == 2 { this.read_env(cx); } cx.notify(); @@ -609,21 +667,29 @@ impl HackApp { .min_h_0() .child(header) .child(div().px_5().pt_2().child(tabs)) - .child( - v_flex() - .id("detail-body") - .flex_1() - .min_h_0() - .overflow_y_scroll() - .child(if self.tab == 2 { - self.env_tab(cx).into_any_element() - } else { + .map(|col| { + if self.tab == 1 { + // The Logs tab scrolls inside its own virtualized list. + let names: Vec = d.services.iter().map(|s| s.name.clone()).collect(); + col.child(self.render_logs(&names, cx)) + } else { + col.child( v_flex() - .child(services_table(&d.services, cx)) - .child(self.log_tail(cx)) - .into_any_element() - }), - ) + .id("detail-body") + .flex_1() + .min_h_0() + .overflow_y_scroll() + .child(if self.tab == 2 { + self.env_tab(cx).into_any_element() + } else { + v_flex() + .child(services_table(&d.services, cx)) + .child(self.log_tail(cx)) + .into_any_element() + }), + ) + } + }) } fn env_tab(&self, cx: &mut Context) -> impl IntoElement { diff --git a/apps/desktop/core/src/lib.rs b/apps/desktop/core/src/lib.rs index a37469092..b365509ee 100644 --- a/apps/desktop/core/src/lib.rs +++ b/apps/desktop/core/src/lib.rs @@ -4,7 +4,7 @@ //! - [`runner`]: runs the CLI as the user with timeouts. //! - [`daemon`]: read-only `hackd` socket client. //! - [`state`]: project display state and actions. -//! - [`viewmodel`]: list and detail state machines. +//! - [`viewmodel`]: list and detail state machines; [`logs`]: the Logs tab. //! - [`source`]: live and fixture data sources; [`fixtures`]: canned states. //! //! The app keeps no state of its own beyond window layout and holds no @@ -12,6 +12,7 @@ pub mod daemon; pub mod fixtures; +pub mod logs; pub mod models; mod process; pub mod runner; diff --git a/apps/desktop/core/src/logs.rs b/apps/desktop/core/src/logs.rs new file mode 100644 index 000000000..b24000478 --- /dev/null +++ b/apps/desktop/core/src/logs.rs @@ -0,0 +1,899 @@ +//! Logs tab model (DESIGN.md, "Logs tab"). Pure: no I/O, no clock. +//! +//! Contract: +//! - The buffer is a ring bounded by lines **and** bytes; the oldest lines go +//! first and are counted in `dropped`. +//! - Pause never stops reading: lines keep arriving and are counted as unseen. +//! - Search filters what is shown, never what is kept. +//! - One service filter at a time, passed to the CLI as the positional +//! `hack logs ` (`--services` is Loki-only and native rejects it). +//! - An `end` event while following is "Follow ended" (normal), not an error; +//! an `error` event, an unsuccessful end reason or a non-zero exit is a +//! failure. A stream the app closes itself leaves no banner. +//! - Nothing here is written to disk. + +use std::collections::VecDeque; + +use crate::models::LogEvent; + +pub const MAX_LINES: usize = 10_000; +pub const MAX_BYTES: usize = 8 * 1024 * 1024; +/// Lines longer than this are cut on screen until expanded. +pub const DISPLAY_CUT: usize = 4 * 1024; +/// Longest message kept per line; the rest is dropped on intake and +/// counted in `LogLine::cut`, so one line can never exceed the byte bound. +pub const MAX_LINE_BYTES: usize = 64 * 1024; +pub const FOLLOW_TAIL: u32 = 500; +pub const STOPPED_TAIL: u32 = 200; + +#[derive(Debug, Clone, PartialEq, Eq)] +pub struct LogLine { + pub seq: u64, + pub service: Option, + pub message: String, + pub stderr: bool, + pub error_level: bool, + pub timestamp: Option, + /// The CLI cut this line (`entry.truncated`). + pub truncated_by_hack: bool, + /// Bytes the app dropped from this line on intake (over `MAX_LINE_BYTES`). + pub cut: usize, +} + +impl LogLine { + /// Bytes this line holds in the buffer (message, service, timestamp). + #[must_use] + pub fn retained_bytes(&self) -> usize { + self.message.len() + + self.service.as_ref().map_or(0, String::len) + + self.timestamp.as_ref().map_or(0, String::len) + } +} + +impl LogLine { + /// The text to draw, and how many bytes were hidden, when not expanded. + #[must_use] + pub fn display(&self, expanded: bool) -> (&str, usize) { + if expanded || self.message.len() <= DISPLAY_CUT { + return (&self.message, 0); + } + let mut cut = DISPLAY_CUT; + while !self.message.is_char_boundary(cut) { + cut -= 1; + } + (&self.message[..cut], self.message.len() - cut) + } +} + +#[derive(Debug, Clone, PartialEq, Eq)] +pub enum StreamState { + /// No stream (tab not open, or the app closed it). + Idle, + Connecting, + Following, + /// A `--no-follow` read reached its end. + Finished, + /// A followed stream ended normally (`end` eof/closed, exit 0). + FollowEnded, + Failed { + reason: String, + }, + /// Not available for this project in this version. + Unavailable { + reason: String, + }, +} + +#[derive(Debug, Clone, PartialEq, Eq)] +pub struct LogsModel { + lines: VecDeque, + bytes: usize, + next_seq: u64, + pub dropped: u64, + /// Events the pipe reader discarded when the UI queue was full. + pub skipped: u64, + pub state: StreamState, + /// Following (`true`) or a finite `--no-follow` read. + pub follow: bool, + pub paused: bool, + pub unseen: usize, + pub search: String, + pub service: Option, + /// Lines the user expanded past `DISPLAY_CUT`. + pub expanded: Vec, + max_lines: usize, + max_bytes: usize, +} + +impl Default for LogsModel { + fn default() -> Self { + Self::with_limits(MAX_LINES, MAX_BYTES) + } +} + +impl LogsModel { + #[must_use] + pub fn with_limits(max_lines: usize, max_bytes: usize) -> Self { + Self { + lines: VecDeque::new(), + bytes: 0, + next_seq: 0, + dropped: 0, + skipped: 0, + state: StreamState::Idle, + follow: true, + paused: false, + unseen: 0, + search: String::new(), + service: None, + expanded: Vec::new(), + max_lines, + max_bytes, + } + } + + /// CLI arguments (after `hack`, before `--path`) for the current filter. + /// A stopped project gets a finite read of the last lines. + #[must_use] + pub fn cli_args(&self, project_running: bool) -> Vec { + let mut a = vec!["logs".to_owned(), "--json".to_owned()]; + if project_running { + a.extend(["--tail".to_owned(), FOLLOW_TAIL.to_string()]); + } else { + a.extend([ + "--no-follow".to_owned(), + "--tail".to_owned(), + STOPPED_TAIL.to_string(), + ]); + } + if let Some(s) = &self.service { + a.push(s.clone()); + } + a + } + + /// A new stream starts: clear the buffer and wait for events. + pub fn start(&mut self, follow: bool) { + self.lines.clear(); + self.bytes = 0; + self.dropped = 0; + self.skipped = 0; + self.unseen = 0; + self.expanded.clear(); + self.follow = follow; + self.state = StreamState::Connecting; + } + + /// The app closed the stream (tab or project change): no banner. + pub fn close(&mut self) { + self.state = StreamState::Idle; + } + + pub fn unavailable(&mut self, reason: &str) { + self.lines.clear(); + self.bytes = 0; + self.skipped = 0; + self.state = StreamState::Unavailable { + reason: reason.to_owned(), + }; + } + + /// Choose a service (or all). Returns `true` when the stream must restart. + pub fn set_service(&mut self, service: Option) -> bool { + if self.service == service { + return false; + } + self.service = service; + true + } + + pub fn set_paused(&mut self, paused: bool) { + self.paused = paused; + if !paused { + self.unseen = 0; + } + } + + pub fn toggle_expanded(&mut self, seq: u64) { + if let Some(i) = self.expanded.iter().position(|s| *s == seq) { + self.expanded.remove(i); + } else { + self.expanded.push(seq); + } + } + + #[must_use] + pub fn is_expanded(&self, seq: u64) -> bool { + self.expanded.contains(&seq) + } + + /// Apply one ndjson event from `hack logs --json`. + pub fn apply(&mut self, e: &LogEvent) { + match e.kind.as_str() { + "start" => { + if self.state == StreamState::Connecting { + self.state = if self.follow { + StreamState::Following + } else { + StreamState::Connecting + }; + } + } + "log" => { + if let Some(entry) = &e.entry { + if self.state == StreamState::Connecting && self.follow { + self.state = StreamState::Following; + } + self.push(LogLine { + seq: 0, + service: entry.service.clone(), + message: entry.message.clone(), + stderr: entry.stream.as_deref() == Some("stderr"), + error_level: entry.level.as_deref() == Some("error"), + timestamp: entry.timestamp.clone(), + truncated_by_hack: entry.truncated == Some(true), + cut: 0, + }); + } + } + "error" => { + self.state = StreamState::Failed { + reason: e + .message + .clone() + .unwrap_or_else(|| "log stream failed".into()), + }; + } + "end" => { + let reason = e.reason.as_deref().unwrap_or(""); + let failed = + reason == "error" || (reason.starts_with("exit:") && reason != "exit:0"); + if failed { + // Keep a specific reason from an earlier `error` event. + if !matches!(self.state, StreamState::Failed { .. }) { + self.state = StreamState::Failed { + reason: format!("log stream ended with {reason}"), + }; + } + } else if !matches!(self.state, StreamState::Failed { .. }) { + self.state = if self.follow { + StreamState::FollowEnded + } else { + StreamState::Finished + }; + } + } + // `notice` (and unparsable lines the runner wraps as notices) + // and `heartbeat` carry no log line. + _ => {} + } + } + + /// The CLI process exited. A non-zero exit is a failure even after an + /// `end`; exit 0 without an `end` is a normal end. + pub fn exited(&mut self, code: Option) { + match code { + Some(0) => { + if matches!(self.state, StreamState::Connecting | StreamState::Following) { + self.state = if self.follow { + StreamState::FollowEnded + } else { + StreamState::Finished + }; + } + } + other => { + let status = crate::runner::exit_description(other); + if let StreamState::Failed { reason } = &mut self.state { + reason.push_str(&format!(" ({status})")); + } else if self.state != StreamState::Idle { + self.state = StreamState::Failed { + reason: format!("hack logs: {status}"), + }; + } + } + } + } + + fn push(&mut self, mut line: LogLine) { + line.seq = self.next_seq; + self.next_seq += 1; + if line.message.len() > MAX_LINE_BYTES { + let mut keep = MAX_LINE_BYTES; + while !line.message.is_char_boundary(keep) { + keep -= 1; + } + line.cut = line.message.len() - keep; + // A new string: `truncate` would keep the large allocation. + line.message = line.message[..keep].to_owned(); + } + self.bytes += line.retained_bytes(); + self.lines.push_back(line); + if self.paused { + self.unseen += 1; + } + // Strict: lines are capped on intake, so this always terminates + // with the buffer within both bounds. + while self.lines.len() > self.max_lines || self.bytes > self.max_bytes { + if let Some(old) = self.lines.pop_front() { + self.bytes -= old.retained_bytes(); + self.dropped += 1; + self.expanded.retain(|s| *s != old.seq); + } + } + } + + #[must_use] + pub fn len(&self) -> usize { + self.lines.len() + } + + #[must_use] + pub fn is_empty(&self) -> bool { + self.lines.is_empty() + } + + #[must_use] + pub fn bytes(&self) -> usize { + self.bytes + } + + /// Buffer indices of the lines to show (see [`Self::visible`]). + #[must_use] + pub fn visible_indices(&self) -> Vec { + let q = self.search.trim().to_lowercase(); + (0..self.lines.len()) + .filter(|&i| { + q.is_empty() || { + let l = &self.lines[i]; + l.message.to_lowercase().contains(&q) + || l.service + .as_deref() + .is_some_and(|s| s.to_lowercase().contains(&q)) + } + }) + .collect() + } + + /// Position in `visible` (buffer indices) of the line with sequence + /// `seq`, or of the next surviving line if it was evicted. Used to keep + /// a paused viewport on the lines being read while old lines drop off. + #[must_use] + pub fn position_of_seq(&self, visible: &[usize], seq: u64) -> usize { + visible.partition_point(|&i| self.lines.get(i).is_some_and(|l| l.seq < seq)) + } + + #[must_use] + pub fn line(&self, index: usize) -> Option<&LogLine> { + self.lines.get(index) + } + + /// The expanded lines still in the buffer, oldest first. + #[must_use] + pub fn expanded_lines(&self) -> Vec<&LogLine> { + self.lines + .iter() + .filter(|l| self.expanded.contains(&l.seq)) + .collect() + } + + /// Lines to show: all, or those matching the search (case-insensitive). + #[must_use] + pub fn visible(&self) -> Vec<&LogLine> { + let q = self.search.trim().to_lowercase(); + if q.is_empty() { + return self.lines.iter().collect(); + } + self.lines + .iter() + .filter(|l| { + l.message.to_lowercase().contains(&q) + || l.service + .as_deref() + .is_some_and(|s| s.to_lowercase().contains(&q)) + }) + .collect() + } + + /// The banner for the current state, if any. + #[must_use] + pub fn banner(&self) -> Option { + match &self.state { + StreamState::Idle | StreamState::Following => None, + StreamState::Connecting => Some("Connecting…".into()), + StreamState::Finished => Some("End of log".into()), + StreamState::FollowEnded => Some("Log stream ended".into()), + StreamState::Failed { reason } => Some(format!("Log stream failed: {reason}")), + StreamState::Unavailable { reason } => Some(reason.clone()), + } + } + + /// A finite read was opened while the project was stopped; once it runs, + /// the tab should switch to following. + #[must_use] + pub fn should_follow_now(&self, project_running: bool) -> bool { + project_running + && !self.follow + && !matches!( + self.state, + StreamState::Unavailable { .. } | StreamState::Idle + ) + } + + /// Offer Reconnect after the stream ended or failed. + #[must_use] + pub fn can_reconnect(&self) -> bool { + matches!( + self.state, + StreamState::FollowEnded | StreamState::Failed { .. } + ) + } +} + +/// Canned log streams for the Logs tab fixtures. +pub mod fixtures { + use super::{LogsModel, StreamState}; + use crate::models::{LogEntry, LogEvent}; + + #[derive(Debug, Clone, Copy, PartialEq, Eq)] + pub enum LogsFixture { + Following, + PausedSearch, + LongLines, + ServiceFilter, + FollowEnded, + Failed, + Native, + Stopped, + } + + impl LogsFixture { + pub const ALL: [Self; 8] = [ + Self::Following, + Self::PausedSearch, + Self::LongLines, + Self::ServiceFilter, + Self::FollowEnded, + Self::Failed, + Self::Native, + Self::Stopped, + ]; + + #[must_use] + pub fn name(self) -> &'static str { + match self { + Self::Following => "logs-following", + Self::PausedSearch => "logs-paused-search", + Self::LongLines => "logs-long-lines", + Self::ServiceFilter => "logs-service-filter", + Self::FollowEnded => "logs-follow-ended", + Self::Failed => "logs-failed", + Self::Native => "logs-native", + Self::Stopped => "logs-stopped", + } + } + + #[must_use] + pub fn parse(name: &str) -> Option { + Self::ALL.into_iter().find(|f| f.name() == name) + } + } + + fn log(service: &str, level: &str, message: &str) -> LogEvent { + LogEvent { + kind: "log".into(), + entry: Some(LogEntry { + service: Some(service.into()), + message: message.into(), + stream: Some(if level == "error" { "stderr" } else { "stdout" }.into()), + timestamp: None, + level: Some(level.into()), + truncated: None, + }), + ..LogEvent::default() + } + } + + fn ev(kind: &str, reason: Option<&str>, message: Option<&str>) -> LogEvent { + LogEvent { + kind: kind.into(), + reason: reason.map(str::to_owned), + message: message.map(str::to_owned), + ..LogEvent::default() + } + } + + /// The canned "following" stream, as events. + #[must_use] + pub fn events() -> Vec { + traffic() + } + + fn traffic() -> Vec { + let mut v = vec![ev("start", None, None)]; + for i in 0..40 { + v.push(match i % 5 { + 0 => log("api", "info", &format!("GET /health 200 {}ms", 1 + i % 4)), + 1 => log("web", "info", &format!("GET /products?page={i} 200 12ms")), + 2 => log( + "worker", + "info", + &format!("processed job {} in 41ms", 1000 + i), + ), + 3 => log("db", "info", "checkpoint complete: wrote 12 buffers"), + _ => log("api", "warn", "slow query: 312ms on orders.list"), + }); + } + v.push(log( + "worker", + "error", + "retrying job 1039: connection reset by peer", + )); + v + } + + /// A model in the state the fixture shows. + #[must_use] + pub fn model(f: LogsFixture) -> LogsModel { + let mut m = LogsModel::default(); + match f { + LogsFixture::Native => { + m.unavailable(crate::source::NATIVE_LOGS_NOTE); + return m; + } + LogsFixture::Stopped => { + m.start(false); + for e in traffic().iter().take(12) { + m.apply(e); + } + m.apply(&ev("end", Some("eof"), None)); + m.exited(Some(0)); + return m; + } + _ => m.start(true), + } + if f == LogsFixture::ServiceFilter { + m.set_service(Some("worker".into())); + for e in traffic().iter().filter(|e| { + e.entry + .as_ref() + .is_none_or(|x| x.service.as_deref() == Some("worker")) + }) { + m.apply(e); + } + return m; + } + for e in traffic() { + m.apply(&e); + } + match f { + LogsFixture::PausedSearch => { + m.skipped = 128; + m.set_paused(true); + for e in traffic().iter().skip(1).take(7) { + m.apply(e); + } + m.search = "health".into(); + } + LogsFixture::LongLines => { + let long = format!( + "payload dump: {}", + "{\"id\":42,\"items\":[1,2,3],\"note\":\"lorem ipsum dolor sit amet\"} " + .repeat(120) + ); + m.apply(&log("api", "info", &long)); + let mut cut = log("web", "info", "response body (truncated by hack) …"); + if let Some(entry) = cut.entry.as_mut() { + entry.truncated = Some(true); + } + m.apply(&cut); + } + LogsFixture::FollowEnded => { + m.apply(&ev("end", Some("eof"), None)); + m.exited(Some(0)); + } + LogsFixture::Failed => { + // A stream failure while Docker itself is up, so the + // render agrees with the status bar's "Docker ok". + m.apply(&ev("error", None, Some("Loki WebSocket error"))); + m.exited(Some(1)); + } + _ => {} + } + debug_assert!(!matches!(m.state, StreamState::Idle)); + m + } +} + +#[cfg(test)] +mod tests { + use super::fixtures::{LogsFixture, model}; + use super::*; + use crate::models::LogEntry; + + fn line(msg: &str) -> LogEvent { + LogEvent { + kind: "log".into(), + entry: Some(LogEntry { + service: Some("api".into()), + message: msg.into(), + ..LogEntry::default() + }), + ..LogEvent::default() + } + } + + fn ev(kind: &str, reason: Option<&str>) -> LogEvent { + LogEvent { + kind: kind.into(), + reason: reason.map(str::to_owned), + ..LogEvent::default() + } + } + + #[test] + fn ring_buffer_bounds_lines_and_bytes() { + let mut m = LogsModel::with_limits(3, 1_000); + m.start(true); + for i in 0..5 { + m.apply(&line(&format!("l{i}"))); + } + assert_eq!(m.len(), 3); + assert_eq!(m.dropped, 2); + assert_eq!(m.visible()[0].message, "l2"); + + // "api" (3) + 6 = 9 bytes per line; the bound is strict. + let mut b = LogsModel::with_limits(100, 10); + b.start(true); + b.apply(&line("aaaaaa")); + b.apply(&line("bbbbbb")); + assert_eq!(b.len(), 1, "byte bound drops the oldest"); + assert!(b.bytes() <= 10); + b.apply(&line(&"z".repeat(20))); + assert!(b.bytes() <= 10, "a line over the bound is not kept"); + } + + #[test] + fn huge_lines_are_capped_on_intake() { + let mut m = LogsModel::default(); + m.start(true); + m.apply(&line(&"y".repeat(MAX_LINE_BYTES + 5000))); + let l = m.visible()[0]; + assert_eq!(l.message.len(), MAX_LINE_BYTES); + assert!( + l.message.capacity() < MAX_LINE_BYTES + 1024, + "no oversized allocation kept" + ); + assert_eq!(l.cut, 5000); + assert_eq!(m.bytes(), MAX_LINE_BYTES + "api".len()); + } + + #[test] + fn loki_error_then_end_error_keeps_the_specific_reason() { + let mut m = LogsModel::default(); + m.start(true); + m.apply(&LogEvent { + kind: "error".into(), + message: Some("Loki query failed: 502".into()), + ..LogEvent::default() + }); + m.apply(&ev("end", Some("error"))); + m.exited(Some(1)); + assert_eq!( + m.state, + StreamState::Failed { + reason: "Loki query failed: 502 (exit 1)".into() + } + ); + } + + #[test] + fn paused_anchor_survives_eviction() { + let mut m = LogsModel::with_limits(5, 1_000_000); + m.start(true); + for i in 0..5 { + m.apply(&line(&format!("l{i}"))); + } + // Reading line seq 3 (index 3) while paused. + let anchor = m.line(3).unwrap().seq; + m.set_paused(true); + m.apply(&line("l5")); + m.apply(&line("l6")); // evicts seq 0 and 1 + let v = m.visible_indices(); + let pos = m.position_of_seq(&v, anchor); + assert_eq!( + m.line(v[pos]).unwrap().message, + "l3", + "same line, new index" + ); + assert_eq!(pos, 1); + // An evicted anchor lands on the next surviving line. + assert_eq!(m.line(v[m.position_of_seq(&v, 0)]).unwrap().message, "l2"); + } + + #[test] + fn pause_keeps_reading_and_counts_unseen() { + let mut m = LogsModel::default(); + m.start(true); + m.set_paused(true); + m.apply(&line("a")); + m.apply(&line("b")); + assert_eq!((m.len(), m.unseen), (2, 2)); + m.set_paused(false); + assert_eq!(m.unseen, 0); + } + + #[test] + fn search_filters_view_not_buffer() { + let mut m = LogsModel::default(); + m.start(true); + m.apply(&line("GET /health")); + m.apply(&line("POST /orders")); + m.search = "HEALTH".into(); + assert_eq!(m.visible().len(), 1); + assert_eq!(m.visible_indices(), vec![0]); + assert_eq!(m.line(0).unwrap().message, "GET /health"); + assert_eq!(m.len(), 2); + m.search = "api".into(); + assert_eq!(m.visible().len(), 2, "service name matches too"); + } + + #[test] + fn service_filter_is_positional_and_restarts_once() { + let mut m = LogsModel::default(); + assert_eq!(m.cli_args(true), ["logs", "--json", "--tail", "500"]); + assert!(m.set_service(Some("api".into()))); + assert!(!m.set_service(Some("api".into()))); + assert_eq!(m.cli_args(true).last().map(String::as_str), Some("api")); + assert!( + !m.cli_args(true) + .iter() + .any(|a| a == "--services" || a == "--compose") + ); + assert_eq!( + m.cli_args(false), + ["logs", "--json", "--no-follow", "--tail", "200", "api"] + ); + } + + #[test] + fn long_lines_are_cut_on_a_char_boundary_until_expanded() { + let msg = "é".repeat(DISPLAY_CUT); // 2 bytes each + let l = LogLine { + seq: 7, + service: None, + message: msg.clone(), + stderr: false, + error_level: false, + timestamp: None, + truncated_by_hack: false, + cut: 0, + }; + let (shown, hidden) = l.display(false); + assert!(shown.len() <= DISPLAY_CUT); + assert_eq!(shown.len() + hidden, msg.len()); + assert_eq!(l.display(true), (msg.as_str(), 0)); + let mut m = LogsModel::default(); + m.toggle_expanded(7); + assert!(m.is_expanded(7)); + m.toggle_expanded(7); + assert!(!m.is_expanded(7)); + } + + #[test] + fn compose_stderr_log_and_failed_end_are_a_failure() { + let mut model = LogsModel::default(); + model.start(true); + let records = [ + r#"{"type":"log","entry":{"message":"Cannot connect to the Docker daemon","stream":"stderr"}}"#, + r#"{"type":"end","reason":"exit:1"}"#, + ]; + for record in records { + model.apply(&serde_json::from_str(record).unwrap()); + } + model.exited(Some(1)); + assert!(model.visible()[0].stderr); + assert!(matches!(model.state, StreamState::Failed { .. })); + assert!(model.banner().unwrap().contains("exit 1")); + } + + #[test] + fn end_states_are_distinguished() { + let mut follow_end = LogsModel::default(); + follow_end.start(true); + follow_end.apply(&ev("start", None)); + assert_eq!(follow_end.state, StreamState::Following); + follow_end.apply(&ev("end", Some("closed"))); + follow_end.exited(Some(0)); + assert_eq!(follow_end.state, StreamState::FollowEnded); + assert!(follow_end.can_reconnect()); + + let mut finished = LogsModel::default(); + finished.start(false); + finished.apply(&line("x")); + finished.apply(&ev("end", Some("eof"))); + finished.exited(Some(0)); + assert_eq!(finished.state, StreamState::Finished); + assert!(!finished.can_reconnect()); + + let mut bad_end = LogsModel::default(); + bad_end.start(true); + bad_end.apply(&line("x")); + bad_end.apply(&ev("end", Some("exit:1"))); + assert!(matches!(bad_end.state, StreamState::Failed { .. })); + + let mut nonzero = LogsModel::default(); + nonzero.start(true); + nonzero.apply(&ev("end", Some("eof"))); + nonzero.exited(Some(1)); + assert!( + matches!(nonzero.state, StreamState::Failed { .. }), + "exit≠0 wins over end" + ); + + let mut closed = LogsModel::default(); + closed.start(true); + closed.close(); + closed.exited(None); + assert_eq!( + closed.state, + StreamState::Idle, + "app-closed stream shows nothing" + ); + assert_eq!(closed.banner(), None); + } + + #[test] + fn a_stopped_read_switches_to_follow_when_the_project_starts() { + let mut m = LogsModel::default(); + m.start(false); + m.apply(&ev("end", Some("eof"))); + m.exited(Some(0)); + assert!(!m.should_follow_now(false), "still stopped"); + assert!(m.should_follow_now(true), "started: follow"); + m.start(true); + assert!(!m.should_follow_now(true), "already following"); + let mut n = LogsModel::default(); + n.unavailable("native"); + assert!(!n.should_follow_now(true), "native stays unavailable"); + } + + #[test] + fn unavailable_clears_the_previous_projects_lines() { + let mut m = LogsModel::default(); + m.start(true); + m.apply(&line("old project")); + m.skipped = 7; + m.unavailable("no folder"); + assert_eq!(m.skipped, 0, "previous project's overflow count is cleared"); + assert!(m.is_empty()); + assert!(m.visible().is_empty()); + assert_eq!(m.banner().as_deref(), Some("no folder")); + } + + #[test] + fn every_logs_fixture_builds_its_state() { + for f in LogsFixture::ALL { + assert_eq!(LogsFixture::parse(f.name()), Some(f)); + let m = model(f); + match f { + LogsFixture::Following | LogsFixture::LongLines | LogsFixture::ServiceFilter => { + assert_eq!(m.state, StreamState::Following, "{f:?}"); + } + LogsFixture::PausedSearch => { + assert!(m.paused && m.unseen > 0 && !m.search.is_empty()) + } + LogsFixture::FollowEnded => assert_eq!(m.state, StreamState::FollowEnded), + LogsFixture::Failed => assert!(matches!(m.state, StreamState::Failed { .. })), + LogsFixture::Native => assert!(matches!(m.state, StreamState::Unavailable { .. })), + LogsFixture::Stopped => assert_eq!(m.state, StreamState::Finished), + } + } + assert!( + model(LogsFixture::ServiceFilter) + .visible() + .iter() + .all(|l| l.service.as_deref() == Some("worker")) + ); + } +} diff --git a/apps/desktop/core/src/runner.rs b/apps/desktop/core/src/runner.rs index a86a9738e..724ca9d9f 100644 --- a/apps/desktop/core/src/runner.rs +++ b/apps/desktop/core/src/runner.rs @@ -12,7 +12,11 @@ use crate::process::Child; use std::io::{BufRead, BufReader, Read}; use std::path::{Path, PathBuf}; use std::process::{Command, Stdio}; -use std::sync::mpsc; +use std::sync::{ + Arc, + atomic::{AtomicU64, Ordering}, + mpsc, +}; use std::thread; use std::time::{Duration, Instant}; @@ -318,55 +322,300 @@ impl CliRunner { /// `NotFound` or `Spawn`. pub fn stream_logs(&self, args: &[&str], path: Option<&Path>) -> Result { let mut child = self.spawn(args, path)?; + // Diagnostics must not fill the stderr pipe and block ndjson output. + // Drain concurrently with a bounded tail, as for finite CLI reads. + let stderr = read_all(child.stderr.take(), Keep::Tail(STDERR_KEEP)); let stdout = child.stdout.take(); - let (tx, rx) = mpsc::channel(); + // Never block on the UI: the CLI ignores stdout backpressure and + // would otherwise buffer output in its own process without a bound. + let (tx, rx) = mpsc::sync_channel(STREAM_QUEUE); + let skipped = Arc::new(AtomicU64::new(0)); + let reader_skipped = skipped.clone(); let reader = thread::spawn(move || { - let Some(stdout) = stdout else { return }; - for line in BufReader::new(stdout).lines() { - let Ok(line) = line else { break }; + let mut diagnostics = String::new(); + let Some(stdout) = stdout else { + return diagnostics; + }; + let mut r = BufReader::new(stdout); + while let Some((line, cut)) = read_bounded_line(&mut r, MAX_RAW_LINE) { if line.trim().is_empty() { continue; } - let event = serde_json::from_str::(&line).unwrap_or_else(|_| LogEvent { - kind: "notice".into(), - message: Some(line), - ..LogEvent::default() - }); - if tx.send(event).is_err() { - break; + // A record cut at MAX_RAW_LINE is no longer valid JSON: show + // a visible marker instead of dropping it silently. + let mut event = if cut { + oversized_record_event() + } else { + match serde_json::from_str::(&line) { + Ok(event) if !event.kind.is_empty() => event, + _ => { + append_diagnostic(&mut diagnostics, &line); + LogEvent { + kind: "notice".into(), + message: Some(line), + ..LogEvent::default() + } + } + } + }; + // Cap before queueing, so the queue's memory is bounded too. + cap_event(&mut event); + match tx.try_send(event) { + Ok(()) => {} + Err(mpsc::TrySendError::Full(_)) => { + reader_skipped.fetch_add(1, Ordering::Relaxed); + } + Err(mpsc::TrySendError::Disconnected(_)) => break, } } + diagnostics }); Ok(LogStream { child, - rx, + rx: Some(rx), reader: Some(reader), + skipped, + stderr, + stdout_diagnostics: String::new(), + stderr_diagnostics: None, + reaped: false, }) } } +/// Keep only a bounded UTF-8 diagnostic tail, including pretty JSON fragments. +fn append_diagnostic(text: &mut String, line: &str) { + text.push_str(line); + text.push('\n'); + if text.len() > STDERR_KEEP { + let mut cut = text.len() - STDERR_KEEP; + while !text.is_char_boundary(cut) { + cut += 1; + } + *text = text[cut..].to_owned(); + } +} + +/// Events queued between the reader thread and the UI. +pub const STREAM_QUEUE: usize = 1024; +/// Cap for the top-level `message` of an event (notices, errors). +pub const MAX_NOTICE_BYTES: usize = 4 * 1024; +/// Cap for every other short text field (kind, timestamps, service, ...). +pub const MAX_FIELD_BYTES: usize = 256; +/// Upper bound on the text one queued event retains after `cap_event`: +/// the log message, the notice message and eight short fields. +pub const MAX_EVENT_BYTES: usize = + crate::logs::MAX_LINE_BYTES + MAX_NOTICE_BYTES + 8 * MAX_FIELD_BYTES; +/// Upper bound on queued text: `STREAM_QUEUE * MAX_EVENT_BYTES` (~70 MiB). +pub const MAX_QUEUED_BYTES: usize = STREAM_QUEUE * MAX_EVENT_BYTES; + +/// Bound every text field an event retains, so one event never holds more +/// than `MAX_EVENT_BYTES` (cut on a character boundary into a new, small +/// string). A capped log message marks its entry truncated. +fn cap_event(e: &mut LogEvent) { + fn cap(s: &mut String, max: usize) -> bool { + if s.len() <= max { + return false; + } + let mut keep = max; + while !s.is_char_boundary(keep) { + keep -= 1; + } + *s = s[..keep].to_owned(); + true + } + fn cap_opt(s: &mut Option, max: usize) { + if let Some(v) = s.as_mut() { + cap(v, max); + } + } + cap(&mut e.kind, MAX_FIELD_BYTES); + cap_opt(&mut e.ts, MAX_FIELD_BYTES); + cap_opt(&mut e.backend, MAX_FIELD_BYTES); + cap_opt(&mut e.reason, MAX_FIELD_BYTES); + cap_opt(&mut e.message, MAX_NOTICE_BYTES); + if let Some(entry) = e.entry.as_mut() { + cap_opt(&mut entry.service, MAX_FIELD_BYTES); + cap_opt(&mut entry.stream, MAX_FIELD_BYTES); + cap_opt(&mut entry.timestamp, MAX_FIELD_BYTES); + cap_opt(&mut entry.level, MAX_FIELD_BYTES); + if cap(&mut entry.message, crate::logs::MAX_LINE_BYTES) { + entry.truncated = Some(true); + } + } +} + +/// Text bytes an event retains (for the bound's test). +#[cfg(test)] +fn event_bytes(e: &LogEvent) -> usize { + let o = |s: &Option| s.as_ref().map_or(0, String::len); + e.kind.len() + + o(&e.ts) + + o(&e.backend) + + o(&e.reason) + + o(&e.message) + + e.entry.as_ref().map_or(0, |x| { + x.message.len() + o(&x.service) + o(&x.stream) + o(&x.timestamp) + o(&x.level) + }) +} + +/// Longest raw ndjson line read; the rest of a longer line is skipped. +pub const MAX_RAW_LINE: usize = 1024 * 1024; + +/// Read one `\n`-terminated line, keeping at most `max` bytes of it. +/// `None` at end of input (or on a read error). +fn read_bounded_line(r: &mut R, max: usize) -> Option<(String, bool)> { + let mut buf = Vec::new(); + let mut saw_any = false; + let mut cut = false; + loop { + let chunk = r.fill_buf().ok()?; + if chunk.is_empty() { + return saw_any.then(|| (String::from_utf8_lossy(&buf).into_owned(), cut)); + } + saw_any = true; + let (take, done) = match chunk.iter().position(|b| *b == b'\n') { + Some(i) => (i, true), + None => (chunk.len(), false), + }; + let room = max.saturating_sub(buf.len()); + cut |= take > room; + buf.extend_from_slice(&chunk[..take.min(room)]); + r.consume(if done { take + 1 } else { take }); + if done { + return Some((String::from_utf8_lossy(&buf).into_owned(), cut)); + } + } +} + +/// Shown in place of a record longer than `MAX_RAW_LINE`. +fn oversized_record_event() -> LogEvent { + LogEvent { + kind: "log".into(), + entry: Some(crate::models::LogEntry { + service: None, + message: format!( + "[a log record over {} KB was dropped by the app; use Copy command to read it in a terminal]", + MAX_RAW_LINE / 1024 + ), + stream: None, + timestamp: None, + level: Some("warn".into()), + truncated: Some(true), + }), + ..LogEvent::default() + } +} + +/// One poll of a stream: up to `max` events, and the exit code once the +/// process has exited **and** retained events have been delivered. Queue +/// overflow is discarded and reported separately in `skipped`. +#[derive(Debug, Default)] +pub struct StreamPoll { + pub events: Vec, + pub exit: Option>, + /// Events discarded while the UI queue was full since the previous poll. + pub skipped: u64, +} + /// A running log stream. Drop retries group cleanup for at most two seconds, /// then detaches a reader still held open by a descendant outside that group. pub struct LogStream { child: Child, - rx: mpsc::Receiver, - reader: Option>, + /// `None` only while dropping (see `Drop`). + rx: Option>, + reader: Option>, + skipped: Arc, + stderr: mpsc::Receiver, + stdout_diagnostics: String, + stderr_diagnostics: Option, + /// `poll` observed the exit and reaped the child: its PID (and group id) + /// may now belong to another process, so Drop must not signal it. + reaped: bool, } impl LogStream { /// Events received so far, without blocking. pub fn drain(&self) -> Vec { - self.rx.try_iter().collect() + self.rx + .as_ref() + .map(|rx| rx.try_iter().collect()) + .unwrap_or_default() } /// Wait up to `timeout` for the next event. pub fn next_timeout(&self, timeout: Duration) -> Option { - self.rx.recv_timeout(timeout).ok() + self.rx.as_ref()?.recv_timeout(timeout).ok() + } + + /// Up to `max` queued events. `exit` is set only when the reader has + /// reached end of output and the queue is empty; overflow is reported + /// separately. Its inner value is `None` when a signal ended the process. + pub fn poll(&mut self, max: usize) -> StreamPoll { + // Checked before draining: once the reader is done, nothing new + // arrives, so an under-full batch means the queue is exhausted. + let reader_done = self + .reader + .as_ref() + .is_none_or(thread::JoinHandle::is_finished); + let mut events: Vec = self + .rx + .as_ref() + .map(|rx| rx.try_iter().take(max).collect()) + .unwrap_or_default(); + let mut exit = if reader_done && events.len() < max { + if let Some(reader) = self.reader.take() { + self.stdout_diagnostics = reader.join().unwrap_or_default(); + } + let status = self.child.try_wait().ok().flatten(); + self.reaped |= status.is_some(); + status.map(|s| s.code()) + } else { + None + }; + if exit.is_some() && self.stderr_diagnostics.is_none() { + match self.stderr.try_recv() { + Ok(diagnostics) => self.stderr_diagnostics = Some(diagnostics), + Err(mpsc::TryRecvError::Empty) => exit = None, + Err(mpsc::TryRecvError::Disconnected) => {} + } + } + if let Some(code) = exit.filter(|code| *code != Some(0)) { + let stderr = self + .stderr_diagnostics + .as_ref() + .map_or("", |d| d.text.as_str()); + let reason = match parse_result(&self.stdout_diagnostics, stderr, code) { + Err(RunError::Cli(error)) => error.message, + _ if !stderr.trim().is_empty() => stderr.trim().to_owned(), + _ => self.stdout_diagnostics.trim().to_owned(), + }; + if !reason.is_empty() { + let mut event = LogEvent { + kind: "error".into(), + message: Some(reason), + ..LogEvent::default() + }; + cap_event(&mut event); + events.push(event); + } + } + StreamPoll { + events, + exit, + skipped: self.skipped.swap(0, Ordering::Relaxed), + } } } impl Drop for LogStream { fn drop(&mut self) { + // Disconnect before signalling the owned process group and joining. + drop(self.rx.take()); + if self.reaped { + // Already reaped by `poll`, which also joined the reader. + return; + } signal_kill_group(&mut self.child); if let Some(r) = self.reader.take() { // A group signal can race a descendant fork on macOS. Keep the @@ -387,6 +636,11 @@ impl Drop for LogStream { } } +#[must_use] +pub fn exit_description(code: Option) -> String { + code.map_or_else(|| "killed by signal".into(), |code| format!("exit {code}")) +} + /// Which part of a pipe to keep while draining it. #[derive(Clone, Copy)] enum Keep { @@ -513,9 +767,17 @@ fn kill_group(child: &mut Child) { let _ = child.wait(); } +#[cfg(test)] +thread_local! { + /// Group signals sent by this thread (tests prove a reaped stream sends none). + static GROUP_SIGNALS: std::cell::Cell = const { std::cell::Cell::new(0) }; +} + /// Only for the log-drop path, which has not reaped its child. Avoid /// Child::kill here: its status check may reap and release the PID reservation. fn signal_kill_group(child: &mut Child) { + #[cfg(test)] + GROUP_SIGNALS.with(|n| n.set(n.get() + 1)); if let Ok(pid) = i32::try_from(child.id()) { // SAFETY: plain syscall on the group we created with process_group(0). unsafe { @@ -665,15 +927,17 @@ mod tests { ), ); let fake = Self { dir }; - // macOS may scan a newly written executable on first launch. Keep - // that delay outside every test's product-timeout measurement. + // Keep macOS first-launch scanning outside product timeouts. fake.runner() .run_json::(&["__warmup"], None, Duration::from_secs(60)) .expect("fake CLI warmup"); fake } fn runner(&self) -> CliRunner { - CliRunner::new(self.dir.join("hack")).with_env("HACK_HOME", &self.dir.to_string_lossy()) + CliRunner::new(self.dir.join("hack")) + .with_env("HACK_HOME", &self.dir.to_string_lossy()) + .with_env("HOME", &self.dir.to_string_lossy()) + .with_env("HACK_DAEMON_AUTOSTART_DISABLE", "1") } } @@ -773,8 +1037,9 @@ mod tests { #[test] fn timeout_kills_the_process_group() { // The child spawns a grandchild that would outlive a plain kill. - // FakeHack warms the script before this timed run; `ready` proves - // the body ran, rather than timing out before it exercised cleanup. + // The timeout leaves room for a slow first start of a new script + // (macOS scans freshly written executables); `ready` tells a slow + // start apart from a runner failure. let f = FakeHack::new( "echo ready > \"$HACK_HOME/ready\"\nsleep 300 &\necho $! > \"$HACK_HOME/grandchild\"\nwait", ); @@ -810,6 +1075,221 @@ mod tests { assert_retired(pid); } + #[test] + fn flood_finishes_without_ui_drain_and_reports_bounded_loss() { + let total = STREAM_QUEUE + 10_000; + let f = FakeHack::new(&format!( + "i=0; while [ $i -lt {total} ]; do echo '{{\"type\":\"log\",\"entry\":{{\"message\":\"flood\"}}}}'; i=$((i+1)); done" + )); + let mut stream = f.runner().stream_logs(&["logs"], None).unwrap(); + let deadline = Instant::now() + T; + // No receiver calls: both the child and reader must finish even + // though the UI has consumed nothing and the queue is full. + while !stream.reader.as_ref().unwrap().is_finished() { + assert!(Instant::now() < deadline, "reader blocked on UI queue"); + thread::sleep(Duration::from_millis(5)); + } + while stream.child.try_wait().unwrap().is_none() { + assert!(Instant::now() < deadline, "writer blocked on stdout"); + thread::sleep(Duration::from_millis(5)); + } + let poll = stream.poll(STREAM_QUEUE + 1); + assert_eq!(poll.events.len(), STREAM_QUEUE); + assert_eq!(poll.skipped, 10_000); + assert!(poll.events.iter().map(event_bytes).sum::() <= MAX_QUEUED_BYTES); + assert_eq!( + stream.poll(STREAM_QUEUE + 1).skipped, + 0, + "count delivered once" + ); + } + + fn failed_stream_banner(body: &str) -> String { + let f = FakeHack::new(body); + let mut stream = f.runner().stream_logs(&["logs"], None).unwrap(); + let mut model = crate::logs::LogsModel::default(); + model.start(true); + let deadline = Instant::now() + T; + loop { + let poll = stream.poll(100); + for event in poll.events { + model.apply(&event); + } + if let Some(code) = poll.exit { + model.exited(code); + return model.banner().unwrap(); + } + assert!(Instant::now() < deadline, "stream did not exit"); + thread::sleep(Duration::from_millis(5)); + } + } + + #[test] + fn pretty_failure_envelope_reaches_the_banner_with_human_status() { + let banner = failed_stream_banner( + r#"cat <<'JSON' +{ + "ok": false, + "error": { + "code": "E_PROJECT_NOT_FOUND", + "message": "No project found at /work/missing" + } +} +JSON +exit 1"#, + ); + assert!( + banner.contains("No project found at /work/missing (exit 1)"), + "{banner}" + ); + assert!(!banner.contains("Some(")); + } + + #[test] + fn stderr_failure_reason_reaches_the_banner_and_diagnostics_are_bounded() { + let banner = failed_stream_banner("echo 'docker executable unavailable' >&2; exit 127"); + assert!( + banner.contains("docker executable unavailable (exit 127)"), + "{banner}" + ); + let mut diagnostics = String::new(); + for _ in 0..100 { + append_diagnostic(&mut diagnostics, &"é".repeat(STDERR_KEEP)); + assert!(diagnostics.len() <= STDERR_KEEP); + assert!(diagnostics.capacity() <= STDERR_KEEP * 2); + } + assert_eq!(exit_description(None), "killed by signal"); + } + + #[test] + fn log_stream_drains_stderr_without_blocking_stdout() { + let f = FakeHack::new( + "head -c 2097152 /dev/zero >&2; echo '{\"type\":\"log\",\"entry\":{\"message\":\"after diagnostics\"}}'", + ); + let s = f.runner().stream_logs(&["logs"], None).unwrap(); + let event = s.next_timeout(T).expect("stderr must not block stdout"); + assert_eq!(event.entry.unwrap().message, "after diagnostics"); + } + + #[test] + fn log_stream_reports_exit_only_after_every_line() { + // Many lines, then exit 3: the exit must not be reported while lines + // are still queued or unread, even with a small batch size. + let f = FakeHack::new( + "i=0; while [ $i -lt 300 ]; do echo \"{\\\"type\\\":\\\"log\\\",\\\"entry\\\":{\\\"message\\\":\\\"l$i\\\"}}\"; i=$((i+1)); done; echo '{\"type\":\"end\",\"reason\":\"eof\"}'; exit 3", + ); + let mut s = f.runner().stream_logs(&["logs"], None).unwrap(); + let mut got = 0; + let mut exit = None; + for _ in 0..500 { + let p = s.poll(7); + got += p.events.len(); + if let Some(code) = p.exit { + exit = Some(code); + break; + } + thread::sleep(Duration::from_millis(5)); + } + assert_eq!(got, 301, "every line and the end event delivered first"); + assert_eq!(exit, Some(Some(3))); + } + + #[test] + fn dropping_a_stream_after_poll_reaped_it_sends_no_signal() { + let f = FakeHack::new("echo '{\"type\":\"end\",\"reason\":\"eof\"}'"); + let mut s = f.runner().stream_logs(&["logs"], None).unwrap(); + let deadline = Instant::now() + Duration::from_secs(60); + while s.poll(64).exit.is_none() { + assert!(Instant::now() < deadline, "stream never reported its exit"); + thread::sleep(Duration::from_millis(20)); + } + let before = GROUP_SIGNALS.with(std::cell::Cell::get); + drop(s); + assert_eq!( + GROUP_SIGNALS.with(std::cell::Cell::get), + before, + "a reaped PID may be reused; Drop must not signal it" + ); + } + + #[test] + fn dropping_a_stream_with_a_full_queue_does_not_hang() { + // Far more lines than STREAM_QUEUE, never polled: overflow is + // discarded. Drop must join the reader and retire its child. + let f = FakeHack::new( + "i=0; while [ $i -lt 20000 ]; do echo '{\"type\":\"heartbeat\"}'; i=$((i+1)); done; exec sleep 300", + ); + let s = f.runner().stream_logs(&["logs"], None).unwrap(); + thread::sleep(Duration::from_millis(500)); + let pid = i32::try_from(s.child.id()).unwrap(); + assert!( + !process_gone(pid), + "positive control: stream child is alive" + ); + completes(move || drop(s)); + assert_retired(pid); + } + + #[test] + fn oversized_records_leave_a_visible_marker() { + let f = FakeHack::new(&format!( + "printf '{{\"type\":\"log\",\"entry\":{{\"message\":\"'; head -c {} /dev/zero | tr '\\0' x; printf '\"}}}}\\n'; echo '{{\"type\":\"log\",\"entry\":{{\"message\":\"after\"}}}}'", + MAX_RAW_LINE + 1000 + )); + let s = f.runner().stream_logs(&["logs"], None).unwrap(); + let a = s.next_timeout(T).expect("marker"); + let b = s.next_timeout(T).expect("next event"); + let marker = a.entry.unwrap(); + assert_eq!(marker.truncated, Some(true)); + assert!(marker.message.contains("dropped by the app")); + assert_eq!(b.entry.unwrap().message, "after"); + } + + #[test] + fn queued_events_are_capped_before_send() { + let big = "x".repeat(crate::logs::MAX_LINE_BYTES * 3); + let mut e = LogEvent { + kind: big.clone(), + ts: Some(big.clone()), + backend: Some(big.clone()), + reason: Some(big.clone()), + message: Some(big.clone()), + entry: Some(crate::models::LogEntry { + service: Some(big.clone()), + message: big.clone(), + stream: Some(big.clone()), + timestamp: Some(big.clone()), + level: Some(big), + truncated: None, + }), + }; + cap_event(&mut e); + assert!(event_bytes(&e) <= MAX_EVENT_BYTES, "{}", event_bytes(&e)); + let entry = e.entry.as_ref().unwrap(); + assert_eq!(entry.message.len(), crate::logs::MAX_LINE_BYTES); + assert!(entry.message.capacity() < crate::logs::MAX_LINE_BYTES + 1024); + assert_eq!(entry.truncated, Some(true)); + assert_eq!(e.message.as_ref().unwrap().len(), MAX_NOTICE_BYTES); + assert_eq!(entry.timestamp.as_ref().unwrap().len(), MAX_FIELD_BYTES); + assert_eq!(e.reason.as_ref().unwrap().len(), MAX_FIELD_BYTES); + const { assert!(MAX_QUEUED_BYTES < 80 * 1024 * 1024) }; + } + + #[test] + fn bounded_line_reader_skips_the_tail_of_huge_lines() { + let data = format!("{}\nnext\n", "x".repeat(50)); + let mut r = std::io::BufReader::with_capacity(8, data.as_bytes()); + assert_eq!( + read_bounded_line(&mut r, 10).unwrap(), + ("x".repeat(10), true) + ); + assert_eq!( + read_bounded_line(&mut r, 10).unwrap(), + ("next".into(), false) + ); + assert!(read_bounded_line(&mut r, 10).is_none()); + } + #[test] fn path_lookup_ignores_relative_entries() { let f = FakeHack::new("echo '{}'"); diff --git a/apps/desktop/core/src/source.rs b/apps/desktop/core/src/source.rs index e769fce5b..6a154f824 100644 --- a/apps/desktop/core/src/source.rs +++ b/apps/desktop/core/src/source.rs @@ -115,8 +115,34 @@ impl DaemonRecovery { /// Native finite logs are a single-service snapshot (not ndjson), so the /// combined tail is not available for native projects in v0. -pub const NATIVE_LOGS_NOTE: &str = - "Native logs are read one service at a time; they arrive with the Logs tab."; +pub const NATIVE_LOGS_NOTE: &str = "Native logs are read one service at a time and are not shown in this preview yet. In a terminal: HACK_RUNTIME_BACKEND=native hack logs --no-follow --tail 200"; + +/// A running log stream: live `hack logs --json`, or canned events. +pub trait LogFeed: Send { + /// Up to `max` events without blocking, overflow count, and exit after + /// retained events are delivered (see [`crate::runner::LogStream::poll`]). + fn poll(&mut self, max: usize) -> crate::runner::StreamPoll; +} + +impl LogFeed for crate::runner::LogStream { + fn poll(&mut self, max: usize) -> crate::runner::StreamPoll { + crate::runner::LogStream::poll(self, max) + } +} + +/// Canned events delivered in batches; then it keeps "following" silently. +pub struct FixtureFeed(pub Vec); + +impl LogFeed for FixtureFeed { + fn poll(&mut self, max: usize) -> crate::runner::StreamPoll { + let n = max.min(self.0.len()); + crate::runner::StreamPoll { + events: self.0.drain(..n).collect(), + exit: None, + skipped: 0, + } + } +} pub trait Source: Send + Sync { /// Read names-only env metadata; diagnostics must never contain CLI output. @@ -152,6 +178,18 @@ pub trait Source: Send + Sync { /// Why no tail could be read; shown in place of the tail. fn log_tail(&self, repo_root: &Path, native: bool, lines: u32) -> Result, String>; + + /// Start a log stream for the Logs tab (`args` from `LogsModel::cli_args`). + /// Dropping the feed stops the stream. + /// + /// # Errors + /// Why no stream can be opened (native projects in v0, spawn failure). + fn open_logs( + &self, + repo_root: &Path, + native: bool, + args: &[String], + ) -> Result, String>; } /// Live data from a matching hackd. CLI listing fallback needs R18: @@ -199,6 +237,22 @@ impl LiveSource { } impl Source for LiveSource { + fn open_logs( + &self, + repo_root: &Path, + native: bool, + args: &[String], + ) -> Result, String> { + if native { + return Err(NATIVE_LOGS_NOTE.into()); + } + let args: Vec<&str> = args.iter().map(String::as_str).collect(); + self.runner + .stream_logs(&args, Some(repo_root)) + .map(|s| Box::new(s) as Box) + .map_err(|e| e.to_string()) + } + fn env(&self, root: &Path, native: bool) -> crate::env::EnvView { // env list walks ancestors. Refuse missing or shadowed project markers // instead of attributing an ancestor's names to the selected project. @@ -499,6 +553,32 @@ fn cli_selects_exactly(root: &Path) -> bool { pub struct FixtureSource(pub Fixture); impl Source for FixtureSource { + fn open_logs( + &self, + _repo_root: &Path, + native: bool, + args: &[String], + ) -> Result, String> { + if native { + return Err(NATIVE_LOGS_NOTE.into()); + } + // Replay the "following" fixture, filtered to the chosen service. + // The service filter is the trailing positional, if any. + let service = args + .last() + .filter(|a| !a.starts_with("--") && a.parse::().is_err() && a.as_str() != "logs"); + let events = crate::logs::fixtures::events() + .into_iter() + .filter(|e| { + service.is_none() + || e.entry + .as_ref() + .is_none_or(|x| x.service.as_ref() == service) + }) + .collect(); + Ok(Box::new(FixtureFeed(events))) + } + fn env(&self, _root: &Path, _native: bool) -> crate::env::EnvView { crate::env::fixture(self.0.name()) } @@ -677,13 +757,19 @@ esac // Complete macOS first-launch scanning before the fake daemon's // bounded accept window starts. version --json has no fixture effects. CliRunner::new(&hack) + .with_env("HOME", &dir.to_string_lossy()) + .with_env("HACK_HOME", &dir.to_string_lossy()) + .with_env("HACK_DAEMON_AUTOSTART_DISABLE", "1") .version(Duration::from_secs(60)) .unwrap(); let (daemon, server) = fake_daemon( r#"{"runtime_ok":true,"projects":[{"name":"svc","status":"running","repo_root":"/work/svc"},{"name":"nat","status":"unavailable","repo_root":"/work/nat","backend":"native"}]}"#, ); let src = LiveSource { - runner: CliRunner::new(&hack).with_env("HACK_HOME", &dir.to_string_lossy()), + runner: CliRunner::new(&hack) + .with_env("HACK_HOME", &dir.to_string_lossy()) + .with_env("HOME", &dir.to_string_lossy()) + .with_env("HACK_DAEMON_AUTOSTART_DISABLE", "1"), daemon: Some(daemon), }; let snap = src.snapshot().unwrap(); @@ -727,11 +813,17 @@ esac "#!/bin/sh\ncase \"$1\" in version) echo '{\"ok\":true,\"data\":{\"schema\":\"hack.version.v1\",\"version\":\"5.0.0\"}}' ;; projects) echo '{}' ;; esac\n", ); CliRunner::new(&hack) + .with_env("HOME", &dir.to_string_lossy()) + .with_env("HACK_HOME", &dir.to_string_lossy()) + .with_env("HACK_DAEMON_AUTOSTART_DISABLE", "1") .version(Duration::from_secs(60)) .unwrap(); let (daemon, server) = fake_daemon("{}"); let src = LiveSource { - runner: CliRunner::new(&hack), + runner: CliRunner::new(&hack) + .with_env("HOME", &dir.to_string_lossy()) + .with_env("HACK_HOME", &dir.to_string_lossy()) + .with_env("HACK_DAEMON_AUTOSTART_DISABLE", "1"), daemon: Some(daemon), }; let r = src.snapshot(); @@ -911,6 +1003,25 @@ esac let _ = std::fs::remove_dir_all(&b); } + #[test] + fn fixture_logs_honour_the_service_filter_and_native_note() { + let src = FixtureSource(Fixture::Running); + let args: Vec = ["logs", "--json", "--tail", "500", "worker"] + .iter() + .map(|s| (*s).to_owned()) + .collect(); + let mut feed = src.open_logs(Path::new("/work/x"), false, &args).unwrap(); + let events = feed.poll(1000).events; + assert!( + events + .iter() + .filter_map(|e| e.entry.as_ref()) + .all(|e| e.service.as_deref() == Some("worker")) + ); + assert!(feed.poll(1000).events.is_empty(), "delivered once"); + assert!(src.open_logs(Path::new("/work/x"), true, &args).is_err()); + } + #[test] fn lifecycle_actions_run_to_completion_without_an_app_deadline() { // A Start slower than any read timeout still completes; Open keeps diff --git a/apps/desktop/core/tests/dump_state.rs b/apps/desktop/core/tests/dump_state.rs index 1ecf7ca20..9eaf80abd 100644 --- a/apps/desktop/core/tests/dump_state.rs +++ b/apps/desktop/core/tests/dump_state.rs @@ -81,6 +81,9 @@ esac // Keep macOS first-launch scanning outside the fake daemon's accept // deadline, using the same bounded runner as the application. hack_desktop_core::runner::CliRunner::new(&hack) + .with_env("HOME", &home.to_string_lossy()) + .with_env("HACK_HOME", &home.to_string_lossy()) + .with_env("HACK_DAEMON_AUTOSTART_DISABLE", "1") .version(std::time::Duration::from_secs(60)) .unwrap(); let server = serve_snapshot(&home); diff --git a/apps/desktop/screenshots/fixtures-dark.png b/apps/desktop/screenshots/fixtures-dark.png index 51ce1fe01..7c9a49c2c 100644 Binary files a/apps/desktop/screenshots/fixtures-dark.png and b/apps/desktop/screenshots/fixtures-dark.png differ diff --git a/apps/desktop/screenshots/fixtures-light.png b/apps/desktop/screenshots/fixtures-light.png index 68520107b..9dba8c433 100644 Binary files a/apps/desktop/screenshots/fixtures-light.png and b/apps/desktop/screenshots/fixtures-light.png differ diff --git a/apps/desktop/screenshots/logs-dark.png b/apps/desktop/screenshots/logs-dark.png new file mode 100644 index 000000000..957614502 Binary files /dev/null and b/apps/desktop/screenshots/logs-dark.png differ diff --git a/apps/desktop/screenshots/logs-light.png b/apps/desktop/screenshots/logs-light.png new file mode 100644 index 000000000..28d1985c3 Binary files /dev/null and b/apps/desktop/screenshots/logs-light.png differ