diff --git a/crates/eureka-state/tests/cli.rs b/crates/eureka-state/tests/cli.rs index f2feb80..2a4e229 100644 --- a/crates/eureka-state/tests/cli.rs +++ b/crates/eureka-state/tests/cli.rs @@ -19,7 +19,10 @@ use huginn_mind::{ AdmissionFacts, HuginnMindRequest, HuginnMindResponse, IndexStatus, MindRefusal, MindStatus, PersonaStatus, PipelineAdmissionOutcome, PipelineDocumentView, PipelinePageItems, PipelineProvenance, PipelineSelectionPage, PipelineStatus, }; +use huginn_daemon::{Daemon, Hits, IndexSink, SearchTicket}; +use huginn_mind::MindStore; use serde_json::{Value, json}; +use std::time::Duration; use tempfile::TempDir; mod common; @@ -1046,3 +1049,160 @@ fn a_conflicting_batch_is_the_minds_no() { assert_eq!(ran.code, 1); assert_eq!(ran.out, format!("Conflict\nconflicts {CAMPAIGN}:campaign:self\n")); } + +/// The chunks a source index holds, served from memory: each text embeds to one +/// vector and each held chunk scores the same, so the search is the daemon's own +/// path (`sources::search`, its identity check and its citation fields) with +/// nothing about ranking in it. +struct Desk { + built_under: &'static str, + embeds_under: &'static str, + held: Vec, + tickets: u64, + done: Vec<(SearchTicket, Result)>, +} + +/// An embedder that sends every text to one vector, under a digest. +struct Flat(&'static str); + +impl huginn_daemon::index::Embedder for Flat { + fn model_identity(&mut self, _within: Duration) -> anyhow::Result { + Ok(huginn_daemon::index::ModelIdentity { name: "flat".into(), digest: self.0.into(), dimensions: 2 }) + } + + fn embed(&mut self, texts: &[String], _within: Duration) -> anyhow::Result>> { + Ok(texts.iter().map(|_| vec![1.0, 0.0]).collect()) + } +} + +impl huginn_daemon::sources::SourceIndex for Desk { + fn describe_sources(&mut self, _within: Duration) -> anyhow::Result { + Ok(huginn_daemon::sources::DescribedSources::Labelled { + meta: huginn_daemon::sources::SourcesMeta { + managed_by: "huginn".into(), + corpus: huginn_daemon::sources::CORPUS.into(), + model: "flat".into(), + model_digest: self.built_under.into(), + dimensions: 2, + chunker_version: huginn_daemon::sources::chunk::CHUNKER_VERSION, + }, + cursors: Default::default(), + }) + } + + fn recreate_sources(&mut self, _meta: &huginn_daemon::sources::SourcesMeta) -> anyhow::Result<()> { + anyhow::bail!("a daemon never writes the source index") + } + + fn set_cursors(&mut self, _cursors: &std::collections::BTreeMap) -> anyhow::Result<()> { + anyhow::bail!("a daemon never writes the source index") + } + + fn delete_by(&mut self, _repo: &str, _paths: Option<&[String]>) -> anyhow::Result<()> { + anyhow::bail!("a daemon never writes the source index") + } + + fn upsert_sources(&mut self, _points: &[huginn_daemon::sources::SourcePoint]) -> anyhow::Result<()> { + anyhow::bail!("a daemon never writes the source index") + } + + fn search_sources( + &mut self, + _vector: &[f32], + repo: Option<&str>, + limit: u32, + _within: Duration, + ) -> anyhow::Result> { + Ok(self + .held + .iter() + .filter(|payload| repo.is_none_or(|repo| payload.repo == repo)) + .take(limit as usize) + .map(|payload| huginn_daemon::sources::SourceRow { score: 0.9, payload: payload.clone() }) + .collect()) + } +} + +impl IndexSink for Desk { + fn committed(&mut self, _mind: &huginn_mind::Mind, _writes: &[PipelineRef]) -> anyhow::Result<()> { + Ok(()) + } + fn status(&self) -> IndexStatus { + IndexStatus::Current + } + fn search(&mut self, _text: &str, _top_k: u32, _among: Vec) -> Result { + Err("this index holds sources only".into()) + } + fn search_sources(&mut self, query: &huginn_mind::SourceQuery) -> Result { + self.tickets += 1; + let ticket = SearchTicket(self.tickets); + let mut embedder = Flat(self.embeds_under); + let found = huginn_daemon::sources::search(&mut embedder, self, query, Duration::from_secs(5)).map_err(|error| format!("{error:#}")); + self.done.push((ticket, found)); + Ok(ticket) + } + fn sources_searched(&mut self) -> Vec<(SearchTicket, Result)> { + std::mem::take(&mut self.done) + } + fn search_deadline(&self) -> Duration { + Duration::from_secs(5) + } + fn searched(&mut self) -> Vec<(SearchTicket, Result)> { + Vec::new() + } + fn abandon(&mut self, _ticket: SearchTicket) {} +} + +/// A daemon over a temporary mind whose source index was built under +/// `built_under` and whose embedder reports `embeds_under`. +fn source_daemon(built_under: &'static str, embeds_under: &'static str) -> Server { + let chunk = |repo: &str, path: &str, line_start: u32, line_end: u32, text: &str| huginn_daemon::sources::SourcePayload { + repo: repo.into(), + path: path.into(), + commit: "abcdef1234567890".into(), + blob: "b".into(), + line_start, + line_end, + chunk_index: 0, + ext: "rs".into(), + text: text.into(), + }; + let desk = Desk { + built_under, + embeds_under, + held: vec![ + chunk("Demo", "src/lib.rs", 3, 9, "fn cursor() {}"), + chunk("Demo", "README.md", 1, 2, "# Demo"), + chunk("Other", "x.rs", 1, 1, "fn other() {}"), + ], + tickets: 0, + done: Vec::new(), + }; + let root = tempfile::tempdir().unwrap(); + let mut daemon = Daemon::new(huginn_mind::Mind::open(root.path(), &slug(INSTANCE)).unwrap(), desk); + let admitted = answered(daemon.handle(HuginnMindRequest::Admit(batch(INSTANCE, vec![identity(INSTANCE)])), chrono::Utc::now())); + assert!(matches!(admitted, HuginnMindResponse::Admit(PipelineAdmissionOutcome::Committed { .. })), "{admitted:?}"); + serve(root, daemon) +} + +/// `huginn sources` end to end: the binary asks a daemon over CultNet, the +/// daemon answers from its source index through the worker path, and the +/// citation is printed for the chunk the filter and limit chose. +#[test] +fn sources_prints_a_citation_for_a_search_answered_by_a_daemon() { + let server = source_daemon("d1", "d1"); + let found = at(server.addr, &["sources", "where is the cursor kept", "--repo", "Demo", "--top-k", "1"], None); + assert_eq!(found.code, 0, "{}", found.err); + assert!(found.out.starts_with("Demo:src/lib.rs:3-9 @abcdef1 0.900\nfn cursor() {}"), "{}", found.out); + assert!(!found.out.contains("README") && !found.out.contains("Other"), "{}", found.out); +} + +/// A daemon whose embedder is not the model the index was built under refuses +/// the search naming both, and the CLI exits 1 with that word. +#[test] +fn sources_refuses_naming_both_models_when_the_daemon_embeds_under_another() { + let server = source_daemon("d1", "d2"); + let refused = at(server.addr, &["sources", "anything"], None); + assert_eq!(refused.code, 1, "{}", refused.err); + assert!(refused.out.contains("d1") && refused.out.contains("d2"), "{}", refused.out); +} diff --git a/crates/eureka-state/tests/common/mod.rs b/crates/eureka-state/tests/common/mod.rs index eb1ca0d..0c0da62 100644 --- a/crates/eureka-state/tests/common/mod.rs +++ b/crates/eureka-state/tests/common/mod.rs @@ -148,7 +148,10 @@ impl Drop for Server { } } -pub fn serve(root: TempDir, mut daemon: TestDaemon) -> Server { +pub fn serve + Send + 'static>( + root: TempDir, + mut daemon: Daemon, +) -> Server { let mut hub = bind("127.0.0.1:0".parse().unwrap(), &daemon.runtime_id()).unwrap(); let addr = hub.local_addr().unwrap(); let stopping = Arc::new(AtomicBool::new(false)); diff --git a/crates/huginn-daemon/src/bin/huginn-sources.rs b/crates/huginn-daemon/src/bin/huginn-sources.rs index 63115f7..fc5a291 100644 --- a/crates/huginn-daemon/src/bin/huginn-sources.rs +++ b/crates/huginn-daemon/src/bin/huginn-sources.rs @@ -81,9 +81,14 @@ fn main() -> ExitCode { let mut forge = HttpForge::new(&options.forge_url); let mut journal = |report: &huginn_daemon::sources::Report| println!("{report}"); match refresh(&mut embedder, &mut index, &mut forge, &options.state_root, &mut journal) { - Ok(refreshed) if refreshed.failed.is_empty() => ExitCode::SUCCESS, + Ok(refreshed) if refreshed.is_clean() => ExitCode::SUCCESS, Ok(refreshed) => { - eprintln!("huginn-sources: {} repos failed: {}", refreshed.failed.len(), refreshed.failed.join(", ")); + if !refreshed.failed.is_empty() { + eprintln!("huginn-sources: {} repos failed: {}", refreshed.failed.len(), refreshed.failed.join(", ")); + } + if let Some(why) = &refreshed.retirement_withheld { + eprintln!("huginn-sources: no repo was retired: {why}"); + } ExitCode::from(1) } Err(error) => { diff --git a/crates/huginn-daemon/src/index/fakes.rs b/crates/huginn-daemon/src/index/fakes.rs index 0245d46..bf8833f 100644 --- a/crates/huginn-daemon/src/index/fakes.rs +++ b/crates/huginn-daemon/src/index/fakes.rs @@ -258,6 +258,8 @@ pub(crate) struct SourcesState { pub(crate) fail_upserts_after: Option, pub(crate) cursor_writes: Vec>, pub(crate) deletes: Vec<(String, Option>)>, + /// A delete naming this repo refuses. + pub(crate) fail_deletes_of: Option, /// The repo filter and limit of every search, in order. pub(crate) searches: Vec<(Option, u32)>, } @@ -327,6 +329,9 @@ impl SourceIndex for FakeSources { fn delete_by(&mut self, repo: &str, paths: Option<&[String]>) -> Result<()> { let mut state = self.state.lock().unwrap(); + if state.fail_deletes_of.as_deref() == Some(repo) { + bail!("the vector store refused the delete"); + } state.deletes.push((repo.to_owned(), paths.map(<[String]>::to_vec))); state.points.retain(|_, point| { !(point.payload.repo == repo && paths.is_none_or(|paths| paths.contains(&point.payload.path))) diff --git a/crates/huginn-daemon/src/sources/forge.rs b/crates/huginn-daemon/src/sources/forge.rs index 60cc24f..b9f8914 100644 --- a/crates/huginn-daemon/src/sources/forge.rs +++ b/crates/huginn-daemon/src/sources/forge.rs @@ -6,8 +6,7 @@ use std::time::Duration; use anyhow::{Context, Result, ensure}; use serde_json::Value; -use super::{Forge, SourceRepo}; -use crate::index::{Http, Method}; +use super::{Catalog, Forge, SourceRepo}; /// The organisation whose repositories are the corpus. pub const ORG: &str = "GameCult"; @@ -19,13 +18,38 @@ const CALL_TIMEOUT: Duration = Duration::from_secs(60); /// API, the unauthenticated view of it: what an anonymous reader may see is /// what is indexed. pub struct HttpForge { - http: Http, + agent: ureq::Agent, base: String, } impl HttpForge { pub fn new(base_url: &str) -> Self { - Self { http: Http::new(base_url, CALL_TIMEOUT), base: base_url.trim_end_matches('/').to_owned() } + let agent: ureq::Agent = + ureq::Agent::config_builder().timeout_global(Some(CALL_TIMEOUT)).http_status_as_error(false).build().into(); + Self { agent, base: base_url.trim_end_matches('/').to_owned() } + } + + /// One GET: the status, the count of repos the forge says it holds + /// (`X-Total-Count`, which only a header carries, so this does not go + /// through `index::Http`), and the body. + fn page(&self, path: &str) -> Result<(u16, Option, String)> { + let url = format!("{}{path}", self.base); + let mut response = self.agent.get(&url).call().with_context(|| format!("no answer from {url}"))?; + let status = response.status().as_u16(); + let total = response.headers().get("x-total-count").and_then(|value| value.to_str().ok()).and_then(|value| value.trim().parse().ok()); + let body = response.body_mut().read_to_string().with_context(|| format!("reading the answer from {url}"))?; + Ok((status, total, body)) + } +} + +/// Why the pages listed are not the whole catalog, or `None` when the repos +/// listed (before any are filtered out) are the count the forge states. A forge +/// that states no count cannot vouch for its list. +pub(crate) fn shortfall(listed: usize, stated: Option) -> Option { + match stated { + None => Some("the forge stated no X-Total-Count".to_owned()), + Some(total) if total == listed => None, + Some(total) => Some(format!("the forge counts {total} repos and {listed} were listed")), } } @@ -45,20 +69,23 @@ pub(crate) fn indexable(page: &Value) -> Result> { } impl Forge for HttpForge { - fn catalog(&mut self) -> Result> { + fn catalog(&mut self) -> Result { let mut found = Vec::new(); + let (mut listed_in_all, mut stated) = (0, None); for page in 1.. { let path = format!("/api/v1/orgs/{ORG}/repos?page={page}&limit={PAGE}"); - let (status, answer) = self.http.request(Method::Get, &path, None)?; + let (status, total, answer) = self.page(&path)?; ensure!(status == 200, "the forge answered {status} to {path}"); + stated = stated.or(total); let parsed: Value = serde_json::from_str(&answer).context("the forge's repo page is not JSON")?; let listed = parsed.as_array().map_or(0, Vec::len); + listed_in_all += listed; found.extend(indexable(&parsed)?); if listed < PAGE { break; } } - Ok(found) + Ok(Catalog { repos: found, shortfall: shortfall(listed_in_all, stated) }) } fn clone_url(&self, name: &str) -> String { @@ -91,8 +118,9 @@ mod tests { assert_eq!(names, ["Open"]); } - /// Serves each page in `pages` once, to the request that names it. - fn serve(pages: Vec) -> String { + /// Serves each page in `pages` once, to the request that names it, with + /// `total` as its `X-Total-Count` when given. + fn serve(pages: Vec, total: Option) -> String { use std::io::{Read, Write}; let listener = std::net::TcpListener::bind("127.0.0.1:0").unwrap(); let address = listener.local_addr().unwrap(); @@ -103,25 +131,48 @@ mod tests { let read = stream.read(&mut buffer).unwrap(); let request = String::from_utf8_lossy(&buffer[..read]).into_owned(); let body = if request.contains("page=1&") { &pages[0] } else { &pages[1] }; - write!(stream, "HTTP/1.1 200 OK\r\ncontent-type: application/json\r\ncontent-length: {}\r\nconnection: close\r\n\r\n{body}", body.len()).unwrap(); + let count = total.map_or_else(String::new, |total| format!("x-total-count: {total}\r\n")); + write!(stream, "HTTP/1.1 200 OK\r\ncontent-type: application/json\r\n{count}content-length: {}\r\nconnection: close\r\n\r\n{body}", body.len()).unwrap(); } }); format!("http://{address}") } + fn page_of(range: std::ops::Range) -> String { + let repos: Vec = range.map(|n| json!({ "name": format!("r{n}"), "default_branch": "main", "private": false, "archived": false, "mirror": false, "empty": false })).collect(); + Value::Array(repos).to_string() + } + /// The catalog reads every page until one is short: a full first page /// does not end it, and a short second page does; clone URLs name the org. #[test] fn the_catalog_reads_pages_until_a_short_one_and_clones_from_the_org() { - let page = |range: std::ops::Range| { - let repos: Vec = range.map(|n| json!({ "name": format!("r{n}"), "default_branch": "main", "private": false, "archived": false, "mirror": false, "empty": false })).collect(); - Value::Array(repos).to_string() - }; - let base = serve(vec![page(0..PAGE), page(PAGE..PAGE + 3)]); + let base = serve(vec![page_of(0..PAGE), page_of(PAGE..PAGE + 3)], Some(PAGE + 3)); let mut forge = HttpForge::new(&format!("{base}/")); - let names: Vec = forge.catalog().unwrap().into_iter().map(|repo| repo.name).collect(); + let catalog = forge.catalog().unwrap(); + let names: Vec = catalog.repos.into_iter().map(|repo| repo.name).collect(); assert_eq!(names.len(), PAGE + 3); assert_eq!((names[0].as_str(), names[PAGE + 2].as_str()), ("r0", "r52")); + assert_eq!(catalog.shortfall, None, "the pages sum to the forge's count"); assert_eq!(forge.clone_url("Demo"), format!("{base}/GameCult/Demo.git")); } + + /// A catalog whose pages do not sum to the count the forge states is + /// marked short: a forge that caps a page below ours stops the paging at + /// the first short page, and an empty answer or a missing count vouches for + /// nothing. + #[test] + fn a_catalog_that_does_not_sum_to_the_forges_count_is_short() { + let truncated = serve(vec![page_of(0..30)], Some(62)); + let catalog = HttpForge::new(&truncated).catalog().unwrap(); + assert_eq!(catalog.repos.len(), 30); + assert!(catalog.shortfall.as_deref().unwrap().contains("counts 62 repos and 30 were listed")); + + let empty = serve(vec!["[]".into()], Some(62)); + let catalog = HttpForge::new(&empty).catalog().unwrap(); + assert!(catalog.repos.is_empty() && catalog.shortfall.is_some()); + + let unstated = serve(vec![page_of(0..3)], None); + assert!(HttpForge::new(&unstated).catalog().unwrap().shortfall.unwrap().contains("X-Total-Count")); + } } diff --git a/crates/huginn-daemon/src/sources/mod.rs b/crates/huginn-daemon/src/sources/mod.rs index 4d30f16..0a45872 100644 --- a/crates/huginn-daemon/src/sources/mod.rs +++ b/crates/huginn-daemon/src/sources/mod.rs @@ -13,6 +13,8 @@ //! lives in the collection's own metadata, so it is lost with the collection //! and the two cannot disagree about what is held. It is advanced only after //! the repo's deletes and upserts are acknowledged: it may lag and never leads. +//! It is dropped before the repo's first delete, so a run that fails between +//! the delete and the upserts leaves no cursor, and the repo is ingested whole. //! The cursors object is written whole by this job alone, once per repo. pub mod chunk; @@ -129,9 +131,19 @@ pub struct SourceRepo { pub default_branch: String, } +/// The repos the forge listed as indexable, and whether the list is the whole +/// of what the forge holds. A list that is not whole cannot say which repos +/// are gone. +#[derive(Clone, Debug, PartialEq, Eq)] +pub struct Catalog { + pub repos: Vec, + /// Why the list may be missing repos; `None` when it is the forge's count. + pub shortfall: Option, +} + /// The forge: which repos exist, and where to fetch them from. pub trait Forge { - fn catalog(&mut self) -> Result>; + fn catalog(&mut self) -> Result; fn clone_url(&self, name: &str) -> String; } @@ -178,10 +190,20 @@ fn extension(path: &str) -> String { name.rsplit_once('.').map(|(_, extension)| extension.to_ascii_lowercase()).unwrap_or_default() } -/// The names the refresh reports as failed; empty when every repo converged. +/// What a refresh reports: the names that failed, and why nothing was retired +/// if the catalog could not be trusted to say which repos are gone. #[derive(Debug, Default)] pub struct Refreshed { pub failed: Vec, + /// Set when retirement was skipped because the catalog was empty or short. + pub retirement_withheld: Option, +} + +impl Refreshed { + /// Every repo converged and the catalog could be trusted to retire from. + pub fn is_clean(&self) -> bool { + self.failed.is_empty() && self.retirement_withheld.is_none() + } } /// Brings the collection to the forge's current state. A collection that @@ -222,10 +244,24 @@ pub fn refresh( let catalog = forge.catalog().context("listing the forge's repos")?; let mirrors = mirrors_dir(state_root); - retire(index, &catalog, &mirrors, &mut cursors)?; - let mut refreshed = Refreshed::default(); - for repo in &catalog { + // Retirement reads the catalog as the complete list of what exists. An + // empty or short list would retire everything it fails to name, so then + // nothing is retired and the run says so. + let withheld = match (&catalog.shortfall, catalog.repos.is_empty()) { + (Some(why), _) => Some(format!("the forge's catalog is incomplete ({why})")), + (None, true) => Some("the forge's catalog lists no indexable repos".to_owned()), + (None, false) => None, + }; + match withheld { + Some(why) => { + eprintln!("huginn-sources: retiring nothing: {why}"); + refreshed.retirement_withheld = Some(why); + } + None => retire(index, &catalog.repos, &mirrors, &mut cursors, &mut refreshed.failed), + } + + for repo in &catalog.repos { let started = Instant::now(); match refresh_repo(embedder, index, forge, &mirrors, repo, &identity.dimensions, &mut cursors) { Ok(mut report) => { @@ -243,13 +279,17 @@ pub fn refresh( /// Drops every repo the catalog no longer lists: its chunks, its cursor and /// its mirror, in that order, so a failure leaves the cursor to find it again. +/// One repo's failure is logged and named in `failed`; the others go on. Names +/// are compared as written: a repo renamed only by case is a new repo to the +/// index (its points carry the old spelling), so the old one is retired. fn retire( index: &mut S, catalog: &[SourceRepo], mirrors: &Path, cursors: &mut BTreeMap, -) -> Result<()> { - let listed = |name: &str| catalog.iter().any(|repo| repo.name.eq_ignore_ascii_case(name)); + failed: &mut Vec, +) { + let listed = |name: &str| catalog.iter().any(|repo| repo.name == name); let mut gone: BTreeSet = cursors.keys().filter(|name| !listed(name)).cloned().collect(); if let Ok(entries) = std::fs::read_dir(mirrors) { for entry in entries.flatten() { @@ -261,15 +301,22 @@ fn retire( } } for name in gone { - index.delete_by(&name, None).with_context(|| format!("deleting the chunks of retired {name}"))?; - if cursors.remove(&name).is_some() { - index.set_cursors(cursors).context("dropping a retired repo's cursor")?; - } - let mirror = Mirror::path_for(mirrors, &name); - if mirror.exists() { - std::fs::remove_dir_all(&mirror).with_context(|| format!("removing {}", mirror.display()))?; + if let Err(error) = retire_one(index, &name, mirrors, cursors) { + eprintln!("huginn-sources: retiring {name} failed, the others go on: {error:#}"); + failed.push(name); } } +} + +fn retire_one(index: &mut S, name: &str, mirrors: &Path, cursors: &mut BTreeMap) -> Result<()> { + index.delete_by(name, None).with_context(|| format!("deleting the chunks of retired {name}"))?; + if cursors.remove(name).is_some() { + index.set_cursors(cursors).context("dropping a retired repo's cursor")?; + } + let mirror = Mirror::path_for(mirrors, name); + if mirror.exists() { + std::fs::remove_dir_all(&mirror).with_context(|| format!("removing {}", mirror.display()))?; + } Ok(()) } @@ -295,6 +342,14 @@ fn refresh_repo( Some(old) if mirror.is_ancestor(old, &head) => Some(mirror.changed_paths(old, &head)?.into_iter().collect()), _ => None, }; + // From the first delete on, the collection holds no single commit's chunks + // for this repo until the run finishes. Say so before changing anything: + // the cursor is dropped, so a run that dies half way leaves a repo the next + // run re-ingests whole, whatever head it finds then. A cursor left at the + // old commit would let a later head that reverts the files hide the hole. + if cursors.remove(&repo.name).is_some() { + index.set_cursors(cursors).context("dropping the cursor before the repo's chunks change")?; + } match &changed { Some(paths) => { let paths: Vec = paths.iter().cloned().collect(); diff --git a/crates/huginn-daemon/src/sources/tests.rs b/crates/huginn-daemon/src/sources/tests.rs index 9378559..698f828 100644 --- a/crates/huginn-daemon/src/sources/tests.rs +++ b/crates/huginn-daemon/src/sources/tests.rs @@ -58,6 +58,7 @@ impl Remote { struct FakeForge { repos: Vec<(SourceRepo, PathBuf)>, + shortfall: Option, } impl FakeForge { @@ -67,13 +68,14 @@ impl FakeForge { .iter() .map(|(name, remote)| (SourceRepo { name: (*name).into(), default_branch: "main".into() }, remote.path())) .collect(), + shortfall: None, } } } impl Forge for FakeForge { - fn catalog(&mut self) -> Result> { - Ok(self.repos.iter().map(|(repo, _)| repo.clone()).collect()) + fn catalog(&mut self) -> Result { + Ok(Catalog { repos: self.repos.iter().map(|(repo, _)| repo.clone()).collect(), shortfall: self.shortfall.clone() }) } fn clone_url(&self, name: &str) -> String { @@ -182,7 +184,7 @@ fn unchanged_repo_makes_no_embed_call() { assert_eq!((reports[0].files, reports[0].chunks), (0, 0)); } -/// An upsert that fails mid-repo leaves the cursor on the old commit, and the +/// An upsert that fails mid-repo leaves no cursor (no commit is held whole), and the /// next run converges to the points a clean run of the same history makes. #[test] fn cursor_never_leads() { @@ -203,7 +205,8 @@ fn cursor_never_leads() { } let (refreshed, _) = rig.run(&mut forge); assert_eq!(refreshed.unwrap().failed, ["Demo"], "one repo failed, the run finished"); - assert_eq!(rig.index.cursors().get("Demo"), Some(&a), "the cursor never leads the points"); + assert_eq!(rig.index.cursors().get("Demo"), None, "no commit's chunks are held whole, so no cursor says one is"); + assert_ne!(a, b); rig.index.state.lock().unwrap().fail_upserts_after = None; rig.converged(&mut forge); @@ -214,6 +217,44 @@ fn cursor_never_leads() { assert_eq!(rig.index.ids(), clean.index.ids()); } +/// A run that fails between a repo's deletes and its upserts leaves chunks of +/// no single commit. A head that then reverts the files to the old content +/// makes an empty diff from the old commit, which would keep the deleted +/// chunks deleted; the repo is ingested whole instead. +#[test] +fn a_failed_run_then_a_revert_converges() { + let remote = Remote::new(); + let at_a: Vec<(String, Vec)> = (0..40).map(|n| (format!("f{n}.rs"), format!("file {n} +").into_bytes())).collect(); + let original: Vec<(&str, &[u8])> = at_a.iter().map(|(path, bytes)| (path.as_str(), bytes.as_slice())).collect(); + remote.commit(&original, &[]); + let mut rig = Rig::new(); + let mut forge = FakeForge::of(&[("Demo", &remote)]); + rig.converged(&mut forge); + + let at_b: Vec<(String, Vec)> = (0..40).map(|n| (format!("f{n}.rs"), format!("file {n} edited +").into_bytes())).collect(); + let edited: Vec<(&str, &[u8])> = at_b.iter().map(|(path, bytes)| (path.as_str(), bytes.as_slice())).collect(); + remote.commit(&edited, &[]); + { + let mut state = rig.index.state.lock().unwrap(); + state.fail_upserts_after = Some(state.upsert_calls + 1); + } + let (refreshed, _) = rig.run(&mut forge); + assert_eq!(refreshed.unwrap().failed, ["Demo"]); + assert!(rig.index.held().len() < 40, "the failed run left a partial repo"); + + rig.index.state.lock().unwrap().fail_upserts_after = None; + let c = remote.commit(&original, &[]); + rig.converged(&mut forge); + assert_eq!(rig.index.cursors().get("Demo"), Some(&c)); + let mut clean = Rig::new(); + clean.converged(&mut forge); + assert_eq!(rig.index.held().len(), 40, "every file of the reverted head is held"); + assert_eq!(rig.index.held(), clean.index.held()); + assert_eq!(rig.index.ids(), clean.index.ids()); +} + /// One repo failing is logged and the others go on. #[test] fn one_repos_failure_does_not_stop_the_others() { @@ -265,6 +306,88 @@ fn retired_repo_leaves_nothing() { assert!(!rig.mirror("Gone").exists() && rig.mirror("Keep").exists()); } +/// An empty catalog, and a catalog the forge says is short, retire nothing and +/// the run says why; the same forge answering whole retires what it dropped. +#[test] +fn an_empty_or_short_catalog_retires_nothing() { + let (keep, gone) = (Remote::new(), Remote::new()); + let k = keep.commit(&[("a.rs", b"a\n")], &[]); + gone.commit(&[("b.rs", b"b\n")], &[]); + let mut rig = Rig::new(); + rig.converged(&mut FakeForge::of(&[("Keep", &keep), ("Gone", &gone)])); + let both = rig.index.held(); + + let mut empty = FakeForge::of(&[]); + let (refreshed, reports) = rig.run(&mut empty); + let refreshed = refreshed.unwrap(); + assert!(reports.is_empty() && refreshed.retirement_withheld.as_deref().unwrap().contains("no indexable repos")); + assert_eq!(rig.index.held(), both, "an empty answer deleted nothing"); + assert!(rig.mirror("Gone").exists() && rig.mirror("Keep").exists()); + assert_eq!(rig.index.cursors().len(), 2); + + let mut short = FakeForge::of(&[("Keep", &keep)]); + short.shortfall = Some("the forge counts 2 repos and 1 were listed".into()); + let (refreshed, _) = rig.run(&mut short); + assert!(refreshed.unwrap().retirement_withheld.unwrap().contains("incomplete")); + assert_eq!(rig.index.held(), both, "a short answer deleted nothing"); + assert!(rig.mirror("Gone").exists()); + + let (refreshed, _) = rig.run(&mut FakeForge::of(&[("Keep", &keep)])); + assert_eq!(refreshed.unwrap().retirement_withheld, None); + assert_eq!(rig.index.held(), [held("Keep", "a.rs", 0, &k)], "a whole answer retires what the forge dropped"); +} + +/// A retirement that fails is that repo's failure alone: it is named, the +/// other retirements and the live repos go on, and the failed one keeps its +/// cursor and mirror so the next run finds it again. +#[test] +fn a_retire_error_skips_that_repo_and_the_run_goes_on() { + let (keep, first, second) = (Remote::new(), Remote::new(), Remote::new()); + let k = keep.commit(&[("a.rs", b"a\n")], &[]); + first.commit(&[("b.rs", b"b\n")], &[]); + second.commit(&[("c.rs", b"c\n")], &[]); + let mut rig = Rig::new(); + rig.converged(&mut FakeForge::of(&[("Keep", &keep), ("First", &first), ("Second", &second)])); + let edited = keep.commit(&[("a.rs", b"a edited\n")], &[]); + rig.index.state.lock().unwrap().fail_deletes_of = Some("First".into()); + + let (refreshed, reports) = rig.run(&mut FakeForge::of(&[("Keep", &keep)])); + assert_eq!(refreshed.unwrap().failed, ["First"]); + assert!(!rig.mirror("Second").exists() && rig.mirror("First").exists()); + assert_eq!(rig.index.cursors().into_keys().collect::>(), ["First", "Keep"]); + assert_eq!(reports.len(), 1, "the live repo was refreshed after the failed retirement"); + assert_ne!(k, edited); + + rig.index.state.lock().unwrap().fail_deletes_of = None; + rig.converged(&mut FakeForge::of(&[("Keep", &keep)])); + assert!(!rig.mirror("First").exists()); + assert_eq!(rig.index.cursors().into_keys().collect::>(), ["Keep"]); + assert_eq!(rig.index.held(), [held("Keep", "a.rs", 0, &edited)]); +} + +/// A repo the forge now spells with another case is another repo to the +/// index: the points under the old spelling do not stay behind. +#[test] +fn a_repo_renamed_only_by_case_keeps_no_stale_points() { + let remote = Remote::new(); + remote.commit(&[("a.rs", b"a\n")], &[]); + let mut rig = Rig::new(); + rig.converged(&mut FakeForge::of(&[("Demo", &remote)])); + let head = remote.commit(&[("b.rs", b"b\n")], &[]); + rig.converged(&mut FakeForge::of(&[("demo", &remote)])); + let repos: BTreeSet = rig.index.held().into_iter().map(|(repo, _, _, _)| repo).collect(); + assert_eq!(repos, BTreeSet::from(["demo".to_owned()])); + assert_eq!(rig.index.cursors().into_iter().collect::>(), [("demo".to_owned(), head)]); +} + +/// A run is clean only when no repo failed and retirement was not withheld. +#[test] +fn a_run_is_clean_only_when_nothing_failed_and_retirement_was_not_withheld() { + assert!(Refreshed::default().is_clean()); + assert!(!Refreshed { failed: vec!["Demo".into()], retirement_withheld: None }.is_clean()); + assert!(!Refreshed { failed: Vec::new(), retirement_withheld: Some("short".into()) }.is_clean()); +} + /// A mirror no cursor knows (a first ingest that never finished) is retired /// with its points too. #[test] diff --git a/crates/huginn-daemon/tests/live_index_smoke.rs b/crates/huginn-daemon/tests/live_index_smoke.rs index 8c2b7e6..a3d2274 100644 --- a/crates/huginn-daemon/tests/live_index_smoke.rs +++ b/crates/huginn-daemon/tests/live_index_smoke.rs @@ -141,8 +141,10 @@ struct OneRepo { } impl huginn_daemon::sources::Forge for OneRepo { - fn catalog(&mut self) -> anyhow::Result> { - Ok(self.inner.catalog()?.into_iter().filter(|repo| repo.name == self.name).collect()) + fn catalog(&mut self) -> anyhow::Result { + // One repo of the forge on purpose: this list is a subset, not the forge's count. + let repos = self.inner.catalog()?.repos.into_iter().filter(|repo| repo.name == self.name).collect(); + Ok(huginn_daemon::sources::Catalog { repos, shortfall: None }) } fn clone_url(&self, name: &str) -> String { @@ -168,7 +170,7 @@ fn a_small_repo_is_refreshed_into_a_scratch_source_collection_and_found() { let mut embedder = OllamaEmbedder::new(&ollama, &model); let mut index = QdrantIndex::new(&qdrant).with_sources_collection(&scratch); let mut forge = OneRepo { inner: HttpForge::new(&forge), name: repo.clone() }; - assert_eq!(forge.catalog().unwrap().len(), 1, "{repo} is a public, live, non-empty repo on this forge"); + assert_eq!(forge.catalog().unwrap().repos.len(), 1, "{repo} is a public, live, non-empty repo on this forge"); let mut lines = Vec::new(); let refreshed = refresh(&mut embedder, &mut index, &mut forge, &root, &mut |report| lines.push(report.to_string())); let found = refreshed