src/clone_signals/backfill.rs
//! **Backfill `embeddings_groups` for history consumed before the group-twin
//! lens could embed** — the v2 half of [`crate::mimir_groupclone`]'s shipping.
//!
//! The `clone_shapes` feed's deferred half ([`super::feed::embed_groups`])
//! embeds the semantic groups of every commit it consumes — from the day the
//! feed gained that half. Everything consumed BEFORE that day has functions in
//! the lake and no group vectors in the warehouse, so the `group_twins` lens
//! reports those repos with `groups_without_vector > 0` forever: the pairs are
//! counted, never judged. This sweep closes that, through the SAME seam: per
//! member, the function chunks at its CONSUMED offset, handed to
//! `embed_groups`, which is content-addressed — a body that already has a
//! vector in this model space costs nothing, so the sweep is idempotent and a
//! re-run after a crash finishes the remainder.
//!
//! Error-isolated per member, the `lake_readfast` discipline: one member whose
//! checkout is gone (or whose room will not build) lands in
//! [`GroupsBackfillSweep::failed`] with its reason and NEVER starves the
//! members behind it.
//!
//! The chunk reader is INJECTED (production:
//! [`super::feed::read_chunks`] over the workspace's lake) so the sweep's
//! behaviour — fills what is missing, isolates failures, finds nothing twice —
//! is pinned by tests that need no lake.
use std::path::{Path, PathBuf};
use anyhow::Result;
use super::feed::NormRun;
use crate::vector::fnvec::{self, FnChunk};
use crate::warehouse::iceberg::IcebergWarehouse;
/// One member the sweep embedded for.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct MemberFill {
pub repo: String,
/// The consumed offset the chunks were read at.
pub seq: i64,
/// Function chunks read.
pub functions: usize,
/// Distinct group bodies ≥ the weight floor that had NO vector.
pub distinct_missing: usize,
/// Vectors written (≤ `distinct_missing`: an empty pool writes nothing).
pub new_vectors: usize,
pub embed_ms: u64,
pub ms: u64,
}
impl std::fmt::Display for MemberFill {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
write!(
f,
"groups backfill {}@{}: {} function(s), {} group bodies missing, {} vector(s) embedded in {:.1}s ({:.1}s total)",
self.repo,
self.seq,
self.functions,
self.distinct_missing,
self.new_vectors,
self.embed_ms as f64 / 1000.0,
self.ms as f64 / 1000.0
)
}
}
/// What one sweep did, member by member — written so a job row can account for
/// EVERY member: filled, already full, or failed with why.
#[derive(Debug, Default)]
pub struct GroupsBackfillSweep {
/// Members that had missing group vectors, now embedded.
pub filled: Vec<MemberFill>,
/// Members whose groups were ALL present already in this model space.
pub full: usize,
/// `(member, why)` — a failed member never stops the members behind it.
pub failed: Vec<(String, String)>,
}
impl GroupsBackfillSweep {
pub fn line(&self) -> String {
format!(
"{} member(s) filled ({} vector(s)), {} already full, {} failed",
self.filled.len(),
self.filled.iter().map(|m| m.new_vectors).sum::<usize>(),
self.full,
self.failed.len()
)
}
}
/// **The sweep.** `members` is `(name, checkout, consumed seq)` — only members
/// the lake has consumed belong here (the caller reads the offsets). `read`
/// hands back one member's function chunks at that offset (production:
/// [`super::feed::read_chunks`]); `embedding` is the SAME embedder the
/// `clone_shapes` feed uses, so the vectors land in the same model space.
pub fn backfill_groups(
wh: &IcebergWarehouse,
members: &[(String, PathBuf, i64)],
read: &mut dyn FnMut(&str, &Path, i64) -> Result<Vec<FnChunk>>,
embedding: &mut dyn fnvec::Embedding,
) -> GroupsBackfillSweep {
let mut sweep = GroupsBackfillSweep::default();
for (name, checkout, seq) in members {
let t0 = std::time::Instant::now();
let run: Result<(usize, NormRun)> = (|| {
let chunks = read(name, checkout, *seq)?;
let run = super::feed::embed_groups(wh, &chunks, embedding)?;
Ok((chunks.len(), run))
})();
match run {
Ok((_, r)) if r.distinct == 0 => sweep.full += 1,
Ok((functions, r)) => sweep.filled.push(MemberFill {
repo: name.clone(),
seq: *seq,
functions,
distinct_missing: r.distinct,
new_vectors: r.new_vectors,
embed_ms: r.embed_ms,
ms: t0.elapsed().as_millis() as u64,
}),
Err(e) => sweep.failed.push((name.clone(), format!("{e:#}"))),
}
}
sweep
}
#[cfg(test)]
#[cfg(feature = "code-room")] // the Bag test embedder lives with the room fixtures
mod tests {
use std::collections::HashSet;
use super::*;
use crate::warehouse::iceberg::TABLE_EMBEDDINGS_GROUPS;
fn chunk(file: &str, text: &str) -> FnChunk {
FnChunk {
node: crate::vector::fnvec::FnNode {
moniker: format!("t/{file}"),
file: file.to_string(),
start_line: 1,
end_line: text.lines().count() as u32,
},
content_hash: crate::vector::chunk::hash_text(text),
text: text.to_string(),
}
}
/// The group-twins lens fixtures: two functions whose big def-use group
/// clears the weight floor.
fn file_a() -> String {
"pub fn summarise(input: &[String]) -> u64 {\n let mut acc: Vec<(String, u64)> = Vec::new();\n let mut carry: u64 = 7;\n for raw in input {\n let w = raw.len() as u64 + carry;\n carry = carry.wrapping_mul(31) ^ w;\n acc.push((raw.clone(), w));\n }\n acc.sort_by(|x, y| y.1.cmp(&x.1));\n acc.len() as u64\n}\n".into()
}
fn file_b() -> String {
"pub fn roster(names: &[String]) -> Option<String> {\n let mut seen = std::collections::BTreeMap::new();\n for name in names {\n let trimmed = name.trim().to_string();\n if trimmed.is_empty() {\n continue;\n }\n *seen.entry(trimmed).or_insert(0usize) += 1;\n }\n let busiest = seen.iter().max_by_key(|(_, n)| **n).map(|(k, _)| k.clone());\n busiest\n}\n".into()
}
/// The exact keys `embed_groups` would store for `text`.
fn group_keys(text: &str) -> HashSet<String> {
crate::mimir_groupclone::segment(text)
.unwrap()
.into_iter()
.filter(|g| g.norm.tokens >= crate::mimir_groupclone::MIN_GROUP_TOKENS)
.map(|g| g.norm.exact)
.collect()
}
/// RED FIRST (2026-10-20): written against a `backfill_groups` that did
/// nothing — it failed on `filled.len()` — then the sweep was implemented.
///
/// Pins the three behaviours the job relies on:
/// 1. missing `embeddings_groups` rows of ALREADY-CONSUMED members are
/// filled (the vectors are readable back, in the embedder's space);
/// 2. a member that fails is REPORTED and never starves the members
/// behind it (`beta` fills although `broken` sits before it);
/// 3. a second sweep is content-addressed: nothing is embedded again and
/// the members count as `full`.
#[test]
fn the_sweep_fills_missing_rows_isolates_failures_and_finds_nothing_twice() {
let tmp = tempfile::tempdir().unwrap();
let wh = IcebergWarehouse::open(tmp.path()).unwrap();
let bag = crate::mimir_lens::room_feed::fixture::vectors::Bag::default();
let mut embedding = crate::vector::fnvec::Loaded { embedder: &bag, backend: "test bag (cpu)" };
let members = vec![
("alpha".to_string(), PathBuf::from("/nowhere/alpha"), 3i64),
("broken".to_string(), PathBuf::from("/nowhere/broken"), 1i64),
("beta".to_string(), PathBuf::from("/nowhere/beta"), 7i64),
];
let mut read = |name: &str, _checkout: &Path, seq: i64| -> Result<Vec<FnChunk>> {
match name {
"alpha" => {
assert_eq!(seq, 3, "the sweep reads at the CONSUMED offset");
Ok(vec![chunk("src/a.rs", &file_a())])
}
"beta" => Ok(vec![chunk("src/b.rs", &file_b())]),
other => anyhow::bail!("{other}: the checkout is gone (test)"),
}
};
let sweep = backfill_groups(&wh, &members, &mut read, &mut embedding);
assert_eq!(sweep.filled.len(), 2, "alpha and beta fill although broken failed between them: {sweep:?}");
assert_eq!(sweep.filled[0].repo, "alpha");
assert_eq!(sweep.filled[1].repo, "beta", "the failure is ISOLATED: beta still ran: {sweep:?}");
assert!(sweep.filled.iter().all(|m| m.new_vectors > 0), "{sweep:?}");
assert_eq!(sweep.failed.len(), 1, "{sweep:?}");
assert!(sweep.failed[0].0 == "broken" && sweep.failed[0].1.contains("checkout is gone"), "{sweep:?}");
// the rows are really there, in the embedder's model space
let profile = crate::vector::store::Embedder::profile(&bag).id();
let keys = group_keys(&file_a());
assert!(!keys.is_empty(), "the fixture must carry at least one group over the floor");
let refs: HashSet<&str> = keys.iter().map(|s| s.as_str()).collect();
let (vectors, _) = wh
.block_on(crate::vector::store::read_vectors_present_in(&wh, TABLE_EMBEDDINGS_GROUPS, &profile, &refs))
.unwrap();
assert_eq!(vectors.len(), keys.len(), "every missing group body of alpha got its vector");
// the second sweep finds nothing to do — content-addressed, idempotent
let again = backfill_groups(&wh, &members, &mut read, &mut embedding);
assert_eq!(again.filled.len(), 0, "{again:?}");
assert_eq!(again.full, 2, "{again:?}");
assert_eq!(again.failed.len(), 1, "broken is still said, not swallowed: {again:?}");
}
}