Merge eureka-body/source-search-index-safety (cursor, catalog, retire, CLI e2e; report h3)
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
This commit is contained in:
Meta Vens
commit
44b3bf5e91
8 changed files with 444 additions and 40 deletions
|
|
@ -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<huginn_daemon::sources::SourcePayload>,
|
||||
tickets: u64,
|
||||
done: Vec<(SearchTicket, Result<huginn_mind::SourceHits, String>)>,
|
||||
}
|
||||
|
||||
/// 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<huginn_daemon::index::ModelIdentity> {
|
||||
Ok(huginn_daemon::index::ModelIdentity { name: "flat".into(), digest: self.0.into(), dimensions: 2 })
|
||||
}
|
||||
|
||||
fn embed(&mut self, texts: &[String], _within: Duration) -> anyhow::Result<Vec<Vec<f32>>> {
|
||||
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<huginn_daemon::sources::DescribedSources> {
|
||||
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<String, String>) -> 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<Vec<huginn_daemon::sources::SourceRow>> {
|
||||
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<S: MindStore> IndexSink<S> for Desk {
|
||||
fn committed(&mut self, _mind: &huginn_mind::Mind<S>, _writes: &[PipelineRef]) -> anyhow::Result<()> {
|
||||
Ok(())
|
||||
}
|
||||
fn status(&self) -> IndexStatus {
|
||||
IndexStatus::Current
|
||||
}
|
||||
fn search(&mut self, _text: &str, _top_k: u32, _among: Vec<String>) -> Result<SearchTicket, String> {
|
||||
Err("this index holds sources only".into())
|
||||
}
|
||||
fn search_sources(&mut self, query: &huginn_mind::SourceQuery) -> Result<SearchTicket, String> {
|
||||
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<huginn_mind::SourceHits, String>)> {
|
||||
std::mem::take(&mut self.done)
|
||||
}
|
||||
fn search_deadline(&self) -> Duration {
|
||||
Duration::from_secs(5)
|
||||
}
|
||||
fn searched(&mut self) -> Vec<(SearchTicket, Result<Hits, String>)> {
|
||||
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);
|
||||
}
|
||||
|
|
|
|||
|
|
@ -148,7 +148,10 @@ impl Drop for Server {
|
|||
}
|
||||
}
|
||||
|
||||
pub fn serve(root: TempDir, mut daemon: TestDaemon) -> Server {
|
||||
pub fn serve<I: IndexSink<OwnedRedbMessagePackBackingStore> + Send + 'static>(
|
||||
root: TempDir,
|
||||
mut daemon: Daemon<OwnedRedbMessagePackBackingStore, I>,
|
||||
) -> 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));
|
||||
|
|
|
|||
|
|
@ -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) => {
|
||||
|
|
|
|||
|
|
@ -258,6 +258,8 @@ pub(crate) struct SourcesState {
|
|||
pub(crate) fail_upserts_after: Option<u32>,
|
||||
pub(crate) cursor_writes: Vec<std::collections::BTreeMap<String, String>>,
|
||||
pub(crate) deletes: Vec<(String, Option<Vec<String>>)>,
|
||||
/// A delete naming this repo refuses.
|
||||
pub(crate) fail_deletes_of: Option<String>,
|
||||
/// The repo filter and limit of every search, in order.
|
||||
pub(crate) searches: Vec<(Option<String>, 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)))
|
||||
|
|
|
|||
|
|
@ -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<usize>, 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<usize>) -> Option<String> {
|
||||
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<Vec<SourceRepo>> {
|
|||
}
|
||||
|
||||
impl Forge for HttpForge {
|
||||
fn catalog(&mut self) -> Result<Vec<SourceRepo>> {
|
||||
fn catalog(&mut self) -> Result<Catalog> {
|
||||
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>) -> 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<String>, total: Option<usize>) -> 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<usize>) -> String {
|
||||
let repos: Vec<Value> = 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<usize>| {
|
||||
let repos: Vec<Value> = 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<String> = forge.catalog().unwrap().into_iter().map(|repo| repo.name).collect();
|
||||
let catalog = forge.catalog().unwrap();
|
||||
let names: Vec<String> = 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"));
|
||||
}
|
||||
}
|
||||
|
|
|
|||
|
|
@ -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<SourceRepo>,
|
||||
/// Why the list may be missing repos; `None` when it is the forge's count.
|
||||
pub shortfall: Option<String>,
|
||||
}
|
||||
|
||||
/// The forge: which repos exist, and where to fetch them from.
|
||||
pub trait Forge {
|
||||
fn catalog(&mut self) -> Result<Vec<SourceRepo>>;
|
||||
fn catalog(&mut self) -> Result<Catalog>;
|
||||
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<String>,
|
||||
/// Set when retirement was skipped because the catalog was empty or short.
|
||||
pub retirement_withheld: Option<String>,
|
||||
}
|
||||
|
||||
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<E: Embedder, S: SourceIndex, F: Forge>(
|
|||
|
||||
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<E: Embedder, S: SourceIndex, F: Forge>(
|
|||
|
||||
/// 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<S: SourceIndex>(
|
||||
index: &mut S,
|
||||
catalog: &[SourceRepo],
|
||||
mirrors: &Path,
|
||||
cursors: &mut BTreeMap<String, String>,
|
||||
) -> Result<()> {
|
||||
let listed = |name: &str| catalog.iter().any(|repo| repo.name.eq_ignore_ascii_case(name));
|
||||
failed: &mut Vec<String>,
|
||||
) {
|
||||
let listed = |name: &str| catalog.iter().any(|repo| repo.name == name);
|
||||
let mut gone: BTreeSet<String> = 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<S: SourceIndex>(
|
|||
}
|
||||
}
|
||||
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<S: SourceIndex>(index: &mut S, name: &str, mirrors: &Path, cursors: &mut BTreeMap<String, String>) -> 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<E: Embedder, S: SourceIndex, F: Forge>(
|
|||
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<String> = paths.iter().cloned().collect();
|
||||
|
|
|
|||
|
|
@ -58,6 +58,7 @@ impl Remote {
|
|||
|
||||
struct FakeForge {
|
||||
repos: Vec<(SourceRepo, PathBuf)>,
|
||||
shortfall: Option<String>,
|
||||
}
|
||||
|
||||
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<Vec<SourceRepo>> {
|
||||
Ok(self.repos.iter().map(|(repo, _)| repo.clone()).collect())
|
||||
fn catalog(&mut self) -> Result<Catalog> {
|
||||
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<u8>)> = (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<u8>)> = (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::<Vec<_>>(), ["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::<Vec<_>>(), ["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<String> = 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::<Vec<_>>(), [("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]
|
||||
|
|
|
|||
|
|
@ -141,8 +141,10 @@ struct OneRepo {
|
|||
}
|
||||
|
||||
impl huginn_daemon::sources::Forge for OneRepo {
|
||||
fn catalog(&mut self) -> anyhow::Result<Vec<huginn_daemon::sources::SourceRepo>> {
|
||||
Ok(self.inner.catalog()?.into_iter().filter(|repo| repo.name == self.name).collect())
|
||||
fn catalog(&mut self) -> anyhow::Result<huginn_daemon::sources::Catalog> {
|
||||
// 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
|
||||
|
|
|
|||
Loading…
Reference in a new issue