diff --git a/crates/huginn-daemon/src/daemon.rs b/crates/huginn-daemon/src/daemon.rs index f04f45d..4507712 100644 --- a/crates/huginn-daemon/src/daemon.rs +++ b/crates/huginn-daemon/src/daemon.rs @@ -551,6 +551,83 @@ pub(crate) mod tests { fn abandon(&mut self, _ticket: SearchTicket) {} } + /// An index that answers source searches from a script and counts how often + /// it was asked. + struct SourcesIndex { + asked: Arc>, + hits: SourceHits, + } + + impl IndexSink for SourcesIndex { + fn committed(&mut self, _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 cannot search the mind".into()) + } + + fn search_sources(&mut self, _query: &SourceQuery) -> Result { + *self.asked.lock().unwrap() += 1; + Ok(SearchTicket(7)) + } + + fn sources_searched(&mut self) -> Vec<(SearchTicket, Result)> { + vec![(SearchTicket(7), Ok(self.hits.clone()))] + } + + fn search_deadline(&self) -> std::time::Duration { + std::time::Duration::ZERO + } + + fn searched(&mut self) -> Vec<(SearchTicket, Result)> { + Vec::new() + } + + fn abandon(&mut self, _ticket: SearchTicket) {} + } + + /// A source search is checked at the door before the index is asked, is + /// answered with the hits the index found and nothing joined from the mind, + /// turns a failure into `Unavailable`, and is refused for another mind. + #[test] + fn a_source_search_reaches_the_index_only_when_well_formed_and_answers_its_hits() { + use huginn_mind::{SourceHit, SourceModel}; + let hits = SourceHits { + indexed_under: SourceModel { name: "m".into(), digest: "d".into() }, + hits: vec![SourceHit { repo: "R".into(), path: "p.rs".into(), commit: "c".into(), line_start: 1, line_end: 2, score: 0.5, text: "t".into() }], + }; + let asked = Arc::new(Mutex::new(0)); + let root = tempfile::tempdir().unwrap(); + let mind = Mind::open(root.path(), &slug(INSTANCE)).unwrap(); + let mut daemon = Daemon::new(mind, SourcesIndex { asked: Arc::clone(&asked), hits: hits.clone() }); + let request = |instance: &str, top_k: u32| HuginnMindRequest::SearchSources { + instance: slug(instance), + query: SourceQuery { text: huginn_mind::eureka_pipeline::Line("find".into()), top_k, repo: None }, + }; + + let Handled::Searching(search) = daemon.handle(request(INSTANCE, 5), now()) else { panic!("a well-formed source search is a search") }; + assert_eq!((search.ticket, *asked.lock().unwrap()), (SearchTicket(7), 1)); + let mut finished = daemon.searched(); + let (ticket, found) = finished.remove(0); + assert_eq!(ticket, SearchTicket(7)); + assert_eq!(daemon.finish(search, found), HuginnMindResponse::Sources(hits)); + + let Handled::Searching(search) = daemon.handle(request(INSTANCE, 5), now()) else { panic!() }; + let failed = daemon.finish(search, Found::Failed("the index fell over".into())); + assert!(matches!(failed, HuginnMindResponse::Refused(MindRefusal::Unavailable { ref detail }) if detail.contains("fell over")), "{failed:?}"); + + for refused in [request(INSTANCE, 0), request(INSTANCE, 51), request(OTHER, 5)] { + let answer = daemon.handle(refused, now()).answered(); + assert!(matches!(answer, HuginnMindResponse::Refused(_)), "{answer:?}"); + } + assert_eq!(*asked.lock().unwrap(), 2, "a refused query never reached the index"); + } + /// A daemon over the instance's mind under `state_root`, with no index. pub(crate) fn open_unindexed( state_root: &Path, diff --git a/crates/huginn-daemon/src/index/tests.rs b/crates/huginn-daemon/src/index/tests.rs index c4fb3f6..2382a98 100644 --- a/crates/huginn-daemon/src/index/tests.rs +++ b/crates/huginn-daemon/src/index/tests.rs @@ -2057,3 +2057,30 @@ fn a_worker_stuck_on_the_minds_collection_still_answers_a_source_search() { assert_eq!((answered.hits[0].repo.as_str(), answered.hits[0].path.as_str()), ("Huginn", "a.rs")); assert_eq!(answered.indexed_under.digest, "d1"); } + +/// Source searches take their own tickets, share the queue's bound with the +/// mind's searches, and an abandon removes the one ticket named from either. +#[test] +fn source_searches_have_distinct_tickets_share_the_queue_bound_and_abandon_by_ticket() { + let shared = shared_with(Vec::new(), None, Vec::new(), false); + let mut sink = WorkerSink { shared, identified: true, deadline: SEARCH_DEADLINE }; + let first = ask_sources(&mut sink, "a").unwrap(); + let second = ask_sources(&mut sink, "b").unwrap(); + let mind = ask_sink(&mut sink, "m", 3).unwrap(); + assert!(first != second && second != mind && first != mind, "{first:?} {second:?} {mind:?}"); + + abandon(&mut sink, first); + { + let guard = lock(&sink.shared); + let left: Vec = guard.source_searches.iter().map(|queued| queued.ticket).collect(); + assert_eq!((left, guard.searches.len()), (vec![second], 1), "only the named ticket went"); + } + + for _ in 0..SEARCHES_QUEUED_MAX - 2 { + ask_sources(&mut sink, "fill").unwrap(); + } + assert!(ask_sources(&mut sink, "over").is_err(), "the bound counts both kinds"); + assert!(ask_sink(&mut sink, "over", 3).is_err(), "and refuses both kinds"); + abandon(&mut sink, second); + assert!(ask_sink(&mut sink, "now", 3).is_ok(), "a freed slot is a slot"); +} diff --git a/crates/huginn-daemon/src/sources/chunk.rs b/crates/huginn-daemon/src/sources/chunk.rs index 7a65ace..394d0ad 100644 --- a/crates/huginn-daemon/src/sources/chunk.rs +++ b/crates/huginn-daemon/src/sources/chunk.rs @@ -131,4 +131,23 @@ mod tests { let held = Chunk { line_start: 3, line_end: 4, text: "fn a() {}".into() }; assert_eq!(embedded_text("Huginn", "src/a.rs", &held), "Huginn/src/a.rs\nfn a() {}"); } + + /// The character bound is inclusive: a line of exactly the bound is kept + /// whole, one more character is cut; and the running count of a window + /// includes each joining newline. + #[test] + fn the_character_bound_is_inclusive_and_counts_the_joining_newlines() { + let exact = "z".repeat(CHUNK_CHARS_MAX); + let kept = chunk(&format!("{exact}\n")); + assert_eq!((kept.len(), kept[0].text.chars().count()), (1, CHUNK_CHARS_MAX)); + let over = chunk(&format!("{exact}z\nnext\n")); + let shown: Vec<(u32, u32, usize)> = over.iter().map(|held| (held.line_start, held.line_end, held.text.chars().count())).collect(); + assert_eq!(shown, [(1, 1, CHUNK_CHARS_MAX), (2, 2, 4)]); + + // 13 lines of 100 characters are 1,312 with their newlines; the 14th would make 1,413. + let hundred = "h".repeat(100); + let text: String = (0..20).map(|_| format!("{hundred}\n")).collect(); + let ranges: Vec<(u32, u32)> = chunk(&text).iter().map(|held| (held.line_start, held.line_end)).collect(); + assert_eq!(ranges, [(1, 13), (14, 20)]); + } } diff --git a/crates/huginn-daemon/src/sources/forge.rs b/crates/huginn-daemon/src/sources/forge.rs index 989abe5..60cc24f 100644 --- a/crates/huginn-daemon/src/sources/forge.rs +++ b/crates/huginn-daemon/src/sources/forge.rs @@ -90,4 +90,38 @@ mod tests { let names: Vec = indexable(&page).unwrap().into_iter().map(|repo| repo.name).collect(); assert_eq!(names, ["Open"]); } + + /// Serves each page in `pages` once, to the request that names it. + fn serve(pages: Vec) -> String { + use std::io::{Read, Write}; + let listener = std::net::TcpListener::bind("127.0.0.1:0").unwrap(); + let address = listener.local_addr().unwrap(); + std::thread::spawn(move || { + for _ in 0..pages.len() { + let (mut stream, _) = listener.accept().unwrap(); + let mut buffer = [0_u8; 2048]; + 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(); + } + }); + format!("http://{address}") + } + + /// 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 mut forge = HttpForge::new(&format!("{base}/")); + let names: Vec = forge.catalog().unwrap().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!(forge.clone_url("Demo"), format!("{base}/GameCult/Demo.git")); + } } diff --git a/crates/huginn-daemon/src/sources/tests.rs b/crates/huginn-daemon/src/sources/tests.rs index 2e94287..16ba5d8 100644 --- a/crates/huginn-daemon/src/sources/tests.rs +++ b/crates/huginn-daemon/src/sources/tests.rs @@ -406,4 +406,18 @@ fn a_chunk_id_is_its_content_address() { assert_ne!(base, other); } assert_ne!(chunk_id("R", "pb", "lob", 0), chunk_id("R", "p", "blob", 0), "fields are separated, not concatenated"); + +/// Chunks are embedded in batches of at most `BATCH`, however many a repo has. +#[test] +fn chunks_are_embedded_in_bounded_batches() { + let remote = Remote::new(); + let many: Vec<(String, Vec)> = (0..70).map(|n| (format!("f{n}.txt"), format!("file {n}\n").into_bytes())).collect(); + let files: Vec<(&str, &[u8])> = many.iter().map(|(path, bytes)| (path.as_str(), bytes.as_slice())).collect(); + remote.commit(&files, &[]); + let mut rig = Rig::new(); + rig.converged(&mut FakeForge::of(&[("Demo", &remote)])); + let sizes: Vec = rig.embedder.state.lock().unwrap().embedded.iter().map(Vec::len).collect(); + assert_eq!(sizes, [BATCH, BATCH, 6]); + assert_eq!(rig.index.held().len(), 70); +} }