src/mimir_lens/queued.rs
//! **Lens answers that are not computed on the request.**
//!
//! A lens over a room that is not built costs the room build (28 s on dwarf for
//! the whole workspace, 2026-10-16), and `world_impact` costs tens of seconds on
//! top. A gRPC door that folded them on the request held its caller that long,
//! and the Lenses tab's 20 s client bound cut it off — one slow lens blanked the
//! whole tab. So the `Mimir.*` doors ask [`ready`] first, exactly as
//! `CodeRoom.Answer` does ([`crate::code_room::reader::queued`]):
//!
//! * the answer computed from the room version the cache holds NOW is kept →
//! it is handed out at once;
//! * otherwise the answer is [`Prepared::Due`]: the server queues
//! [`compute_and_keep`] as a jera job and answers [`queued_body`] (`queued` /
//! `running`, with the job id), and the next ask finds the answer kept.
//!
//! An answer is kept per (room version, lens, canonical request) in the same
//! cache the pane answers use, so a consume that moves the room moves its
//! version and a kept answer is never handed out for a room it was not computed
//! from. A refusal that needs no room (not consumed, no lake, not a lens) and the
//! lens that reads no room ([`super::ROOMLESS`]) are answered at once.
use std::sync::Arc;
use serde_json::{Value, json};
pub use crate::code_room::reader::queued::Prepared;
use crate::code_room::Reach;
/// The key a lens answer is kept and queued under: `Mimir.<lens>@<build> <request>`.
/// The build's sha is in it: a deploy of new lens code never serves an answer
/// the old code kept in the lake for the same room state.
/// A wire zero (`limit 0`, `z_threshold 0.0` …) is the lens's default, so it is
/// left out like any value that says nothing.
pub fn key_of(lens: &str, params: &Value) -> String {
let mut p = params.clone();
if let Value::Object(o) = &mut p {
o.retain(|_, v| v.as_f64() != Some(0.0));
}
format!("Mimir.{lens}@{} {}", env!("NORNIR_BUILD_SHA"), crate::code_room::reader::queued::canonical(&p))
}
fn text(v: &Value) -> Arc<str> {
Arc::from(v.to_string())
}
fn repo_of(params: &Value) -> &str {
params.get("repo").and_then(Value::as_str).unwrap_or("")
}
/// Is `lens` answered on the queue? Every table-fed lens that reads a room,
/// and the SOURCE door too — a source fold is never inline in an RPC (the
/// jera law); its empty-repo refusal is cheap either way.
pub fn is_queued(lens: &str) -> bool {
(super::LENSES.contains(&lens) || super::SOURCE_LENSES.contains(&lens)) && !super::ROOMLESS.contains(&lens)
}
/// **Answer at once, or say it is due.** Ready only with the answer computed
/// from the very room the cache holds now, or with a refusal that costs nothing.
pub fn ready(lens: &str, reach: &Reach<'_>, params: &Value) -> Prepared {
if !is_queued(lens) {
return Prepared::Ready(text(&super::answer(lens, reach, params)));
}
#[cfg(feature = "code-room")]
{
let Ok(handle) = reach.code_room() else { return Prepared::Ready(text(&super::answer(lens, reach, params))) };
let key = key_of(lens, params);
match handle.peek(&super::scope_of(lens, repo_of(params))) {
// not consumed, lake unavailable …: the lens says so, at once
Err(_) => Prepared::Ready(text(&super::answer(lens, reach, params))),
Ok(None) => Prepared::Due { key },
// nothing consumed: the refusal costs nothing
Ok(Some(room)) if room.scope().repos.is_empty() => Prepared::Ready(text(&super::answer(lens, reach, params))),
Ok(Some(room)) => {
if let Some(body) = handle.answers().get(room.version(), &key) {
return Prepared::Ready(body);
}
// THE LAKE IS THE CACHE (the owner, 2026-10-18): a restart
// empties the memory cache, not the lake — an answer kept under
// THIS room version serves at once instead of folding again.
// The guard above is dropped: the lake read never holds the lock.
if let Some(body) = reach
.warehouse
.and_then(|wh| wh.read_lens_answer(room.state_key(), &key).ok().flatten())
{
let body: Arc<str> = Arc::from(body);
handle.answers().put(room.version(), key, body.clone());
return Prepared::Ready(body);
}
Prepared::Due { key }
}
}
}
#[cfg(not(feature = "code-room"))]
{
Prepared::Ready(text(&super::answer(lens, reach, params)))
}
}
/// **The job**: build the lens's room if it must be, compute the answer and
/// keep it under that room's version. `queued_ms` is how long the job waited
/// on the lane before it started (the queuer stamped the enqueue; 0 when
/// nothing queued it). Returns the answer's state and size.
pub fn compute_and_keep(lens: &str, reach: &Reach<'_>, params: &Value, queued_ms: u64) -> (String, usize) {
#[cfg(feature = "code-room")]
if let Ok(handle) = reach.code_room() {
let key = key_of(lens, params);
// the room FIRST: its version is the one the answer is kept under
if let Ok(opened) = handle.room_opened(&super::scope_of(lens, repo_of(params))) {
let room = opened.room;
if let Some(body) = handle.answers().get(room.version(), &key) {
return ("kept".to_string(), body.len());
}
// the lake too, like [`ready`]: a job that starts after a restart
// (or behind a newer job that already kept this state's answer)
// must not fold what the lake already holds — the chain drains a
// restart backlog in milliseconds, not one roster recompute
if let Some(body) = reach.warehouse.and_then(|wh| wh.read_lens_answer(room.state_key(), &key).ok().flatten()) {
let body: Arc<str> = Arc::from(body);
let len = body.len();
handle.answers().put(room.version(), key, body);
return ("kept".to_string(), len);
}
let t = std::time::Instant::now();
let mut answer = super::answer(lens, reach, params);
let computed_ms = t.elapsed().as_millis() as u64;
// where the time went, in the meta the pane already reads — the
// room open apart from the lens's own fold: a 15 s snapshot
// rebuild is `room_ms`, never `computed_ms`
if let Some(meta) = answer["lens_meta"].as_object_mut() {
meta.insert("computed_ms".into(), json!(computed_ms));
meta.insert("queued_ms".into(), json!(queued_ms));
meta.insert("room_ms".into(), json!(opened.ms));
meta.insert("room_cached".into(), json!(opened.cached));
}
eprintln!(
"mimir-lens: {lens} v{} computed in {computed_ms} ms (queued {queued_ms} ms, room {} ms {})",
room.version(),
opened.ms,
if opened.cached { "cached" } else { "built" }
);
let state = answer["lens_meta"]["state"].as_str().unwrap_or("").to_string();
let body = text(&answer);
let bytes = body.len();
handle.answers().put(room.version(), key.clone(), body.clone());
// THE LAKE IS THE CACHE: the same answer, kept durably under its
// room version. Best effort — the answer serves from memory either
// way, so a write that fails only costs the next restart a fold.
if let Some(wh) = reach.warehouse {
if let Err(e) = wh.append_lens_answer(room.state_key(), &key, &state, &body) {
eprintln!("lens_answers: keeping `{key}` in the lake failed (the answer still serves from memory): {e:#}");
}
}
return (state, bytes);
}
}
let answer = super::answer(lens, reach, params);
(answer["lens_meta"]["state"].as_str().unwrap_or("").to_string(), answer.to_string().len())
}
/// **Another lens's answer, kept or folded now** — for a lens that reads
/// another lens's findings (`reimplements_shared` reads the copy / twin
/// pairs): the answer kept for the room the cache holds, or the fold, kept so
/// the next ask of that lens (or of this one) is served from it.
pub fn kept_or_fold(lens: &str, reach: &Reach<'_>, params: &Value) -> Value {
let parse = |b: &str| serde_json::from_str(b).unwrap_or(Value::Null);
match ready(lens, reach, params) {
Prepared::Ready(b) => parse(&b),
Prepared::Due { .. } => {
compute_and_keep(lens, reach, params, 0);
match ready(lens, reach, params) {
Prepared::Ready(b) => parse(&b),
Prepared::Due { .. } => super::answer(lens, reach, params),
}
}
}
}
/// The envelope of a lens answer that is on the jera queue: the lens's own
/// shape (`kind`, `worst`, `feeds`, `evidence_note`, `lens_meta`), with
/// `lens_meta.state` `queued` or `running` and the job a person can look up —
/// never an empty finding list a pane could draw as "found nothing".
pub fn queued_body(lens: &str, repo: &str, job_id: &str, running: bool, ms: u64) -> Value {
let word = if running { "running" } else { "queued" };
json!({
"kind": lens,
"worst": [],
"feeds": [],
"evidence_note": format!("{lens}: {word} on jera — job {job_id} ({word} {}s). The answer is computed off the request and kept for this room version; ask again.", ms / 1000),
"lens_meta": {
"state": word,
"lens": lens,
"repo_scope": repo,
"job_id": job_id,
"ms": ms,
},
})
}
#[cfg(all(test, feature = "code-room"))]
mod tests {
use super::*;
use crate::code_room::Scope;
use crate::mimir_lens::room_feed::fixture::{self, SHOP, Without};
fn body(p: Prepared) -> Value {
match p {
Prepared::Ready(b) => serde_json::from_str(&b).unwrap(),
Prepared::Due { key } => panic!("due: {key}"),
}
}
/// A lens over a room that is not built is DUE (asking folds nothing);
/// once the job ran, the same question in any spelling is the kept
/// answer, the lens's own; another lens of that room is its own job.
#[test]
fn a_lens_is_due_until_its_job_ran_then_kept() {
let _g = fixture::serial();
crate::code_room::Room::set_cache_capacity(64);
let ws = fixture::graph_only(Without::default());
let reach = ws.reach();
let ask = json!({ "repo": "", "limit": 0 });
let Prepared::Due { key } = ready("zombie_features", &reach, &ask) else { panic!("nothing is built: due") };
assert_eq!(key, format!("Mimir.zombie_features@{} {{}}", env!("NORNIR_BUILD_SHA")));
assert!(ws.room.peek(&Scope::current()).unwrap().is_none(), "asking folded nothing");
let (state, bytes) = compute_and_keep("zombie_features", &reach, &json!({}), 0);
assert!(["complete", "partial", "empty_fact", "empty_gap"].contains(&state.as_str()), "{state}");
assert!(bytes > 50);
let kept = body(ready("zombie_features", &reach, &ask));
let direct = ws.ask("zombie_features", json!({}));
assert_eq!(kept["worst"], direct["worst"], "the kept answer is the lens's");
assert_eq!(kept["lens_meta"]["state"], direct["lens_meta"]["state"]);
assert!(matches!(ready("villa", &reach, &json!({})), Prepared::Due { .. }), "another lens is its own job");
assert_eq!(compute_and_keep("zombie_features", &reach, &json!({}), 0).0, "kept");
}
/// A refusal that needs no room, a name that is no lens and the lens that
/// reads no room answer at once — never a job.
#[test]
fn refusals_and_the_roomless_lens_answer_at_once() {
let _g = fixture::serial();
let ws = fixture::graph_only(Without::default());
let reach = ws.reach();
let nobody = body(ready("villa", &reach, &json!({ "repo": "nobody" })));
assert_eq!(nobody["lens_meta"]["state"], "not_consumed", "{nobody}");
let none = body(ready("villa", &Reach::none(), &json!({})));
assert_eq!(none["lens_meta"]["state"], "no_room");
let bad = body(ready("no_such_lens", &reach, &json!({})));
assert_eq!(bad["lens_meta"]["state"], "invalid_argument");
let a = body(ready("anomaly", &reach, &json!({ "repo": SHOP })));
assert_eq!(a["kind"], "anomaly");
}
/// **The lake IS the lens cache** (the owner, 2026-10-18: "ok so we use
/// iceberg as a cache for this lenses"). A computed answer is kept in the
/// lake's `lens_answers` table under the room version it was folded from
/// (one row per iceberg transaction of the log), so a server that lost its
/// memory — a restart, a deploy — hands the kept answer out instead of
/// folding again; and a version that moved never reads an old answer.
#[test]
fn a_kept_answer_survives_a_restart_through_the_lake() {
let _g = fixture::serial();
crate::code_room::Room::set_cache_capacity(64);
let ws = fixture::graph_only(Without::default());
let reach = ws.reach();
let ask = json!({ "repo": "", "limit": 0 });
assert!(matches!(ready("zombie_features", &reach, &ask), Prepared::Due { .. }));
let (state, _) = compute_and_keep("zombie_features", &reach, &json!({}), 0);
assert!(["complete", "partial", "empty_fact", "empty_gap"].contains(&state.as_str()), "{state}");
let kept = body(ready("zombie_features", &reach, &ask));
let state_key = ws.room.peek(&Scope::current()).unwrap().expect("the job built the room").state_key().to_string();
// THE RESTART: the memory cache is gone AND the process's rooms are
// gone — the rebuilt room carries a NEW process-local version over
// the SAME lake state. The lake row is keyed by the STATE, never the
// build counter (a fresh process's counter restarts at 1 and would
// collide with an older state's rows).
ws.room.answers().clear();
crate::code_room::Room::clear_cache();
let rebuilt = ws.room.room(&Scope::current()).unwrap();
assert_eq!(rebuilt.state_key(), state_key, "the same lake state IS the same key, whatever the build counter");
let Prepared::Ready(b) = ready("zombie_features", &reach, &ask) else {
panic!("a restart must not cost the fold again: the lake holds the answer for this lake state")
};
let again: Value = serde_json::from_str(&b).unwrap();
assert_eq!(again["worst"], kept["worst"], "the lake's answer is THE kept answer");
assert_eq!(again["lens_meta"]["state"], kept["lens_meta"]["state"]);
// …and the row is bound to ITS state: a moved lake never reads it.
let key = key_of("zombie_features", &json!({}));
let wh = reach.warehouse.unwrap();
assert!(wh.read_lens_answer(&state_key, &key).unwrap().is_some(), "the row is in the lake under its state key");
assert!(wh.read_lens_answer("0000000000000000", &key).unwrap().is_none(), "another state reads nothing");
}
/// A job that starts AFTER a restart must not fold what the lake already
/// holds: compute_and_keep checks the lake like [`ready`] does, and a hit
/// is "kept" — the chain drains a restart backlog in milliseconds instead
/// of recomputing the whole roster.
#[test]
fn a_job_after_a_restart_serves_from_the_lake_without_computing() {
let _g = fixture::serial();
crate::code_room::Room::set_cache_capacity(64);
let ws = fixture::graph_only(Without::default());
let reach = ws.reach();
let (state, _) = compute_and_keep("zombie_features", &reach, &json!({}), 0);
assert!(["complete", "partial", "empty_fact", "empty_gap"].contains(&state.as_str()), "{state}");
// THE RESTART: memory gone, rooms gone; the lake is not.
ws.room.answers().clear();
crate::code_room::Room::clear_cache();
let (state, bytes) = compute_and_keep("zombie_features", &reach, &json!({}), 0);
assert_eq!(state, "kept", "the lake answer serves; the lens is not folded again");
assert!(bytes > 0);
}
/// A computed answer says where its time went: `computed_ms` (the lens's
/// own fold) and `queued_ms` (the wait on the lane, as the queuer measured
/// it) ride in `lens_meta` next to `state` as numbers ≥ 0, with `room_ms` /
/// `room_cached` (the room open: a 15 s snapshot rebuild is not lens
/// compute), so the pane shows them without a UI change.
#[test]
fn a_computed_answer_carries_its_timing() {
let _g = fixture::serial();
crate::code_room::Room::set_cache_capacity(64);
let ws = fixture::graph_only(Without::default());
let reach = ws.reach();
let ask = json!({ "repo": "", "limit": 0 });
assert!(matches!(ready("zombie_features", &reach, &ask), Prepared::Due { .. }), "nothing is built: due");
compute_and_keep("zombie_features", &reach, &json!({}), 7);
let kept = body(ready("zombie_features", &reach, &ask));
let meta = &kept["lens_meta"];
assert!(meta["computed_ms"].as_u64().is_some(), "computed_ms is a number ≥ 0: {kept}");
assert_eq!(meta["queued_ms"].as_u64(), Some(7), "queued_ms is the wait the queuer measured: {kept}");
assert!(meta["room_ms"].as_u64().is_some(), "room_ms is a number ≥ 0: {kept}");
assert_eq!(meta["room_cached"].as_bool(), Some(false), "the job built this room, no cache served it: {kept}");
}
/// RED before: the key was `Mimir.<lens> <request>`, so a server with new
/// lens code served the old code's answers from the lake.
#[test]
fn a_kept_answer_is_keyed_by_the_build_that_computed_it() {
let k = key_of("semantic_twins", &json!({ "limit": 50 }));
assert!(k.starts_with(&format!("Mimir.semantic_twins@{} ", env!("NORNIR_BUILD_SHA"))), "{k}");
}
#[test]
fn the_queued_body_is_a_lens_envelope_that_is_not_a_finding() {
let q = queued_body("world_impact", "nornir", "j7", false, 2500);
assert_eq!((q["kind"].as_str(), q["lens_meta"]["state"].as_str(), q["lens_meta"]["job_id"].as_str()), (Some("world_impact"), Some("queued"), Some("j7")));
assert!(q["worst"].as_array().unwrap().is_empty());
assert!(q["evidence_note"].as_str().unwrap().contains("queued on jera — job j7 (queued 2s)"), "{q}");
assert_eq!(queued_body("villa", "", "j8", true, 0)["lens_meta"]["state"], "running");
}
}