1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
//! `job.tail.<pc_id>` request/reply handler.
//!
//! On-demand only — the backend (on the SPA's behalf) sends a
//! [`JobTailRequest`] with a `result_id` and the agent replies with
//! the current ring-buffer tail of that in-flight job's stdout/stderr
//! from the in-memory [`crate::live_tail`] registry. We never push
//! live output upstream on our own; the SPA polls this while a job is
//! running (same shape as `logs.fetch.<pc_id>`).
//!
//! When the agent holds no live buffer for the `result_id` (the run
//! never started here, or it finished and was evicted past the grace
//! window), the reply carries `found = false` so the backend falls
//! back to the persisted `execution_results` row.
use futures::StreamExt;
use kanade_shared::subject;
use kanade_shared::wire::{JobTailReply, JobTailRequest};
use tracing::{info, warn};
/// Subscribe to `job.tail.<pc_id>` and serve every incoming request.
/// Each request is a cheap in-memory lookup, so we serve them inline
/// on this task — no fan-out spawning needed.
pub async fn serve(client: async_nats::Client, pc_id: String, tracker: crate::staleness::Tracker) {
let subject = subject::job_tail(&pc_id);
// Outer reconnect loop mirrors `logs::serve` — a transient
// subscribe failure or a closed subscription must not kill the
// handler permanently.
loop {
let mut sub =
crate::nats_retry::wait_for_subscribe(&client, &tracker, &subject, "job_tail").await;
info!(subject = %subject, "job.tail handler ready");
while let Some(msg) = sub.next().await {
let Some(reply) = msg.reply.clone() else {
warn!("job.tail without reply subject — caller must use request/reply");
continue;
};
let req: JobTailRequest = match serde_json::from_slice(&msg.payload) {
Ok(r) => r,
Err(e) => {
// A malformed payload still gets a safe `found=false`
// reply (default result_id ""), but warn so a broken
// caller leaves a trace — consistent with the other
// deserialize sites in the agent.
warn!(error = %e, "job.tail: deserialize JobTailRequest; replying not-found");
JobTailRequest::default()
}
};
let body = build_reply(&req);
let payload = match serde_json::to_vec(&body) {
Ok(p) => p,
Err(e) => {
warn!(error = %e, "encode JobTailReply");
continue;
}
};
if let Err(e) = client.publish(reply, payload.into()).await {
warn!(error = %e, "publish job.tail reply");
}
}
warn!(subject = %subject, "job.tail subscription closed; reopening");
crate::nats_retry::reopen_pause().await;
}
}
/// Build the reply for one request: look up the live buffer and
/// snapshot it, or report `found = false`.
fn build_reply(req: &JobTailRequest) -> JobTailReply {
match crate::live_tail::get(&req.result_id) {
Some(tail) => {
let snap = tail.snapshot();
JobTailReply {
found: true,
running: snap.running,
stdout: snap.stdout,
stderr: snap.stderr,
stdout_truncated: snap.stdout_truncated,
stderr_truncated: snap.stderr_truncated,
}
}
None => JobTailReply {
found: false,
..Default::default()
},
}
}
#[cfg(test)]
mod tests {
use super::*;
#[test]
fn unknown_result_id_is_not_found() {
let reply = build_reply(&JobTailRequest {
result_id: "no-such-job".into(),
});
assert!(!reply.found);
assert!(!reply.running);
}
#[test]
fn live_result_id_is_found_and_running() {
let handle = crate::live_tail::register("job-tail-serve-test");
handle.tail().push_stdout(b"progress 50%");
let reply = build_reply(&JobTailRequest {
result_id: "job-tail-serve-test".into(),
});
assert!(reply.found);
assert!(reply.running);
assert_eq!(reply.stdout, "progress 50%");
}
}