{{ title }}
{{ cli }}Warning: {{ warning }}
{% when None %}{% endmatch %} {% for section in inbox.sections %} -
+
, tokio::task::JoinHandle<()>) {
+ let (sender, changes) = watch::channel(0);
+ let observer = tokio::spawn(async move {
+ let mut previous = initial;
+ loop {
+ tokio::time::sleep(Duration::from_secs(1)).await;
+ match super::render_inbox("thirdshift status").await {
+ Ok(html) if html != previous => {
+ previous = html;
+ sender.send_modify(|revision| *revision += 1);
+ }
+ Ok(_) => {}
+ Err(error) => eprintln!("thirdshift Console: can't observe Inbox: {error:#}"),
+ }
+ }
+ });
+ (changes, observer)
+}
+
+pub(super) async fn events(State(changes): State>) -> Response {
+ // WatchStream emits immediately, so every reconnect also catches up.
+ let stream = WatchStream::new(changes)
+ .map(|_| Ok::<_, Infallible>(Event::default().event("commands").data("changed")));
+ (
+ [("x-accel-buffering", "no"), ("cache-control", "no-store")],
+ Sse::new(stream).keep_alive(KeepAlive::default()),
+ )
+ .into_response()
+}
diff --git a/src/console/row.html b/src/console/row.html
index 6cb15eee..7551504c 100644
--- a/src/console/row.html
+++ b/src/console/row.html
@@ -1,5 +1,5 @@
-
-
+ ,
@@ -22,6 +23,11 @@ pub(crate) struct Section {
}
pub(crate) struct Row {
+ pub id: String,
+ pub started_ms: i64,
+ pub live: bool,
+ pub logging: bool,
+ pub log_growth: String,
pub selector: String,
pub logs_url: String,
pub subject: String,
@@ -133,11 +139,19 @@ pub(crate) fn archive(
fn entry_row(entry: &Entry, now: DateTime, root: &std::path::Path) -> Result {
match entry {
- Entry::Record(command) => row(command, now, root),
+ Entry::Record(command) => row(command, now, None, root),
Entry::History(command) => {
let (subject, title) =
subject(&command.repository, &command.kind, command.issue.as_ref());
Ok(Row {
+ id: identity(&format!(
+ "history:{}:{}:{}",
+ command.repository, command.kind, command.started_at
+ )),
+ started_ms: timestamp(&command.started_at),
+ live: false,
+ logging: false,
+ log_growth: String::new(),
selector: Observation::History(command).selector().unwrap_or_default(),
logs_url: command
.command_log
@@ -247,10 +261,35 @@ fn subject(repository: &str, name: &str, issue: Option<&Issue>) -> (String, Stri
}
}
-fn row(command: &Command, now: DateTime, root: &std::path::Path) -> Result {
+fn row(
+ command: &Command,
+ now: DateTime,
+ outer_log: Option<&Path>,
+ root: &Path,
+) -> Result {
let facts = &command.record;
+ let log = outer_log.unwrap_or(&command.command_log);
+ let live = command.state == "live";
let (subject, title) = subject(&facts.repository, &facts.name, facts.issue.as_ref());
Ok(Row {
+ // A Pass's dispatched Runs share its storage path. Their immutable
+ // Command facts distinguish them without tying identity to ordering.
+ id: identity(&format!(
+ "{}\0{}\0{}\0{}\0{}",
+ command.record_path.to_string_lossy(),
+ facts.kind,
+ facts.name,
+ facts.issue.as_ref().map_or("", |issue| issue.url.as_str()),
+ facts.started_at
+ )),
+ started_ms: timestamp(&facts.started_at),
+ live,
+ logging: live && !matches!(command.health, Some(Health::Silent { .. })),
+ log_growth: format!(
+ "{}:{}",
+ log_length(Some(log)),
+ log_length(facts.session_log.as_deref())
+ ),
selector: Observation::Record(command).selector().unwrap_or_default(),
logs_url: logs_url(&command.record_path, root),
subject,
@@ -276,7 +315,7 @@ fn row(command: &Command, now: DateTime, root: &std::path::Path) -> Resul
.dispatched
.iter()
.chain(&command.children)
- .map(|child| row(child, now, root))
+ .map(|child| row(child, now, Some(log), root))
.collect::>()?,
findings: facts.security_findings.clone(),
sessions: command.sessions()?,
@@ -289,6 +328,26 @@ fn logs_url(path: &std::path::Path, root: &std::path::Path) -> String {
.unwrap_or_else(|_| "/?page=logs".into())
}
+fn log_length(path: Option<&Path>) -> u64 {
+ path.and_then(|path| std::fs::metadata(path).ok())
+ .map_or(0, |file| file.len())
+}
+
+fn timestamp(value: &str) -> i64 {
+ DateTime::parse_from_rfc3339(value).map_or(0, |time| time.timestamp_millis())
+}
+
+fn identity(value: &str) -> String {
+ // Encode the exact storage identity: issue numbers and timestamps alone
+ // can collide, and DOM IDs cannot contain whitespace.
+ let encoded: String = value
+ .as_bytes()
+ .iter()
+ .map(|byte| format!("{byte:02x}"))
+ .collect();
+ format!("command-{encoded}")
+}
+
fn elapsed(time: &str, now: DateTime) -> String {
DateTime::parse_from_rfc3339(time)
.map(|time| age(now.signed_duration_since(time).num_seconds().max(0)))
diff --git a/tests/console.rs b/tests/console.rs
index 4fec42f6..d194e76b 100644
--- a/tests/console.rs
+++ b/tests/console.rs
@@ -1,4 +1,6 @@
//! The Console's public seam: the binary, fixture records and HTTP.
+mod support;
+
use std::fs;
use std::io::{BufRead, BufReader};
use std::path::Path;
@@ -16,6 +18,21 @@ struct Console {
reader: Option>,
}
+struct HeldFixture {
+ run: Option,
+ release: std::path::PathBuf,
+}
+
+impl Drop for HeldFixture {
+ fn drop(&mut self) {
+ // Release and reap the fixture even when a regression assertion fails.
+ let _ = fs::write(&self.release, "");
+ if let Some(run) = self.run.take() {
+ run.finish();
+ }
+ }
+}
+
impl Console {
fn start(binary: &Path, home: &Path) -> Self {
let mut child = Command::new(binary)
@@ -104,6 +121,119 @@ fn write_record(home: &Path, number: u64, record: &Value) {
.unwrap();
}
+#[test]
+fn command_changes_push_events_over_http() {
+ let home = tempfile::tempdir().unwrap();
+ configure(home.path());
+ let console = Console::start(Path::new(env!("CARGO_BIN_EXE_thirdshift")), home.path());
+ let response = console.get("/events");
+ assert_eq!(response.status(), 200);
+ assert_eq!(response.headers()["content-type"], "text/event-stream");
+ assert_eq!(response.headers()["x-accel-buffering"], "no");
+ let mut stream = BufReader::new(response);
+ fn changed(stream: &mut impl BufRead) {
+ let mut event = String::new();
+ loop {
+ let mut line = String::new();
+ assert!(stream.read_line(&mut line).unwrap() > 0, "stream ended");
+ if line.trim().is_empty() && event.contains("event: commands") {
+ assert!(event.contains("data: changed"), "{event}");
+ return;
+ }
+ if line.trim().is_empty() {
+ event.clear();
+ } else {
+ event.push_str(&line);
+ }
+ }
+ }
+ changed(&mut stream); // Reconnects always catch up, even if an event was missed.
+ let mut fixture = record(home.path(), 81, None, 0);
+ changed(&mut stream);
+ assert!(console.get("/").text().unwrap().contains("acme/widgets#81"));
+ fixture["stages"][0]["name"] = json!("CI watch");
+ write_record(home.path(), 81, &fixture);
+ changed(&mut stream);
+ assert!(
+ console
+ .get("/")
+ .text()
+ .unwrap()
+ .contains("died at CI watch")
+ );
+ fixture["outcome"] = json!("merged");
+ fixture["ended_at"] = json!(chrono::Local::now().to_rfc3339());
+ write_record(home.path(), 81, &fixture);
+ changed(&mut stream);
+ assert!(console.get("/").text().unwrap().contains(">merged"));
+}
+
+#[test]
+fn heartbeat_observes_log_lengths_and_is_flat_for_silent_runs() {
+ let scenario = support::Scenario::new();
+ scenario.user_config_is("[console]\nport = 0\n");
+ scenario.agent_does(&format!(
+ "touch '{}'\nwhile ! test -f '{}'; do sleep 0.02; done\nexit 1\n",
+ scenario.path("started").display(),
+ scenario.path("release").display()
+ ));
+ let _held = HeldFixture {
+ run: Some(scenario.run_until(&[&scenario.issue_url(7)], &[], "started")),
+ release: scenario.path("release"),
+ };
+ let dir = scenario.path("home/.thirdshift/logs/acme/widgets/commands/issue");
+ let path = fs::read_dir(dir)
+ .unwrap()
+ .map(|entry| entry.unwrap().path())
+ .find(|path| path.to_string_lossy().ends_with(".command.json"))
+ .unwrap();
+ let mut fixture: Value = serde_json::from_slice(&fs::read(&path).unwrap()).unwrap();
+ let session = std::path::PathBuf::from(fixture["session_log"].as_str().unwrap());
+ let log = path.with_file_name(
+ path.file_name()
+ .unwrap()
+ .to_str()
+ .unwrap()
+ .replace(".command.json", ".log"),
+ );
+ let began = (chrono::Local::now() - chrono::Duration::minutes(20)).to_rfc3339();
+ fixture["stages"] = json!([{"name": "implement", "began_at": began, "session_log": session}]);
+ fs::write(&path, serde_json::to_vec(&fixture).unwrap()).unwrap();
+ fs::write(&log, "a").unwrap();
+ fs::write(&session, "ab").unwrap();
+ let console = Console::start(
+ Path::new(env!("CARGO_BIN_EXE_thirdshift")),
+ &scenario.path("home"),
+ );
+ let html = console.get("/").text().unwrap();
+ assert!(html.contains("data-log-growth=\"1:2\""), "{html}");
+ assert!(html.contains("data-logging=\"true\""));
+ fs::write(&session, "abcde").unwrap();
+ assert!(
+ console
+ .get("/")
+ .text()
+ .unwrap()
+ .contains("data-log-growth=\"1:5\"")
+ );
+ for file in [&log, &session] {
+ fs::File::options()
+ .write(true)
+ .open(file)
+ .unwrap()
+ .set_times(
+ fs::FileTimes::new()
+ .set_modified(std::time::SystemTime::now() - Duration::from_secs(20 * 60)),
+ )
+ .unwrap();
+ }
+ let html = console.get("/").text().unwrap();
+ assert!(html.contains("data-state=\"Silent\""));
+ assert!(html.contains("data-logging=\"false\""));
+ assert!(html.contains("data-log-growth=\"1:5\""));
+ drop(console);
+}
+
#[test]
fn history_keeps_every_ended_command_in_time_order_and_filters() {
let home = tempfile::tempdir().unwrap();
@@ -988,6 +1118,7 @@ fn replacing_the_binary_executes_the_replacement_with_the_original_arguments() {
assert!(copied.success());
let console = Console::start(&binary, home.path());
assert_eq!(console.get("/").status(), 200);
+ let _open_stream = console.get("/events");
// A reader must not keep graceful shutdown waiting on an infinite stream.
#[cfg(target_os = "linux")]
let _stream = {
{{ section.title }}{{ section.rows.len() }}
{% if section.rows.is_empty() && section.references.is_empty() %}Nothing here.
{% endif %} {# Only the rendered, auto-escaped RowTemplate is trusted as HTML. #} diff --git a/src/console/live.rs b/src/console/live.rs new file mode 100644 index 00000000..618f4376 --- /dev/null +++ b/src/console/live.rs @@ -0,0 +1,40 @@ +//! One reader per Console; slow browsers only need the latest change notice. +use std::{convert::Infallible, time::Duration}; + +use axum::extract::State; +use axum::response::{ + IntoResponse, Response, Sse, + sse::{Event, KeepAlive}, +}; +use tokio::sync::watch; +use tokio_stream::{StreamExt, wrappers::WatchStream}; + +pub(super) fn observe(initial: String) -> (watch::Receiver{{ row.state }} {{ row.subject }} {{ row.title }}{{ row.age }}
+{% if row.live %}{% endif %}{{ row.state }} {{ row.subject }} {{ row.title }}New since you last looked{{ row.age }}
{{ row.now_line }}
Run and Session logs {% if !row.harness.is_empty() %}{{ row.harness }}
{% endif %} diff --git a/src/logs/status/inbox.rs b/src/logs/status/inbox.rs index 37c9e50d..fe51fd58 100644 --- a/src/logs/status/inbox.rs +++ b/src/logs/status/inbox.rs @@ -1,11 +1,12 @@ //! Owned Inbox presentation for the Console, from the same observations as status. use anyhow::Result; use chrono::{DateTime, Local}; +use std::path::Path; use super::{age, stage_text, ticket_text}; use crate::config::UserConfig; use crate::logs::command_record::Issue; -use crate::logs::reading::{Command, Entry, Group, Observation, Snapshot}; +use crate::logs::reading::{Command, Entry, Group, Health, Observation, Snapshot}; pub(crate) struct Inbox { pub sections: Vec