src/code_room/feeds.rs
//! The non-graph feeds that ride the consumer's offsets.
//!
//! The v1 populate ran these as steps of one pass (`monitor::republish_with`);
//! the populate is gone and so was their trigger — the lenses over them
//! (`anomaly`, `churn`, `change_risk`, `reachable_vuln`, …) and the search
//! tabs would have stayed blind although nothing about the feeds changed
//! (wiring map P7). They are the SAME producer functions, hung on the
//! consumer: once a member is caught up at (seq, sha), every feed that was
//! never tried for that sha runs against the server's clone — which is
//! checked out at exactly that commit — and writes its own row in the lake's
//! `consumer_feed_runs` ledger.
//!
//! | feed | producer (unchanged) | writes |
//! |---|---|---|
//! | `git_heat` | `knowledge::git_heat::scan_repo` | `git_heat_facts` |
//! | `sbom` | `security::warm_prepare` / `warm_resolve` / `persist_sbom` | `sbom_components` |
//! | `dep_graph` | `WorkspaceGraph::build_from_members` + `record_dep_graph` | `dep_graph_edges` (the whole workspace: a member moved) |
//! | `code_index` | `Index::build_scoped` + `index::snapshot::snapshot_to_iceberg` | the tantivy index + `tantivy_index_snapshots` / `_blobs` |
//! | `binary` | `mimir_ingest::artifact` (find + join) | the lake's BINARY stream (`bin_node_events`, `bin_edge_events`, `code_symbol_events`) |
//! | `clone_shapes` | `clone_signals::feed::shapes_member` | `fn_shape_events` (the normalised body per function: exact key, MinHash) |
//! | `embeddings` | `monitor::read_embed_inputs` + `embed_read_inputs` | `embeddings` / `embedding_manifest` / `embedding_snapshots`; per function `embeddings` + `embeddings_norm` |
//!
//! Two of them are not like the others:
//!
//! * **`binary`** (the binary layer, plan lane 9) writes the lake, not the
//! warehouse, and builds NOTHING: it reads the linked artifacts somebody put
//! under `<builds>/mimir_v2/artifacts/<member>/<sha>/<profile>/` for exactly
//! this commit, joins them against the lake's own source rows, and hands the
//! rows to the consumer's seat. No artifact there = the typed `no_artifact`
//! row naming that directory; the hook looks again when one appears.
//! * **`embeddings`** is DEFERRED ([`Feed::prepare`]): inside the run it only
//! reads the member's sources; the embed itself — minutes per member on a CPU
//! embedder — runs afterwards with no lock of the workspace held, so a commit
//! never waits for a vector.
//!
//! Not here on purpose: the FEATURE graph (`feature_dep_edges`) — features
//! are nodes of the code graph now, from the driver's one `cargo metadata`
//! (feature nodes, ENABLES, GATED_BY); a second producer would be a second
//! truth. And `vuln_findings`: it never rode the populate (`warehouse scan
//! --persist` / modgunn write it), so it lost no trigger.
use std::path::PathBuf;
use std::sync::Arc;
use anyhow::{Context, Result, bail};
use mimir_ingest::{EventLog, FeedRow, artifact, schema};
use super::consumer_loop::{DeferredWork, Feed, FeedAt, FeedResult};
use crate::warehouse::iceberg::IcebergWarehouse;
/// What the feeds of one workspace write through and read from.
#[derive(Clone)]
pub struct Sink {
pub warehouse: Arc<IcebergWarehouse>,
/// The workspace's code index (the handle the search RPCs read).
pub index: Arc<crate::index::Index>,
/// `<root>/<ws>/git`.
pub git_dir: PathBuf,
/// `<root>/<ws>/builds/cache/index`.
pub index_dir: PathBuf,
/// Every member of the workspace (the dep graph is cross-member).
pub members: Vec<String>,
/// The job ledger of the process that runs the feeds (a server). With it
/// the `embeddings` feed starts the dependency ring's embed as a
/// background job when the ring moved; `None` (the CLI's inline consume:
/// the process ends with the verb) = the ring is embedded on request only
/// (`nornir workspace embed <ws> --ring`).
pub jobs: Option<crate::jobs::JobSink>,
/// The jera lane a server queues background work of the feeds on (the
/// dependency ring's embed). `None`: the CLI — such work takes a thread.
pub ring_queue: Option<Arc<crate::jobs::chain::ChainQueue>>,
}
/// The feeds, in the order they run.
pub fn standard(sink: &Sink) -> Vec<Box<dyn Feed>> {
vec![
Box::new(GitHeat(sink.clone())),
Box::new(Sbom(sink.clone())),
Box::new(DepGraph(sink.clone())),
Box::new(CodeIndex(sink.clone())),
Box::new(Binary),
Box::new(CloneShapes(sink.clone())),
Box::new(Embeddings::on_this_box(sink.clone())),
]
}
/// The feeds' names, in order (status and reports name them without a sink).
pub const NAMES: [&str; 7] = ["git_heat", "sbom", "dep_graph", "code_index", "binary", "clone_shapes", "embeddings"];
/// The clone signals' per-function shapes (`clone_signals`): the normalised
/// body's exact key and MinHash of every function whose body changed at this
/// commit → `fn_shape_events` — no model, seconds per member, written while
/// the run holds the tree. Then, where this box embeds (an embedder and
/// hardware, or the GPU worker of `$NORNIR_EMBED_REMOTE`), the NORMALISED
/// bodies that have no vector yet → `embeddings_norm`, DEFERRED like the
/// `embeddings` feed (the same embedder, one embedding run at a time). A new
/// feed, so a server that gains it runs it once for every member's current
/// commit: that is the backfill, through the same worker.
pub struct CloneShapes(pub Sink);
#[cfg(all(feature = "vector", feature = "code-room"))]
impl CloneShapes {
fn shapes(&self, at: &FeedAt<'_>) -> Result<(Vec<crate::vector::fnvec::FnChunk>, String)> {
let chunks = crate::clone_signals::feed::read_chunks(at.log, &at.member.name, &at.member.source, at.seq)?;
let run = crate::clone_signals::feed::write_shapes(&self.0.warehouse, &at.member.name, at.seq, at.sha, &chunks)?;
Ok((chunks, run.line()))
}
}
impl Feed for CloneShapes {
fn name(&self) -> &'static str {
"clone_shapes"
}
/// A run that failed on an unreachable GPU worker is due again once it answers.
fn retry(&self, _at: &FeedAt<'_>, last: &FeedRow) -> bool {
last.status == schema::STATUS_ERROR && last.detail.as_deref().is_some_and(|d| d.contains(WORKER_WORDS)) && embed_worker_up()
}
#[cfg(all(feature = "vector", feature = "code-room"))]
fn run(&self, at: &FeedAt<'_>) -> Result<FeedResult> {
let (_, line) = self.shapes(at)?;
let why = crate::monitor::no_embedder_reason().map(|w| w.to_string()).or_else(no_embedding_hardware).unwrap_or_else(|| "not asked".into());
Ok(FeedResult::Ok { rows: 1, detail: format!("shapes: {line}; normalised bodies and groups not embedded here: {why}") })
}
#[cfg(not(all(feature = "vector", feature = "code-room")))]
fn run(&self, _at: &FeedAt<'_>) -> Result<FeedResult> {
Ok(FeedResult::Skipped("built without the `vector` + `code-room` features (the clone signals)".into()))
}
#[cfg(all(feature = "vector", feature = "code-room", any(feature = "embed-tract", feature = "embed-ort")))]
fn prepare(&self, at: &FeedAt<'_>) -> Option<Result<DeferredWork>> {
(crate::monitor::no_embedder_reason().is_none() && no_embedding_hardware().is_none()).then(|| -> Result<DeferredWork> {
let (chunks, line) = self.shapes(at)?;
let wh = self.0.warehouse.clone();
Ok(Box::new(move || {
let _turn = crate::vector::fnvec::embed_turn();
let mut embedding = crate::vector::fnvec::Compiled::default();
let norm = crate::clone_signals::feed::embed_normalised(&wh, &chunks, &mut embedding)?;
// the group twins' half: the normalised semantic GROUPS, same
// embedder, same content-addressing → `embeddings_groups`
let groups = crate::clone_signals::feed::embed_groups(&wh, &chunks, &mut embedding)?;
Ok(FeedResult::Ok {
rows: (norm.new_vectors + groups.new_vectors) as i64,
detail: format!(
"shapes: {line}; normalised bodies: {} distinct new, {} vector(s) embedded in {:.1}s; groups: {} distinct new, {} vector(s) in {:.1}s",
norm.distinct,
norm.new_vectors,
norm.embed_ms as f64 / 1000.0,
groups.distinct,
groups.new_vectors,
groups.embed_ms as f64 / 1000.0
),
})
}) as DeferredWork)
})
}
}
/// Per-file churn / authors of the member's history → `git_heat_facts`.
pub struct GitHeat(pub Sink);
impl Feed for GitHeat {
fn name(&self) -> &'static str {
"git_heat"
}
fn run(&self, at: &FeedAt<'_>) -> Result<FeedResult> {
let scan = crate::knowledge::git_heat::scan_repo(&at.member.source, &at.member.name, uuid::Uuid::new_v4(), chrono::Utc::now())?;
// A member never scans to nothing: 0 files means the walk was wrong,
// and an empty snapshot would read as "no churn anywhere".
if scan.files.is_empty() {
bail!("the git-heat scan of {} found 0 files — not persisting an empty snapshot", at.member.source.display());
}
self.0.warehouse.append_git_heat_scan(&scan)?;
Ok(FeedResult::Ok { rows: scan.files.len() as i64, detail: format!("git_heat snapshot {}", scan.snapshot_id) })
}
}
/// The member's resolved component closure → `sbom_components`.
pub struct Sbom(pub Sink);
impl Feed for Sbom {
fn name(&self) -> &'static str {
"sbom"
}
fn run(&self, at: &FeedAt<'_>) -> Result<FeedResult> {
if !at.member.source.join("Cargo.toml").exists() {
return Ok(FeedResult::Skipped("no Cargo.toml at the repo root: no cargo component closure".into()));
}
// serial: it clones / links missing path-dep siblings under the shared git dir
crate::security::warm_prepare(&at.member.source, &self.0.git_dir);
let (repo, components) = crate::security::warm_resolve(&at.member.source).context("resolve the component closure (cargo metadata)")?;
crate::security::persist_sbom(&self.0.warehouse, &repo, &components)?;
Ok(FeedResult::Ok { rows: components.len() as i64, detail: format!("{} component(s) of {repo}", components.len()) })
}
}
/// The cross-member dependency graph (a direct manifest parse, no cargo) →
/// `dep_graph_edges`. Workspace-wide: rebuilt when any member moved.
pub struct DepGraph(pub Sink);
impl Feed for DepGraph {
fn name(&self) -> &'static str {
"dep_graph"
}
fn run(&self, at: &FeedAt<'_>) -> Result<FeedResult> {
let present: std::collections::BTreeMap<String, PathBuf> = self
.0
.members
.iter()
.map(|m| (m.clone(), self.0.git_dir.join(m)))
.filter(|(_, dir)| dir.join(".git").exists())
.collect();
let graph = crate::warehouse::dep_graph::WorkspaceGraph::build_from_members(&present).context("build the cross-member dep graph")?;
self.0.warehouse.record_dep_graph(at.workspace, &graph)?;
Ok(FeedResult::Ok { rows: present.len() as i64, detail: format!("dep graph over {} member(s)", present.len()) })
}
}
/// The member's files into the workspace code index, and the index snapshot
/// keyed to the member's commit.
pub struct CodeIndex(pub Sink);
impl Feed for CodeIndex {
fn name(&self) -> &'static str {
"code_index"
}
fn run(&self, at: &FeedAt<'_>) -> Result<FeedResult> {
let stats = self.0.index.build_scoped(std::slice::from_ref(&at.member.name)).context("build the code index")?;
// A member WITH a checkout whose walk found nothing is a broken walk,
// not an empty repo: `nornir search` would return nothing and say so nowhere.
if stats.scanned == 0 {
bail!("the code-index walk of `{}` found 0 files under {}", at.member.name, at.member.source.display());
}
let branch = crate::gitio::head_branch(&at.member.source).unwrap_or_else(|_| "unknown".into());
let snap = crate::index::snapshot::snapshot_to_iceberg(&self.0.warehouse, at.workspace, &at.member.name, at.sha, &branch, &self.0.index_dir)
.with_context(|| format!("snapshot the index for {}", at.member.name))?;
Ok(FeedResult::Ok {
rows: stats.scanned as i64,
detail: format!("{} file(s): {} added, {} updated, {} unchanged; snapshot {}", stats.scanned, stats.added, stats.updated, stats.skipped_unchanged, snap.snapshot_id),
})
}
}
/// **The binary hook**: the linked artifacts of the member's commit → the
/// lake's binary stream, on the same node ids as the source.
///
/// It reads; it never builds. Where it looks:
/// `<builds>/mimir_v2/artifacts/<member>/<sha>/` — a cargo target dir of that
/// commit (`<profile>/<bin>`, optionally `<profile>/deps/`), filled by whoever
/// built the commit (`cargo build --release --target-dir <that dir>`, or a
/// copy of the product build's output). The sha in the path is the claim that
/// the artifact is of this commit; nothing is inferred from file times.
///
/// | what it finds | ledger row |
/// |---|---|
/// | no such directory, or no executable of a bin the workspace declares | `no_artifact`, naming the directory — and the hook looks again whenever a consume asks what is behind |
/// | artifacts | `ok`: one binary release (every artifact found, joined against the member's source rows at this commit + the universe ring) |
/// | the workspace does not load (`cargo metadata`) | `error` |
pub struct Binary;
impl Binary {
pub const NAME: &'static str = "binary";
/// The directory the hook reads for `at`.
pub fn dir(at: &FeedAt<'_>) -> PathBuf {
at.layout.artifacts_of(&at.member.name, at.sha)
}
/// `no_artifact`, saying where it looked and what would be found there.
fn absent(dir: &std::path::Path, why: &str) -> FeedResult {
FeedResult::Absent {
state: schema::STATUS_NO_ARTIFACT,
detail: format!(
"{why} — looked in {} (a cargo target dir of this commit: `<profile>/<bin>`; nothing here builds into it — e.g. `cargo build --release --target-dir {}`)",
dir.display(),
dir.display()
),
}
}
}
impl Feed for Binary {
fn name(&self) -> &'static str {
Self::NAME
}
fn run(&self, at: &FeedAt<'_>) -> Result<FeedResult> {
let dir = Self::dir(at);
if !dir.is_dir() {
return Ok(Self::absent(&dir, "no linked artifact of this commit"));
}
let manifest = at.member.source.join("Cargo.toml");
if !manifest.exists() {
return Ok(FeedResult::Skipped("no Cargo.toml at the repo root: no cargo bins to look for".into()));
}
let packages = mimir_binary_packages(&manifest)?;
let bins = artifact::declared_bins(&packages);
let found = artifact::find(&dir, &bins);
if found.is_empty() {
let want: Vec<&str> = bins.iter().map(String::as_str).collect();
return Ok(Self::absent(&dir, &format!("the directory holds no executable of a bin this workspace declares ({})", if want.is_empty() { "it declares none".to_string() } else { want.join(", ") })));
}
// ONE committed state of the lake for the whole join
let pinned = at.log.block_on(at.log.pin())?;
let src = at.log.block_on(artifact::source_half(&pinned, &at.member.name, at.seq)).context("read the source half from the lake")?;
let cx = artifact::JoinContext { packages, sysroot: toolchain_sysroot(&at.member.source), triple: host_triple()? };
let mut dumps = Vec::new();
let mut lines = Vec::new();
for a in &found {
let d = artifact::join(&src, a, &cx, at.sha)?;
lines.push(format!("{} ({})", d.artifact, a.path.display()));
dumps.push(d);
}
Ok(FeedResult::Binaries { dumps, declared_bins: bins, detail: format!("{} artifact(s) from {}: {}", found.len(), dir.display(), lines.join(", ")) })
}
/// `no_artifact` for this sha: look again once the directory is there.
fn retry(&self, at: &FeedAt<'_>, last: &FeedRow) -> bool {
last.status == schema::STATUS_NO_ARTIFACT && Self::dir(at).is_dir() && std::fs::read_dir(Self::dir(at)).is_ok_and(|mut d| d.next().is_some())
}
}
/// `cargo metadata` of the member's workspace (the packages a symbol's crate
/// is resolved to).
fn mimir_binary_packages(manifest: &std::path::Path) -> Result<Vec<artifact::MetaPackage>> {
artifact::cargo_metadata(manifest).map_err(|e| anyhow::anyhow!("cargo metadata of {}: {e}", manifest.display()))
}
/// The sysroot of the toolchain a build in `dir` uses (rustup reads the
/// tree's own `rust-toolchain`), when rustc answers.
fn toolchain_sysroot(dir: &std::path::Path) -> Option<PathBuf> {
let o = std::process::Command::new("rustc").current_dir(dir).args(["--print", "sysroot"]).output().ok()?;
o.status.success().then(|| PathBuf::from(String::from_utf8_lossy(&o.stdout).trim())).filter(|p| p.is_dir())
}
/// This box's target triple (an artifact found here was linked for it).
fn host_triple() -> Result<String> {
let o = std::process::Command::new("rustc").arg("-vV").output().context("rustc -vV")?;
String::from_utf8_lossy(&o.stdout).lines().find_map(|l| l.strip_prefix("host: ").map(|h| h.trim().to_string())).context("rustc -vV printed no host line")
}
/// The member's code → vectors, in two halves. `skipped` (with the reason) in
/// a build without an embedder, and on a box without embedding hardware —
/// the feed is then not retried until asked.
///
/// 1. FUNCTIONS: one vector per function node of the member, cut by the
/// driver's span at this commit ([`crate::vector::fnvec`]) — only the
/// functions whose text changed, deduped by content hash. What the
/// `similar_code` / `semantic_clones` lenses read.
/// 2. LINE WINDOWS: the snapshot free-text search reads
/// (`monitor::embed_workspace_with`), unchanged.
///
/// DEFERRED: [`Feed::prepare`] reads BOTH halves out of the member's tree —
/// the function texts and the line windows, seconds — while the run holds the
/// tree still; the embed, as long as the embedder is slow, is the returned
/// work and runs with nothing held. The dependency ring moves with the
/// lockfile, not with the commit: it is never embedded here — a background
/// job, once per crate@version, started when the member's embed (and its hold
/// on the embedder) is over, written to the process's job ledger.
pub struct Embeddings {
pub sink: Sink,
/// Why an embed nobody is waiting for should not run on this box, if it
/// should not (`vector::select::live_unattended_embed_skip`: no GPU worth
/// using — an AMD on-chip GPU is none — and nobody asked for CPU).
pub no_hardware: fn() -> Option<String>,
/// Does the GPU worker this process names (`$NORNIR_EMBED_REMOTE`) answer
/// now? A run that failed because it did not is due again once it does.
pub worker_up: fn() -> bool,
}
/// Why this box does not embed unattended, if it does not. `pub(crate)`: the
/// `groups_backfill` job honours the SAME gate as the `embeddings` feed.
pub(crate) fn no_embedding_hardware() -> Option<String> {
// a GPU worker named ($NORNIR_EMBED_REMOTE): the hardware is the worker's
#[cfg(all(any(feature = "embed-tract", feature = "embed-ort"), feature = "wire-client"))]
if crate::vector::remote::configured().is_some() {
return None;
}
// asked on every "is anything behind" check of a member whose sha was
// skipped for want of hardware: the probe (dlopen + dir scans) answers
// once a minute, hardware does not come and go faster
#[cfg(all(feature = "embed-tract", feature = "embed-ort"))]
{
static LAST: std::sync::Mutex<Option<(std::time::Instant, Option<String>)>> = std::sync::Mutex::new(None);
let mut last = LAST.lock().unwrap_or_else(|e| e.into_inner());
if let Some((at, why)) = last.as_ref().filter(|(at, _)| at.elapsed() < std::time::Duration::from_secs(60)) {
let _ = at;
return why.clone();
}
let why = crate::vector::select::live_unattended_embed_skip();
*last = Some((std::time::Instant::now(), why.clone()));
why
}
// one embedder compiled (or none): there is no selector to ask, the
// embed step itself says what it can do
#[cfg(not(all(feature = "embed-tract", feature = "embed-ort")))]
{
None
}
}
/// Does the named GPU worker answer (connect + `Embedder.Profile` in this
/// process's model space)? Asked on every "is anything behind" check of a
/// member whose embed failed on the worker: answered once a minute.
fn embed_worker_up() -> bool {
#[cfg(all(any(feature = "embed-tract", feature = "embed-ort"), feature = "wire-client"))]
{
if crate::vector::remote::configured().is_none() {
return false;
}
static LAST: std::sync::Mutex<Option<(std::time::Instant, bool)>> = std::sync::Mutex::new(None);
let mut last = LAST.lock().unwrap_or_else(|e| e.into_inner());
if let Some((_, up)) = last.as_ref().filter(|(at, _)| at.elapsed() < std::time::Duration::from_secs(60)) {
return *up;
}
let up = crate::vector::remote::RemoteEmbedder::from_env().is_ok();
*last = Some((std::time::Instant::now(), up));
up
}
#[cfg(not(all(any(feature = "embed-tract", feature = "embed-ort"), feature = "wire-client")))]
{
false
}
}
/// Does the `embeddings` feed fill the vector tables on THIS box: an embedder
/// is compiled, and there is hardware for an unattended embed (or CPU was
/// asked for). What a completeness check may EXPECT — empty vectors on a box
/// that cannot embed are a fact about the box, not a feed that went missing.
pub fn embeddings_expected() -> bool {
cfg!(any(feature = "embed-tract", feature = "embed-ort")) && no_embedding_hardware().is_none()
}
impl Embeddings {
/// The feed as this box's hardware decides it.
pub fn on_this_box(sink: Sink) -> Self {
Self { sink, no_hardware: no_embedding_hardware, worker_up: embed_worker_up }
}
/// `skipped` — the reason whole, and the verb that embeds anyway — when
/// this box should not embed unattended (hours on a CPU for a repo).
fn not_on_this_box(&self, workspace: &str) -> Option<FeedResult> {
(self.no_hardware)().map(|why| FeedResult::Skipped(format!("{why}. {NOT_UNATTENDED}: `nornir workspace embed {workspace}` embeds now, on what is here")))
}
/// The last run of this sha was skipped for want of hardware, and there
/// is hardware now (a GPU worker was named, a GPU came up): run it again.
/// Any other skip stays settled.
fn hardware_came(&self, last: &FeedRow) -> bool {
last.status == schema::STATUS_SKIPPED
&& last.detail.as_deref().is_some_and(|d| d.contains(NOT_UNATTENDED))
&& crate::monitor::no_embedder_reason().is_none()
&& (self.no_hardware)().is_none()
}
/// The last run of this sha failed because the GPU worker did not answer
/// (down, restarting, the box off), and it answers now: run it again. Any
/// other failure stays settled until the member moves or a fetch forces it.
fn worker_came_back(&self, last: &FeedRow) -> bool {
last.status == schema::STATUS_ERROR && last.detail.as_deref().is_some_and(|d| d.contains(WORKER_WORDS)) && (self.worker_up)()
}
/// The two halves' outcomes as the feed's result.
fn result(functions: (usize, String), rep: crate::monitor::EmbedReport, member: &str, workspace: &str) -> Result<FeedResult> {
if let Some(why) = rep.skipped {
return Ok(FeedResult::Skipped(why));
}
if rep.members_failed > 0 {
bail!("functions: {} — but the line-window embed of `{member}` failed (see `nornir workspace events {workspace}`)", functions.1);
}
Ok(FeedResult::Ok {
rows: (functions.0 + rep.new_vectors) as i64,
detail: format!("functions: {}; line windows: {} chunk(s), {} new vector(s), {} reused", functions.1, rep.occurrences, rep.new_vectors, rep.reused_vectors()),
})
}
/// Both halves in one call, the caller holding the tree throughout.
fn embed(&self, at: &FeedAt<'_>) -> Result<FeedResult> {
// one embedding run at a time in this process (the ring job is the other one)
#[cfg(feature = "vector")]
let _turn = crate::vector::fnvec::embed_turn();
let functions = match function_vectors(&self.sink, at)? {
Ok(line) => line,
Err(why) => return Ok(FeedResult::Skipped(why)),
};
let rep = crate::monitor::embed_workspace_with(&self.sink.warehouse, &self.sink.git_dir, std::slice::from_ref(&at.member.name), at.workspace)?;
Self::result(functions, rep, &at.member.name, at.workspace)
}
}
/// An `ok` result with the ring job's words appended (the job is started here,
/// after the member's embed let go of the embedder).
fn with_ring(ran: Result<FeedResult>, sink: &Sink, workspace: &str, log: &EventLog) -> Result<FeedResult> {
match ran {
Ok(FeedResult::Ok { rows, detail }) => Ok(FeedResult::Ok { rows, detail: format!("{detail}{}", ring_job(sink, workspace, log)) }),
other => other,
}
}
/// The words of the hardware gate's skip — what [`Embeddings::hardware_came`]
/// recognises its own skip by.
const NOT_UNATTENDED: &str = "Not embedded unattended";
/// What every failure to reach the GPU worker says (`vector::remote`: "connect
/// to the embed worker at …", "embed worker … : Embed of …").
const WORKER_WORDS: &str = "embed worker";
impl Feed for Embeddings {
fn name(&self) -> &'static str {
"embeddings"
}
/// A sha skipped for want of hardware is due again once there is some;
/// one that failed on an unreachable worker, once the worker answers.
fn retry(&self, _at: &FeedAt<'_>, last: &FeedRow) -> bool {
self.hardware_came(last) || self.worker_came_back(last)
}
/// The whole feed in one call — what a caller with no place to run
/// deferred work would get. The consumer calls [`Feed::prepare`] instead.
fn run(&self, at: &FeedAt<'_>) -> Result<FeedResult> {
if let Some(skipped) = self.not_on_this_box(at.workspace) {
return Ok(skipped);
}
with_ring(self.embed(at), &self.sink, at.workspace, at.log)
}
fn prepare(&self, at: &FeedAt<'_>) -> Option<Result<DeferredWork>> {
// a build without an embedder, and a box without hardware for an
// unattended embed, have nothing long to defer: `run` says so
(crate::monitor::no_embedder_reason().is_none() && (self.no_hardware)().is_none()).then(|| -> Result<DeferredWork> {
// read BOTH halves out of the tree while the run holds it still
let functions = read_functions(at)?;
let inputs = crate::monitor::read_embed_inputs(&self.sink.git_dir, std::slice::from_ref(&at.member.name), at.workspace)?;
let (sink, member, workspace, log) = (self.sink.clone(), at.member.name.clone(), at.workspace.to_string(), at.log.clone());
let work: DeferredWork = Box::new(move || {
let ran: Result<FeedResult> = (|| {
// one embedding run at a time in this process
#[cfg(feature = "vector")]
let _turn = crate::vector::fnvec::embed_turn();
let functions = match embed_functions(&sink, functions)? {
Ok(line) => line,
Err(why) => return Ok(FeedResult::Skipped(why)),
};
Self::result(functions, crate::monitor::embed_read_inputs(&sink.warehouse, &workspace, inputs), &member, &workspace)
})();
with_ring(ran, &sink, &workspace, &log)
});
Ok(work)
})
}
}
/// The function half in one call: `Ok((new vectors, line))`, or `Err(why)`
/// when this build cannot embed (the feed's `skipped`).
#[cfg(feature = "vector")]
fn function_vectors(sink: &Sink, at: &FeedAt<'_>) -> Result<std::result::Result<(usize, String), String>> {
use crate::vector::fnvec::room::{Outcome, embed_member};
Ok(match embed_member(&sink.warehouse, at.log, &at.member.name, &at.member.source, at.seq, at.sha)? {
Outcome::Skipped(why) => Err(why),
Outcome::Ran(run) => Ok((run.new_vectors, run.line())),
})
}
/// What the function half read out of the tree for a deferred embed.
#[cfg(feature = "vector")]
type FunctionsRead = crate::vector::fnvec::room::MemberRead;
#[cfg(not(feature = "vector"))]
type FunctionsRead = ();
/// The function half's READ: the member's function texts at this commit, out
/// of the tree the run holds still. No model is loaded.
#[cfg(feature = "vector")]
fn read_functions(at: &FeedAt<'_>) -> Result<FunctionsRead> {
crate::vector::fnvec::room::read_member(at.log, &at.member.name, &at.member.source, at.seq, at.sha)
}
/// The function half's EMBED, over what [`read_functions`] read; the tree is
/// not looked at again.
#[cfg(feature = "vector")]
fn embed_functions(sink: &Sink, read: FunctionsRead) -> Result<std::result::Result<(usize, String), String>> {
use crate::vector::fnvec::room::{Outcome, embed_read};
Ok(match embed_read(&sink.warehouse, read)? {
Outcome::Skipped(why) => Err(why),
Outcome::Ran(run) => Ok((run.new_vectors, run.line())),
})
}
/// Start the dependency ring's embed job when the ring moved and this process
/// has a job ledger (a server). The words to append to the feed's detail.
#[cfg(feature = "vector")]
fn ring_job(sink: &Sink, workspace: &str, log: &EventLog) -> &'static str {
use crate::vector::fnvec::ring::{Host, Kick, kick};
let Some(jobs) = &sink.jobs else { return "" };
let host = Host {
workspace: workspace.to_string(),
warehouse: sink.warehouse.clone(),
log: log.clone(),
clones_root: super::Layout::under(sink.warehouse.root()).base().join("clones"),
jobs: Some(jobs.clone()),
queue: sink.ring_queue.clone(),
};
match kick(host) {
Kick::Started => "; dependency ring: background job started",
Kick::Queued => "; dependency ring: queued as a jera job",
_ => "",
}
}
#[cfg(not(feature = "vector"))]
fn ring_job(_: &Sink, _: &str, _: &EventLog) -> &'static str {
""
}
/// Without the `vector` feature there is no vector store in this binary.
#[cfg(not(feature = "vector"))]
fn function_vectors(_: &Sink, _: &FeedAt<'_>) -> Result<std::result::Result<(usize, String), String>> {
Ok(Err("built without the `vector` feature".into()))
}
#[cfg(not(feature = "vector"))]
fn read_functions(_: &FeedAt<'_>) -> Result<FunctionsRead> {
Ok(())
}
#[cfg(not(feature = "vector"))]
fn embed_functions(_: &Sink, _: FunctionsRead) -> Result<std::result::Result<(usize, String), String>> {
Ok(Err("built without the `vector` feature".into()))
}
#[cfg(test)]
mod tests {
use super::*;
/// The feed asks the hardware check BEFORE it reads or embeds anything:
/// with a reason it is `skipped`, keeps the reason whole and names the
/// verb that embeds anyway; without one it is not skipped here.
#[test]
fn the_embeddings_feed_does_not_embed_unattended_without_hardware() {
let dir = tempfile::tempdir().unwrap();
let sink = Sink {
warehouse: Arc::new(IcebergWarehouse::open(&dir.path().join("builds")).unwrap()),
index: Arc::new(crate::index::Index::open_at(dir.path(), &dir.path().join("index")).unwrap()),
git_dir: dir.path().join("git"),
index_dir: dir.path().join("index"),
members: vec!["alpha".into()],
jobs: None,
ring_queue: None,
};
let apu = Embeddings { sink: sink.clone(), worker_up: || false, no_hardware: || Some("no embedding hardware here: the AMD GPU has 512 MB of dedicated VRAM and the model needs 917 MB".into()) };
let Some(FeedResult::Skipped(why)) = apu.not_on_this_box("nordisk") else { panic!("must be skipped") };
assert_eq!(
why,
"no embedding hardware here: the AMD GPU has 512 MB of dedicated VRAM and the model needs 917 MB. Not embedded unattended: `nornir workspace embed nordisk` embeds now, on what is here"
);
let gpu = Embeddings { sink, no_hardware: || None, worker_up: || false };
assert!(gpu.not_on_this_box("nordisk").is_none(), "with hardware the feed goes on to embed");
}
}