src/code_room/mod.rs
//! # The code room of one workspace — the handle every front door uses
//!
//! Mímir v2 (`PLAN-MIMIR-V2-2026-10-12.md`, wiring map `.nornir/mimir-v2-land.md`
//! §6.1). One event log per workspace, one consumer writing it, rooms opened
//! from it:
//!
//! ```text
//! <builds>/mimir_v2/lake/ skade lake, namespace mimir_v2 (the EventLog)
//! <builds>/mimir_v2/clones/<member>/ the consumer's own clone (RepoSpec.worktree)
//! <builds>/mimir_v2/target-shared/ THE nightly target dir of every member (DriverDumper.target_dir;
//! mimir-mirror/<member>/ per member, mimir-ring/ shared)
//! <builds>/mimir_v2/target/<member>/ per member, only with NORNIR_MIMIR_TARGET=per-member
//! <builds>/mimir_v2/out/<member>/ driver IPC dumps (DriverDumper.out_root)
//! <builds>/mimir_v2/std/<toolchain>/ the toolchain's std dump (StdDumper.root)
//! <builds>/mimir_v2/artifacts/<member>/<sha>/<profile>/<bin>
//! linked artifacts of a commit, put there by
//! whoever built it (the binary hook READS)
//! (no directory) Build Thing products: the binary tables under `@build-thing`
//! ([`products`], builder 2)
//! ```
//!
//! `<builds>` is the workspace's warehouse root (`<registry_root>/<ws>/builds`
//! on a server). The lake is its OWN skade catalog (decision D1), never
//! `builds/catalog.redb`.
//!
//! ## The seam
//!
//! Front doors (gRPC handlers, MCP tools, the CLI) never open a lake and never
//! name a room query. They hand a [`Reach`] — what THIS door can reach: the
//! code room, the warehouse — to one of two dispatch points:
//!
//! * [`crate::mimir_lens::answer`] — the 18 Mímir lenses (lane B);
//! * [`reader::answer`] — every other reader of the code graph (lane C).
//!
//! Both take the surface's name and its request as JSON, and answer JSON. A
//! surface that cannot answer says WHY as a typed [`Refusal`] ([`State`]),
//! never as an empty result.
//!
//! ## The `code-room` feature
//!
//! The engine (GraphBLAS, through `mimir-room`) is linked only with the
//! `code-room` feature — on in `server`, off in the thin atoms. Without it this
//! module still compiles: [`CodeRoom`] exists so no signature changes, and
//! every dispatch answers [`State::NotLinked`].
#[cfg(feature = "code-room")]
pub mod backfill;
#[cfg(feature = "code-room")]
pub mod consumer_loop;
#[cfg(feature = "mimir-container")]
pub mod container;
#[cfg(feature = "code-room")]
pub mod doctor;
#[cfg(feature = "code-room")]
pub mod driver;
#[cfg(feature = "code-room")]
pub mod feeds;
#[cfg(feature = "code-room")]
mod lake_lane;
#[cfg(feature = "code-room")]
pub mod preflight;
#[cfg(feature = "code-room")]
pub mod products;
pub mod reader;
#[cfg(feature = "code-room")]
pub mod status;
#[cfg(feature = "code-room")]
pub mod units;
/// The room types a front door names without naming the engine crate.
#[cfg(feature = "code-room")]
pub use mimir_room::{Room, Scope};
/// What an uncommitted tree would change ([`CodeRoom::dry_run`]).
#[cfg(feature = "code-room")]
pub use mimir_ingest::{DryDelta, KeyDelta};
use std::fmt;
use std::path::{Path, PathBuf};
use crate::warehouse::iceberg::IcebergWarehouse;
/// The directory under a workspace's warehouse root that holds its code room.
pub const DIR: &str = "mimir_v2";
/// Where the pieces of one workspace's code room live.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct Layout {
base: PathBuf,
}
impl Layout {
/// The layout under a workspace's warehouse root (`<builds>`).
pub fn under(builds: &Path) -> Self {
Self { base: builds.join(DIR) }
}
/// `<builds>/mimir_v2`.
pub fn base(&self) -> &Path {
&self.base
}
/// The skade lake of the event log.
pub fn lake(&self) -> PathBuf {
self.base.join("lake")
}
/// Where the consumer's own clones live (`clones/<member>`).
pub fn clones(&self) -> PathBuf {
self.base.join("clones")
}
/// The consumer's own clone of `member`.
pub fn clone_of(&self, member: &str) -> PathBuf {
self.base.join("clones").join(member)
}
/// The nightly target dir of `member` alone (`NORNIR_MIMIR_TARGET=per-member`).
pub fn target_of(&self, member: &str) -> PathBuf {
self.base.join("target").join(member)
}
/// The ONE nightly target dir every member dumps into: cargo's own unit
/// hash keys it, so a dependency unit (same package source, features,
/// profile, flags) is compiled once for the whole workspace, not once per
/// member that needs it (2026-10-15: 21 986 units over 20 repos, 7 128
/// distinct). Features are never unified across members — another
/// feature set is another unit.
pub fn shared_target(&self) -> PathBuf {
self.base.join("target-shared")
}
/// The target dir the driver of `member` uses: the shared one, unless
/// `NORNIR_MIMIR_TARGET=per-member` asks for one per member.
pub fn driver_target(&self, member: &str) -> PathBuf {
match std::env::var("NORNIR_MIMIR_TARGET").as_deref().map(str::trim) {
Ok("per-member") => self.target_of(member),
_ => self.shared_target(),
}
}
/// Where the driver's per-commit dumps of `member` land.
pub fn out_of(&self, member: &str) -> PathBuf {
self.base.join("out").join(member)
}
/// The state of the dry runs over `member`'s uncommitted trees: a driver
/// target dir + out dir of their own (never the consumer's).
pub fn dry_of(&self, member: &str) -> PathBuf {
self.base.join("dry").join(member)
}
/// Where the toolchains' std dumps live (`<std>/<toolchain>/{probe,target,out}`).
pub fn std_root(&self) -> PathBuf {
self.base.join("std")
}
/// Where the linked artifacts of `member` at commit `sha` are looked for:
/// a cargo target dir of that commit (`<here>/<profile>/<bin>`, optionally
/// `<profile>/deps/`). Nothing here builds into it — whoever built the
/// commit puts (or points) its output there.
pub fn artifacts_of(&self, member: &str, sha: &str) -> PathBuf {
self.base.join("artifacts").join(member).join(sha)
}
}
/// Why a surface over the code room has no answer. Every refusal carries one.
#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)]
pub enum State {
/// The surface's v2 reader / producer is not written yet.
NotWired,
/// This binary was built without the `code-room` feature (a thin atom):
/// the engine is not linked, the server answers over gRPC.
NotLinked,
/// This front door has no code room in reach (no workspace selected).
NoRoom,
/// The lake cannot be opened here — another process (the server) holds it.
LakeUnavailable,
/// The repo has no consumer offset: nothing of it was consumed yet.
NotConsumed,
/// The pinned nightly driver is not installed on this box.
DriverNotAvailable,
/// The repo is consumed, but no linked artifact of its commit was where
/// the binary hook looks (the note names the directory): a surface over
/// the BINARY half has nothing to read. Nothing here builds one. A binary
/// lens says it too when no Build Thing run has filed an artifact (builder
/// 2) for what it was asked: the note says to build it.
NoArtifact,
/// The request is not answerable as asked (unknown repo, bad argument).
InvalidArgument,
/// The room itself failed.
Internal,
/// A vector lens: no function has a vector, and the build that ran the
/// `embeddings` feed has no embedder compiled (the feed says `skipped`).
NoEmbedder,
/// A vector lens: no function has a vector — the `embeddings` feed has
/// not written them (not run yet, or it failed; the note says which).
VectorsMissing,
}
impl State {
pub const ALL: [State; 11] = [
State::NotWired,
State::NotLinked,
State::NoRoom,
State::LakeUnavailable,
State::NotConsumed,
State::DriverNotAvailable,
State::NoArtifact,
State::InvalidArgument,
State::Internal,
State::NoEmbedder,
State::VectorsMissing,
];
/// The refusal state a machine name spells, if it is one (a computed
/// answer's state is none of these).
pub fn parse(name: &str) -> Option<State> {
Self::ALL.into_iter().find(|s| s.as_str() == name)
}
/// The machine name (`lens_meta.state`, job detail, CLI line).
pub fn as_str(self) -> &'static str {
match self {
State::NotWired => "not_wired",
State::NotLinked => "not_linked",
State::NoRoom => "no_room",
State::LakeUnavailable => "lake_unavailable",
State::NotConsumed => "not_consumed",
State::DriverNotAvailable => "driver_not_available",
State::NoArtifact => "no_artifact",
State::InvalidArgument => "invalid_argument",
State::Internal => "internal",
State::NoEmbedder => "no_embedder",
State::VectorsMissing => "vectors_missing",
}
}
}
impl fmt::Display for State {
fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
f.write_str(self.as_str())
}
}
/// A surface's typed "no answer": the state and the sentence a person reads.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct Refusal {
pub state: State,
pub note: String,
}
impl Refusal {
pub fn new(state: State, note: impl Into<String>) -> Self {
Self { state, note: note.into() }
}
/// The surface's reader is not written yet (the note every v1 stub printed).
pub fn not_wired() -> Self {
Self::new(State::NotWired, crate::mimir_lens::NOT_WIRED_NOTE)
}
/// The JSON form (`{"state": …, "note": …}`), for envelopes that carry it.
pub fn to_json(&self) -> serde_json::Value {
serde_json::json!({ "state": self.state.as_str(), "note": self.note })
}
}
impl fmt::Display for Refusal {
fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
f.write_str(&self.note)
}
}
impl std::error::Error for Refusal {}
/// The code room of ONE workspace. Cheap to construct (no IO): the lake is
/// opened on first use and kept for the life of the handle.
pub struct CodeRoom {
workspace: String,
layout: Layout,
#[cfg(feature = "code-room")]
engine: engine::Engine,
}
impl fmt::Debug for CodeRoom {
fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
f.debug_struct("CodeRoom").field("workspace", &self.workspace).field("base", &self.layout.base).finish()
}
}
impl CodeRoom {
/// The code room of `workspace`, under its warehouse root `builds`.
pub fn new(workspace: &str, builds: &Path) -> Self {
Self {
workspace: workspace.to_string(),
layout: Layout::under(builds),
#[cfg(feature = "code-room")]
engine: engine::Engine::default(),
}
}
pub fn workspace(&self) -> &str {
&self.workspace
}
pub fn layout(&self) -> &Layout {
&self.layout
}
/// Was this binary built with the engine?
pub const fn linked() -> bool {
cfg!(feature = "code-room")
}
}
/// What a front door can reach, handed to the dispatch points. Borrowed: a
/// door builds it per request from the handles it already holds.
#[derive(Clone, Copy)]
pub struct Reach<'a> {
/// The code room of the workspace the request selected.
pub room: Option<&'a CodeRoom>,
/// That workspace's warehouse (the non-graph feeds: heat, tests, sbom …).
pub warehouse: Option<&'a IcebergWarehouse>,
}
impl<'a> Reach<'a> {
pub fn new(room: &'a CodeRoom, warehouse: &'a IcebergWarehouse) -> Self {
Self { room: Some(room), warehouse: Some(warehouse) }
}
/// A door with only a code room (the CLI's fat mode before it opens a warehouse).
pub fn room_only(room: &'a CodeRoom) -> Self {
Self { room: Some(room), warehouse: None }
}
/// A door that reaches nothing (a thin client with no server selected).
pub fn none() -> Self {
Self { room: None, warehouse: None }
}
/// The code room, or the typed refusal a reader returns when there is none.
pub fn code_room(&self) -> Result<&'a CodeRoom, Refusal> {
self.room.ok_or_else(|| Refusal::new(State::NoRoom, "no code room in reach of this front door (no workspace selected)"))
}
}
#[cfg(feature = "code-room")]
mod engine {
//! The linked half: the lake, the log, rooms.
//!
//! The handle model is `mimir_ingest::EventLog`'s (wiring map R13): ONE
//! open lake per process, shared by every clone of the log handle — one
//! writer (the consumer), any number of readers, each at one committed
//! release. A second PROCESS cannot open a held lake; it answers
//! [`State::LakeUnavailable`] and reads through the holder's RPCs.
use std::sync::Mutex;
use mimir_ingest::{EventLog, LakeHeld, LakeSchemaMismatch};
use mimir_room::{Opened, Room, Scope, ScopeError};
use super::{CodeRoom, Refusal, State};
/// The lazily opened log, and the consumer over it.
#[derive(Default)]
pub(super) struct Engine {
log: Mutex<Option<EventLog>>,
pub(super) consumer: super::consumer_loop::ConsumerState,
/// One dry run at a time (they share a target dir per member).
pub(super) dry: Mutex<()>,
/// One HEAVY load at a time: a room build, the consumer's open-set
/// load. Their transients never add up (a restart started both: the
/// 8.7 GB VmHWM on dwarf, 2026-10-16). A kept room passes by it.
pub(super) heavy: crate::monitor::WorkspaceGate,
/// Answers computed from a room, by (room version, surface, request):
/// the JSON text the door hands out (see [`super::reader::ready`]).
pub(super) answers: Mutex<super::reader::AnswerCache>,
/// `world_impact`'s commit windows, by (repo, seq): a window never
/// changes once folded, and folding one costs two as-of rooms.
pub(super) windows: Mutex<crate::mimir_lens::world_impact::Windows>,
}
impl CodeRoom {
/// The workspace's event log: opened (the lake created) on first use,
/// then shared — every caller gets a clone of the same handle, so a
/// reader never waits for the consumer and never opens a second lake.
///
/// SYNC (the open blocks): fine from any thread, but an async task
/// should be on the blocking pool.
pub fn log(&self) -> Result<EventLog, Refusal> {
let mut slot = self.engine.log.lock().unwrap_or_else(|e| e.into_inner());
if let Some(log) = slot.as_ref() {
return Ok(log.clone());
}
let lake = self.layout.lake();
let log = EventLog::open_lake_blocking(&lake).map_err(|e| {
if e.downcast_ref::<LakeSchemaMismatch>().is_some() {
// Not a difference the open evolves in place (added nullable
// columns are): derived data, say exactly what to remove.
return Refusal::new(
State::LakeUnavailable,
format!(
"code room of `{}`: {e:#}. Stop the server, remove {} and start it again (the clones and the driver's target dir beside it can stay)",
self.workspace,
lake.display()
),
);
}
let state = if e.downcast_ref::<LakeHeld>().is_some() { State::LakeUnavailable } else { State::Internal };
Refusal::new(state, format!("code room of `{}`: {e:#}", self.workspace))
})?;
*slot = Some(log.clone());
Ok(log)
}
/// Is the lake open in this process (did anything use the room yet)?
pub fn is_open(&self) -> bool {
self.engine.log.lock().unwrap_or_else(|e| e.into_inner()).is_some()
}
/// Close the lake (the next use reopens it). Only meaningful when no
/// other clone of the log is alive — the lake closes with the last one.
pub fn close(&self) {
self.engine.consumer.close();
self.engine.log.lock().unwrap_or_else(|e| e.into_inner()).take();
}
/// **Build and keep the workspace's current room** (all consumed repos,
/// the dependency ring off) — what the panes ask first, so a server
/// queues this on its jera `rooms` lane after a consume and the first
/// question finds it built. Returns (repos, milliseconds).
pub fn warm(&self) -> Result<(usize, u64), Refusal> {
let t0 = std::time::Instant::now();
let room = self.room(&Scope::current())?;
Ok((room.scope().repos.len(), t0.elapsed().as_millis() as u64))
}
/// **Keep the dependency-ring view alive** (`Scope::*.with_universe(true)`)
/// after a consume. A kept ring fold of THIS workspace's log is
/// opened so it ADVANCES on this lane, and a boundary lens's next
/// ask (`data_reach` / `reachable_vuln` / `system_boundary`) is a
/// cache HIT instead of a refold of the whole universe. With NO kept
/// ring view and `birth` true (the caller saw a system boundary in
/// the lake), the FIRST fold is paid here too: live, the first
/// 140-166 s ring fold happened under a lens on the panes lane and
/// kept losing the post-consume races, so the view was never alive
/// to advance. `birth` false keeps the no-boundary law (a lake the
/// boundary never ran on must not fold 8.2 M nodes to say "no
/// boundary"). `None` = no kept ring view and no birth. Returns
/// (advanced by the delta, wall ms).
pub fn warm_ring(&self, birth: bool) -> Result<Option<(bool, u64)>, Refusal> {
let log = self.log()?;
let scope = match Room::kept_universe_scope_blocking(&log).map_err(|e| self.room_refusal(e))? {
Some(s) => s,
None if birth => Scope::current().with_universe(true),
None => return Ok(None),
};
// ONE retry on a hard error: the post-consume lake compaction
// can sweep a parquet from under the first attempt's pinned read
// (live 2026-10-20: the stolen fold's advance died on a swept
// file and the view was lost); the retry pins the
// post-compaction state.
let opened = match self.room_opened(&scope) {
Ok(o) => o,
Err(r) => {
eprintln!("nornir-consumer: {}: ring view warm attempt failed [{}], retrying once: {r}", self.workspace, r.state);
self.room_opened(&scope)?
}
};
Ok(Some((opened.advanced, opened.ms)))
}
/// The room under `scope` ONLY when it is built and still valid —
/// never a fold (a door that must answer at once queues the build).
/// A repo the log never consumed is [`State::NotConsumed`].
pub fn peek(&self, scope: &Scope) -> Result<Option<Room>, Refusal> {
let log = self.log()?;
Room::peek_blocking(&log, scope).map_err(|e| self.room_refusal(e))
}
/// `world_impact`'s kept commit windows, by (repo, seq).
pub(crate) fn windows(&self) -> std::sync::MutexGuard<'_, crate::mimir_lens::world_impact::Windows> {
self.engine.windows.lock().unwrap_or_else(|e| e.into_inner())
}
/// A traced run was filed: the replay's kept `Viz.CodeMap` answers
/// (`trace_runs` / `trace` / `profile`) are of the runs before it.
pub fn forget_trace_answers(&self) {
self.answers().forget_lods("Viz.CodeMap", &["trace_runs", "trace", "profile"]);
}
/// The kept answers ([`super::reader::ready`]).
pub(crate) fn answers(&self) -> std::sync::MutexGuard<'_, super::reader::AnswerCache> {
self.engine.answers.lock().unwrap_or_else(|e| e.into_inner())
}
/// The room under `scope` (cached per process by `mimir-room`): the
/// last committed release, also while the consumer is mid-commit.
///
/// SYNC (wiring map R14): callable from any thread; an async task
/// should be on the blocking pool. A repo the log never consumed is
/// [`State::NotConsumed`] (R15), never an empty room.
pub fn room(&self, scope: &Scope) -> Result<Room, Refusal> {
Ok(self.room_opened(scope)?.room)
}
/// [`Self::room`], telling how it was got: served from the process
/// cache or built, and the wall clock of the open — what a lens job
/// files as `room_ms` / `room_cached`.
pub fn room_opened(&self, scope: &Scope) -> Result<Opened, Refusal> {
let log = self.log()?;
let t = std::time::Instant::now();
// kept and valid: no build, no wait
if let Some(room) = Room::peek_blocking(&log, scope).map_err(|e| self.room_refusal(e))? {
return Ok(Opened { room, cached: true, advanced: false, ms: t.elapsed().as_millis() as u64 });
}
// a build: through the heavy gate (one big transient at a time);
// `ms` is the whole wait the caller saw, the gate included
let mut opened = self.engine.heavy.enter_as("room build", || Room::open_blocking(&log, scope)).map_err(|e| self.room_refusal(e))?;
opened.ms = t.elapsed().as_millis() as u64;
Ok(opened)
}
fn room_refusal(&self, e: anyhow::Error) -> Refusal {
let state = match e.downcast_ref::<ScopeError>() {
Some(ScopeError::NotConsumed { .. }) => State::NotConsumed,
Some(ScopeError::PastOffset { .. } | ScopeError::NegativeSeq { .. }) => State::InvalidArgument,
None => State::Internal,
};
Refusal::new(state, format!("code room of `{}`: {e:#}", self.workspace))
}
}
}
#[cfg(all(test, feature = "code-room"))]
pub(crate) mod consumer_tests;
#[cfg(test)]
mod tests {
use super::*;
#[test]
fn the_layout_is_the_wiring_maps() {
let l = Layout::under(Path::new("/r/ws/builds"));
assert_eq!(l.lake(), Path::new("/r/ws/builds/mimir_v2/lake"));
assert_eq!(l.clone_of("skade"), Path::new("/r/ws/builds/mimir_v2/clones/skade"));
assert_eq!(l.target_of("skade"), Path::new("/r/ws/builds/mimir_v2/target/skade"));
assert_eq!(l.shared_target(), Path::new("/r/ws/builds/mimir_v2/target-shared"));
assert_eq!(l.out_of("skade"), Path::new("/r/ws/builds/mimir_v2/out/skade"));
assert_eq!(l.std_root(), Path::new("/r/ws/builds/mimir_v2/std"));
assert_eq!(l.artifacts_of("skade", "0b07"), Path::new("/r/ws/builds/mimir_v2/artifacts/skade/0b07"));
}
#[test]
fn a_door_that_reaches_nothing_refuses_by_state() {
let r = Reach::none().code_room().unwrap_err();
assert_eq!(r.state, State::NoRoom);
assert_eq!(r.to_json()["state"], "no_room");
}
#[test]
fn constructing_the_handle_touches_no_disk() {
let dir = tempfile::tempdir().unwrap();
let room = CodeRoom::new("ws", dir.path());
assert_eq!(room.workspace(), "ws");
assert!(!dir.path().join(DIR).exists(), "the lake is opened on first use, not on construction");
}
/// **A room build and the consumer's open-set load never run at once.**
/// They are the process's two big transients (dwarf 2026-10-16: the
/// current room's build peaks at 4 GB, the open-set load holds ~1 GB, and
/// a restart started both together — the 8.7 GB VmHWM). Both go through
/// the room's heavy gate: the test holds it, and both must queue there
/// (explicit sync, no clock), then both run, one at a time.
#[cfg(feature = "code-room")]
#[test]
fn a_room_build_and_the_open_set_load_wait_for_each_other() {
use std::sync::{Arc, mpsc};
// a lake of its own with a consumed repo: its room is built, not kept
let ws = super::reader::fixture::fresh_tiny();
let handle: Arc<CodeRoom> = ws.handle.clone();
let (held_tx, held) = mpsc::channel();
let (go, go_rx) = mpsc::channel::<()>();
let h = handle.clone();
let holder = std::thread::spawn(move || {
h.engine.heavy.enter_as("test", || {
held_tx.send(()).unwrap();
go_rx.recv().unwrap();
})
});
held.recv().unwrap();
let h = handle.clone();
let build = std::thread::spawn(move || h.room(&mimir_room::Scope::current()).map(|r| r.len()));
let h = handle.clone();
let boot = std::thread::spawn(move || h.with_consumer(|_, _| Ok(())));
let t0 = std::time::Instant::now();
while handle.engine.heavy.queued() < 2 {
assert!(
t0.elapsed() < std::time::Duration::from_secs(20),
"{} of the room build and the open-set load wait at the heavy gate — the other ran beside whatever holds it (built: {}, booted: {})",
handle.engine.heavy.queued(),
build.is_finished(),
boot.is_finished()
);
std::thread::yield_now();
}
assert!(!build.is_finished() && !boot.is_finished());
go.send(()).unwrap();
holder.join().unwrap();
assert!(build.join().unwrap().expect("the room builds") > 0);
boot.join().unwrap().expect("the consumer boots");
assert_eq!(handle.engine.heavy.peak_in_flight(), 1, "two heavy loads ran at once");
}
/// An empty lake has no offsets: a room over it is empty, and the open
/// creates the lake where the layout says.
#[cfg(feature = "code-room")]
#[test]
fn the_room_of_a_fresh_workspace_is_empty_and_lives_under_the_layout() {
let dir = tempfile::tempdir().unwrap();
let handle = CodeRoom::new("ws", dir.path());
assert!(!handle.is_open());
let room = handle.room(&mimir_room::Scope::current()).expect("an empty lake opens");
assert!(room.is_empty());
assert!(handle.is_open());
assert!(handle.layout().lake().join("catalog.redb").exists());
}
/// R15 through the handle: a repo nothing consumed is `not_consumed`, a
/// state a front door can act on — not an internal error, not an empty room.
#[cfg(feature = "code-room")]
#[test]
fn a_repo_that_was_never_consumed_refuses_as_not_consumed() {
let dir = tempfile::tempdir().unwrap();
let handle = CodeRoom::new("ws", dir.path());
let r = handle.room(&mimir_room::Scope::repo("skade", None)).err().expect("never consumed");
assert_eq!(r.state, State::NotConsumed, "{r}");
assert!(r.note.contains("skade"));
}
/// A lake that differs by more than added nullable columns (those are
/// evolved in place on open) is not migrated and not appended to: the
/// refusal names the directory to remove.
#[cfg(feature = "code-room")]
#[test]
fn a_lake_of_another_schema_refuses_and_says_what_to_remove() {
use skade::arrow_schema::{DataType, Field, Schema};
let dir = tempfile::tempdir().unwrap();
let handle = CodeRoom::new("ws", dir.path());
let lake = handle.layout().lake();
// a lake whose node table has yesterday's columns
let rt = tokio::runtime::Builder::new_current_thread().enable_all().build().unwrap();
rt.block_on(async {
let wh = skade::open(&lake).await.unwrap();
let old = Schema::new(vec![Field::new("repo", DataType::Utf8, false), Field::new("moniker", DataType::Utf8, false)]);
wh.create_table("mimir_v2.code_node_events", &old).await.unwrap();
});
drop(rt);
let r = handle.log().err().expect("another schema");
assert_eq!(r.state, State::LakeUnavailable, "{r}");
assert!(r.note.contains("not migrated") && r.note.contains(&lake.display().to_string()) && r.note.contains("mimir_v2.code_node_events"), "{r}");
// removing it is the whole remedy
std::fs::remove_dir_all(&lake).unwrap();
assert!(handle.room(&mimir_room::Scope::current()).unwrap().is_empty());
}
/// The cross-process story: a second opener of a held lake (here a second
/// handle; a second process hits the same file lock) is `lake_unavailable`
/// and is told who to ask — it does not get a private copy.
#[cfg(feature = "code-room")]
#[test]
fn a_held_lake_refuses_a_second_opener_as_lake_unavailable() {
let dir = tempfile::tempdir().unwrap();
let holder = CodeRoom::new("ws", dir.path());
holder.log().unwrap();
let second = CodeRoom::new("ws", dir.path());
let r = second.log().err().expect("the lake is held");
assert_eq!(r.state, State::LakeUnavailable, "{r}");
assert!(r.note.contains("single-process"), "{r}");
// every clone of the holder's log is the same open lake
assert!(holder.log().is_ok() && holder.room(&mimir_room::Scope::current()).is_ok());
holder.close();
assert!(second.log().is_ok(), "free once the holder let go");
}
}