src/code_room/reader/queued.rs
//! **Answers that are not computed on the request** (owner 2026-10-15: "the
//! ui should place stuff on jera queue if its not trivial stuff").
//!
//! The pane surfaces over a whole room — the crate graph, the function tier,
//! the call graph, the symbol listing, the 3-D room — cost a fold when the
//! room is not built (seconds) and up to a second and tens of MB of JSON when
//! it is. A server asks [`ready`] first: it answers AT ONCE from what is kept
//! (an answer computed from the room version the cache holds now), or says
//! the answer is [`Prepared::Due`] — the server then queues
//! [`compute_and_keep`] as a jera job and answers `queued` / `running`
//! ([`queued_body`]), and the next ask finds the answer kept.
//!
//! An answer is kept per (room version, surface, request): a consume that
//! moves the room moves its version ([`crate::code_room::Room::version`]),
//! so a kept answer is never handed out for a room it was not computed from.
//! Every other surface is answered on the request, as before ([`super::door`]).
use std::collections::BTreeMap;
use std::sync::Arc;
use serde_json::Value;
use super::Reach;
/// The surfaces a server computes on its jera `rooms` lane, never inside
/// the request.
pub const QUEUED_SURFACES: &[&str] = &["Viz.CrateGraph", "Viz.FunctionTier", "Viz.CallGraph", "Knowledge.Symbols", "Viz.CodeRoom3D", "Viz.CodeMap", "Viz.Binary", "Test.Affected", "Mimir.SystemBoundary"];
/// The request a pane sends for the skeleton (the crate graph, ring off) —
/// what a server computes right after it builds a room.
pub const SKELETON: (&str, &str) = ("Viz.CrateGraph", r#"{"universe":false}"#);
/// Is `surface` computed on the queue (see [`QUEUED_SURFACES`])?
pub fn is_queued(surface: &str) -> bool {
QUEUED_SURFACES.contains(&surface)
}
/// What [`ready`] found.
#[derive(Debug, Clone, PartialEq, Eq)]
pub enum Prepared {
/// The door's answer body (JSON text), to hand out as it is.
Ready(Arc<str>),
/// Must be computed: queue [`compute_and_keep`] under this key (the
/// surface and its request, canonical — repeats of one question merge).
Due { key: String },
}
/// The request in one canonical spelling: keys sorted, and a value that
/// says nothing (`null`, `false`, `""`, `[]`, `{}`) left out — `{}` and
/// `{"universe": false}` are one question.
pub fn canonical(params: &Value) -> String {
fn says_nothing(v: &Value) -> bool {
match v {
Value::Null | Value::Bool(false) => true,
Value::String(s) => s.is_empty(),
Value::Array(a) => a.is_empty(),
Value::Object(o) => o.is_empty(),
_ => false,
}
}
match params {
Value::Object(o) => {
let sorted: BTreeMap<&String, &Value> = o.iter().filter(|(_, v)| !says_nothing(v)).collect();
serde_json::to_string(&sorted).unwrap_or_default()
}
other => other.to_string(),
}
}
/// The body of an answer that is on the queue: not a measurement (no rows),
/// with the job a person can look up and how long it has waited or run.
pub fn queued_body(surface: &str, job_id: &str, running: bool, ms: u64) -> Value {
let (state, words) = if running { ("running", "running") } else { ("queued", "queued") };
serde_json::json!({
"surface": surface,
"state": state,
"job_id": job_id,
"ms": ms,
"note": format!("{words} on jera — job {job_id} ({words} {}s); ask again for the answer", ms / 1000),
})
}
/// Answers kept per (room version, surface, canonical request), least
/// recently used out first, bounded by count and by bytes.
#[derive(Default)]
pub struct AnswerCache {
entries: Vec<Kept>,
tick: u64,
}
struct Kept {
version: u64,
key: String,
body: Arc<str>,
used: u64,
}
impl AnswerCache {
/// Answers kept at most.
const CAP: usize = 48;
/// Bytes of kept answers at most (one function tier of a 21-repo
/// workspace is ~38 MB of JSON).
const CAP_BYTES: usize = 512 << 20;
pub(crate) fn get(&mut self, version: u64, key: &str) -> Option<Arc<str>> {
self.tick += 1;
let tick = self.tick;
self.entries.iter_mut().find(|e| e.version == version && e.key == key).map(|e| {
e.used = tick;
e.body.clone()
})
}
pub(crate) fn put(&mut self, version: u64, key: String, body: Arc<str>) {
self.tick += 1;
// an answer of an older room for the same question goes: it is never
// asked for again (a room's version only moves forward)
self.entries.retain(|e| e.key != key || e.version > version);
if self.entries.iter().any(|e| e.key == key && e.version > version) {
return;
}
self.entries.push(Kept { version, key, body, used: self.tick });
while self.entries.len() > Self::CAP || (self.entries.len() > 1 && self.entries.iter().map(|e| e.body.len()).sum::<usize>() > Self::CAP_BYTES) {
let oldest = self.entries.iter().enumerate().min_by_key(|(_, e)| e.used).map(|(i, _)| i).unwrap_or(0);
self.entries.remove(oldest);
}
}
/// Forget every kept answer of `surface` — for an answer that also reads
/// what does not move the room's version (`Viz.Binary` lists the artifacts
/// Build Thing runs file: a new build must show on the next ask).
pub fn forget(&mut self, surface: &str) {
let prefix = format!("{surface} ");
self.entries.retain(|e| !e.key.starts_with(&prefix));
}
/// Forget the kept answers of `surface` asked with one of `lods` — for a
/// level of detail that reads what does not move the room's version (a
/// traced run filed into the warehouse: `trace_runs` / `trace` / `profile`).
pub fn forget_lods(&mut self, surface: &str, lods: &[&str]) {
let prefix = format!("{surface} ");
let marks: Vec<String> = lods.iter().map(|l| format!("\"lod\":\"{l}\"")).collect();
self.entries.retain(|e| !(e.key.starts_with(&prefix) && marks.iter().any(|m| e.key.contains(m.as_str()))));
}
/// Forget EVERY kept answer — what a process restart does to this cache.
/// Tests use it to prove the lake half serves what memory lost.
pub fn clear(&mut self) {
self.entries.clear();
}
/// How many answers are kept (tests, status).
pub fn len(&self) -> usize {
self.entries.len()
}
pub fn is_empty(&self) -> bool {
self.entries.is_empty()
}
}
fn text(body: &Value) -> Arc<str> {
Arc::from(body.to_string())
}
/// The key of a question: surface + canonical request.
fn key_of(surface: &str, params: &Value) -> String {
format!("{surface} {}", canonical(params))
}
/// **Answer at once, or say it is due.** A surface that is not queued is
/// answered here ([`super::door`]); a refusal that needs no room is too. A
/// queued surface is [`Prepared::Ready`] only when its room is built AND an
/// answer computed from that very room is kept.
pub fn ready(surface: &str, reach: &Reach<'_>, params_json: &str) -> Prepared {
if !is_queued(surface) {
return Prepared::Ready(text(&super::door(surface, reach, params_json)));
}
#[cfg(feature = "code-room")]
{
let params: Value = if params_json.trim().is_empty() {
serde_json::json!({})
} else {
match serde_json::from_str(params_json) {
Ok(v) => v,
// the door words the refusal
Err(_) => return Prepared::Ready(text(&super::door(surface, reach, params_json))),
}
};
let Ok(handle) = reach.code_room() else { return Prepared::Ready(text(&super::door(surface, reach, params_json))) };
let key = key_of(surface, ¶ms);
match handle.peek(&scope_of(surface, ¶ms)) {
// not consumed, lake unavailable …: the door says so, at once
Err(_) => Prepared::Ready(text(&super::door(surface, reach, params_json))),
Ok(None) => Prepared::Due { key },
// nothing consumed: the door's refusal costs nothing
Ok(Some(room)) if room.scope().repos.is_empty() => Prepared::Ready(text(&super::door(surface, reach, params_json))),
Ok(Some(room)) => match handle.answers().get(room.version(), &key) {
Some(body) => Prepared::Ready(body),
None => Prepared::Due { key },
},
}
}
#[cfg(not(feature = "code-room"))]
{
Prepared::Ready(text(&super::door(surface, reach, params_json)))
}
}
/// **The job**: build the room if it must be, compute the answer and keep it
/// under the room's version. Returns the answer's state word and its size.
pub fn compute_and_keep(surface: &str, reach: &Reach<'_>, params_json: &str) -> (String, usize) {
#[cfg(feature = "code-room")]
if let (Ok(handle), Ok(params)) = (reach.code_room(), serde_json::from_str::<Value>(if params_json.trim().is_empty() { "{}" } else { params_json })) {
let key = key_of(surface, ¶ms);
// the room FIRST: its version is the one the answer is kept under
// (the door's own open serves the same snapshot, or a newer one)
if let Ok(room) = handle.room(&scope_of(surface, ¶ms)) {
if let Some(body) = handle.answers().get(room.version(), &key) {
return ("kept".to_string(), body.len());
}
let answer = super::door(surface, reach, params_json);
let body = text(&answer);
let out = (answer["state"].as_str().unwrap_or("").to_string(), body.len());
drop(answer);
handle.answers().put(room.version(), key, body);
return out;
}
}
let body = super::door(surface, reach, params_json);
(body["state"].as_str().unwrap_or("").to_string(), body.to_string().len())
}
/// The room a queued surface reads (the same scope its reader opens).
#[cfg(feature = "code-room")]
fn scope_of(surface: &str, params: &Value) -> crate::code_room::Scope {
use crate::code_room::Scope;
let repo = params.get("repo").and_then(Value::as_str).unwrap_or("");
if surface == "Viz.CrateGraph" {
let ring = params.get("universe").and_then(Value::as_bool).unwrap_or(false);
return if ring { Scope::current().with_universe(true) } else { Scope::current() };
}
if surface == "Mimir.SystemBoundary" {
return crate::mimir_lens::scope_of("system_boundary", repo);
}
if repo.is_empty() { Scope::current() } else { Scope::repo(repo, None) }
}
#[cfg(all(test, feature = "code-room"))]
mod tests {
use super::*;
use crate::code_room::Scope;
use crate::code_room::reader::fixture;
fn rows(body: &str, key: &str) -> Value {
serde_json::from_str::<Value>(body).unwrap()[key].clone()
}
/// The owner's rule: a whole-room pane is never computed on the request.
/// Asked before its room is built it is DUE (and asking builds nothing);
/// once the job ran, the same question — in any spelling — is the kept
/// answer, the door's own; another question of that room is due.
#[test]
fn a_queued_surface_is_due_until_the_job_ran_then_kept() {
// the process's room cache is shared by every test running beside this
// one: room enough that this test's room is not evicted between the
// job and the next ask (that is what "kept" depends on)
crate::code_room::Room::set_cache_capacity(64);
let ws = fixture::fresh_shop();
let reach = Reach::room_only(&ws.handle);
let Prepared::Due { key } = ready("Viz.CrateGraph", &reach, r#"{"universe":false}"#) else { panic!("nothing is built: due") };
assert_eq!(key, "Viz.CrateGraph {}");
assert!(ws.handle.peek(&Scope::current()).unwrap().is_none(), "asking folded nothing");
assert_eq!(ready("Viz.CrateGraph", &reach, "{}"), Prepared::Due { key: key.clone() }, "{{}} is the same question");
let (state, bytes) = compute_and_keep("Viz.CrateGraph", &reach, "{}");
assert_eq!(state, "complete");
assert!(bytes > 100, "{bytes}");
let Prepared::Ready(body) = ready("Viz.CrateGraph", &reach, r#"{"universe": false}"#) else { panic!("kept after the job") };
let direct = super::super::door("Viz.CrateGraph", &reach, "{}");
assert_eq!((rows(&body, "crates"), rows(&body, "edges")), (direct["crates"].clone(), direct["edges"].clone()), "the kept answer is the door's");
assert_eq!(serde_json::from_str::<Value>(&body).unwrap()["surface"], "Viz.CrateGraph");
assert!(matches!(ready("Viz.FunctionTier", &reach, "{}"), Prepared::Due { .. }), "another question of the room is its own job");
assert!(matches!(ready("Viz.CrateGraph", &reach, r#"{"universe":true}"#), Prepared::Due { .. }), "the ring-on room is another room");
assert_eq!(ws.handle.answers().len(), 1);
}
/// Forgetting a surface drops its kept answers and no other.
#[test]
fn forget_drops_one_surfaces_kept_answers() {
let mut c = AnswerCache::default();
c.put(1, "Viz.Binary {}".into(), Arc::from("a"));
c.put(1, "Viz.Binary {\"artifact\":\"x\"}".into(), Arc::from("b"));
c.put(1, "Viz.CrateGraph {}".into(), Arc::from("c"));
c.forget("Viz.Binary");
assert_eq!((c.len(), c.get(1, "Viz.CrateGraph {}").as_deref()), (1, Some("c")));
}
/// A filed trace forgets the trace lods' kept answers and keeps the map's others.
#[test]
fn forget_lods_drops_only_those_lods() {
let mut c = AnswerCache::default();
let k = |p: Value| key_of("Viz.CodeMap", &p);
c.put(1, k(serde_json::json!({ "lod": "trace_runs", "repo": "" })), Arc::from("a"));
c.put(1, k(serde_json::json!({ "lod": "trace", "run_id": "r" })), Arc::from("b"));
c.put(1, k(serde_json::json!({ "lod": "world" })), Arc::from("c"));
c.put(1, "Viz.Binary {\"lod\":\"trace\"}".into(), Arc::from("d"));
c.forget_lods("Viz.CodeMap", &["trace_runs", "trace", "profile"]);
assert_eq!(c.len(), 2);
assert_eq!(c.get(1, &k(serde_json::json!({ "lod": "world" }))).as_deref(), Some("c"));
}
/// A surface that is not on the queue, and a refusal that needs no
/// room, answer at once.
#[test]
fn the_rest_answers_at_once() {
let ws = fixture::fresh_shop();
let reach = Reach::room_only(&ws.handle);
let Prepared::Ready(map) = ready("Knowledge.Map", &reach, &format!(r#"{{"repo":"{}"}}"#, fixture::shop::SHOP)) else { panic!("Knowledge.Map is answered on the request") };
assert!(!map.is_empty());
let Prepared::Ready(bad) = ready("Viz.CrateGraph", &reach, "{not json") else { panic!("a request that is not JSON is refused at once") };
assert_eq!(rows(&bad, "state"), "invalid_argument");
let Prepared::Ready(none) = ready("Viz.CrateGraph", &Reach::none(), "{}") else { panic!("no room in reach: at once") };
assert_eq!(rows(&none, "state"), "no_room");
let Prepared::Ready(unconsumed) = ready("Viz.FunctionTier", &Reach::room_only(&fixture::unconsumed().handle), "{}") else { panic!("nothing consumed: at once") };
assert_eq!(rows(&unconsumed, "state"), "not_consumed");
}
/// A kept answer belongs to ONE room version: never handed out for
/// another, and an older room's answer never replaces a newer one.
#[test]
fn a_kept_answer_is_for_its_room_version_only() {
let mut c = AnswerCache::default();
c.put(7, "q".into(), Arc::from("seven"));
assert_eq!(c.get(7, "q").as_deref(), Some("seven"));
assert_eq!(c.get(8, "q"), None, "a newer room: due");
c.put(8, "q".into(), Arc::from("eight"));
assert_eq!((c.get(7, "q"), c.get(8, "q").as_deref()), (None, Some("eight")), "the older room's answer went");
c.put(7, "q".into(), Arc::from("late seven"));
assert_eq!((c.get(7, "q"), c.get(8, "q").as_deref(), c.len()), (None, Some("eight"), 1), "a late older answer is dropped");
}
#[test]
fn the_request_has_one_spelling() {
let v = |s: &str| serde_json::from_str::<Value>(s).unwrap();
assert_eq!(canonical(&v("{}")), canonical(&v(r#"{"universe":false,"repo":"","kinds":[]}"#)));
assert_eq!(canonical(&v(r#"{"repo":"a","limit":5}"#)), canonical(&v(r#"{"limit":5,"repo":"a"}"#)));
assert_ne!(canonical(&v(r#"{"repo":"a"}"#)), canonical(&v(r#"{"repo":"b"}"#)));
let q = queued_body("Viz.CrateGraph", "j1", true, 4200);
assert_eq!((q["state"].as_str(), q["job_id"].as_str()), (Some("running"), Some("j1")));
assert!(q["note"].as_str().unwrap().contains("running on jera — job j1 (running 4s)"), "{q}");
}
}