use rayon::prelude::*;
use tracing::{debug, warn};
use crate::discover::FileInfo;
use crate::parse::error::ParseError;
use crate::parse::extractor::{extract_file, extract_from_source, ExtractResult};
#[derive(Debug)]
pub struct ParallelParseResult {
pub results: Vec<ExtractResult>,
pub errors: Vec<(String, String)>,
pub files_parsed: usize,
pub files_failed: usize,
}
#[must_use]
pub fn parallel_parse(files: &[FileInfo], project: &str) -> ParallelParseResult {
let results: Vec<Result<ExtractResult, (String, String)>> = files
.par_iter()
.map(|file| {
let lang = file
.language
.ok_or_else(|| (file.relative_path.clone(), "unknown language".to_string()))?;
let result = extract_file(&file.path, lang, project)
.map_err(|e| (file.relative_path.clone(), e.to_string()))?;
debug!(
event = "file_parsed",
path = %file.relative_path,
language = %lang,
nodes = result.nodes.len(),
"file parsed successfully"
);
Ok(result)
})
.collect();
let mut ok_results = Vec::with_capacity(files.len());
let mut errors = Vec::new();
for result in results {
match result {
Ok(r) => ok_results.push(r),
Err((path, msg)) => errors.push((path, msg)),
}
}
let files_parsed = ok_results.len();
let files_failed = errors.len();
ParallelParseResult {
results: ok_results,
errors,
files_parsed,
files_failed,
}
}
pub fn parse_single(file: &FileInfo, project: &str) -> Result<ExtractResult, ParseError> {
let lang = file
.language
.ok_or_else(|| ParseError::UnsupportedLanguage("unknown".to_string()))?;
extract_file(&file.path, lang, project)
}
pub type RamFirstSources = std::collections::HashMap<std::path::PathBuf, Vec<u8>>;
pub fn parallel_parse_ram_first(
files: &[FileInfo],
compressed: &RamFirstSources,
project: &str,
) -> ParallelParseResult {
let results: Vec<Result<ExtractResult, (String, String)>> = files
.par_iter()
.map(|file| {
let lang = file
.language
.ok_or_else(|| (file.relative_path.clone(), "unknown language".to_string()))?;
if let Some(comp_bytes) = compressed.get(&file.path) {
let raw = lz4_flex::decompress_size_prepended(comp_bytes).map_err(|e| {
(
file.relative_path.clone(),
format!("LZ4 decompress failed: {e}"),
)
})?;
let source = String::from_utf8(raw).map_err(|e| {
(
file.relative_path.clone(),
format!("UTF-8 decode failed: {e}"),
)
})?;
let result =
extract_from_source(&file.path.display().to_string(), &source, lang, project)
.map_err(|e| (file.relative_path.clone(), e.to_string()))?;
debug!(
event = "file_parsed_ram_first",
path = %file.relative_path,
language = %lang,
nodes = result.nodes.len(),
"file parsed from RAM-first buffer"
);
Ok(result)
} else {
warn!(
path = %file.relative_path,
"RAM-first: file not in compressed map, falling back to disk read"
);
let result = extract_file(&file.path, lang, project)
.map_err(|e| (file.relative_path.clone(), e.to_string()))?;
Ok(result)
}
})
.collect();
let mut ok_results = Vec::with_capacity(files.len());
let mut errors = Vec::new();
for result in results {
match result {
Ok(r) => ok_results.push(r),
Err((path, msg)) => errors.push((path, msg)),
}
}
let files_parsed = ok_results.len();
let files_failed = errors.len();
ParallelParseResult {
results: ok_results,
errors,
files_parsed,
files_failed,
}
}
#[cfg(all(
test,
feature = "lang-c",
feature = "lang-python",
feature = "lang-rust"
))]
mod tests {
use super::*;
use crate::model::Language;
use crate::test_log_capture::drain_to_string;
use crossbeam_channel::unbounded;
use inklog::domain::core::LoggerSubscriber;
use inklog::{LogRecord, Metrics};
use rayon::ThreadPoolBuilder;
use std::cell::RefCell;
use std::sync::Arc;
use tracing_subscriber::filter::LevelFilter;
use tracing_subscriber::prelude::*;
fn make_file(
dir: &std::path::Path,
name: &str,
content: &str,
language: Option<Language>,
) -> FileInfo {
let path = dir.join(name);
std::fs::write(&path, content).unwrap();
FileInfo {
path,
relative_path: name.to_string(),
language,
size: content.len() as u64,
}
}
#[test]
fn test_parallel_parse_multiple_rust_files() {
let dir = tempfile::tempdir().unwrap();
let files = vec![
make_file(dir.path(), "a.rs", "fn foo() {}", Some(Language::Rust)),
make_file(dir.path(), "b.rs", "fn bar() {}", Some(Language::Rust)),
];
let result = parallel_parse(&files, "testproject");
assert_eq!(result.files_parsed, 2);
assert_eq!(result.files_failed, 0);
assert_eq!(result.results.len(), 2);
assert!(result.errors.is_empty());
}
#[test]
fn test_parallel_parse_mixed_languages() {
let dir = tempfile::tempdir().unwrap();
let files = vec![
make_file(dir.path(), "a.rs", "fn foo() {}", Some(Language::Rust)),
make_file(
dir.path(),
"b.c",
"int foo(void) { return 0; }",
Some(Language::C),
),
make_file(
dir.path(),
"c.py",
"def foo(): pass",
Some(Language::Python),
),
];
let result = parallel_parse(&files, "proj");
assert_eq!(result.files_parsed, 3);
assert_eq!(result.files_failed, 0);
assert_eq!(result.results.len(), 3);
let langs: Vec<Language> = result.results.iter().map(|r| r.language).collect();
assert!(langs.contains(&Language::Rust));
assert!(langs.contains(&Language::C));
assert!(langs.contains(&Language::Python));
}
#[test]
fn test_parallel_parse_failure_handling() {
let dir = tempfile::tempdir().unwrap();
let good = make_file(dir.path(), "a.rs", "fn foo() {}", Some(Language::Rust));
let bad = FileInfo {
path: dir.path().join("does_not_exist.rs"),
relative_path: "does_not_exist.rs".to_string(),
language: Some(Language::Rust),
size: 0,
};
let files = vec![good, bad];
let result = parallel_parse(&files, "proj");
assert_eq!(result.files_parsed, 1, "the good file should parse");
assert_eq!(result.files_failed, 1, "the bad file should fail");
assert_eq!(result.results.len(), 1);
assert_eq!(result.errors.len(), 1);
assert_eq!(result.errors[0].0, "does_not_exist.rs");
assert!(
!result.errors[0].1.is_empty(),
"error message should be non-empty"
);
}
#[test]
fn test_parallel_parse_empty_list() {
let files: Vec<FileInfo> = vec![];
let result = parallel_parse(&files, "proj");
assert_eq!(result.files_parsed, 0);
assert_eq!(result.files_failed, 0);
assert!(result.results.is_empty());
assert!(result.errors.is_empty());
}
#[test]
fn test_parallel_parse_unknown_language_is_error() {
let dir = tempfile::tempdir().unwrap();
let files = vec![make_file(dir.path(), "a.rs", "fn foo() {}", None)];
let result = parallel_parse(&files, "proj");
assert_eq!(result.files_parsed, 0);
assert_eq!(result.files_failed, 1);
assert_eq!(result.errors.len(), 1);
assert_eq!(result.errors[0].0, "a.rs");
assert!(
result.errors[0].1.contains("unknown language"),
"error should mention unknown language: {}",
result.errors[0].1
);
}
#[test]
fn test_parallel_parse_correct_counts() {
let dir = tempfile::tempdir().unwrap();
let files = vec![
make_file(dir.path(), "a.rs", "fn a() {}", Some(Language::Rust)),
make_file(dir.path(), "b.rs", "fn b() {}", Some(Language::Rust)),
make_file(dir.path(), "c.rs", "fn c() {}", Some(Language::Rust)),
FileInfo {
path: dir.path().join("missing.rs"),
relative_path: "missing.rs".to_string(),
language: Some(Language::Rust),
size: 0,
},
FileInfo {
path: dir.path().join("unknown.rs"),
relative_path: "unknown.rs".to_string(),
language: None,
size: 0,
},
];
let result = parallel_parse(&files, "proj");
assert_eq!(result.files_parsed, 3);
assert_eq!(result.files_failed, 2);
assert_eq!(result.results.len(), 3);
assert_eq!(result.errors.len(), 2);
assert_eq!(result.files_parsed + result.files_failed, files.len());
}
#[test]
fn test_parse_single_succeeds() {
let dir = tempfile::tempdir().unwrap();
let file = make_file(dir.path(), "a.rs", "fn foo() {}", Some(Language::Rust));
let result = parse_single(&file, "proj");
assert!(
result.is_ok(),
"parse_single should succeed: {:?}",
result.err()
);
let result = result.unwrap();
assert_eq!(result.language, Language::Rust);
assert!(!result.nodes.is_empty(), "should extract nodes");
assert!(
result.nodes.iter().any(|n| n.name == "foo"),
"should extract the foo function: {:?}",
result.nodes.iter().map(|n| &n.name).collect::<Vec<_>>()
);
}
#[test]
fn test_parse_single_unknown_language_errors() {
let dir = tempfile::tempdir().unwrap();
let file = make_file(dir.path(), "a.rs", "fn foo() {}", None);
let result = parse_single(&file, "proj");
assert!(result.is_err());
match result.unwrap_err() {
ParseError::UnsupportedLanguage(_) => { }
other => panic!("expected UnsupportedLanguage, got {other:?}"),
}
}
#[test]
fn test_parallel_parse_thread_safety_many_files() {
let dir = tempfile::tempdir().unwrap();
let count = 15;
let files: Vec<FileInfo> = (0..count)
.map(|i| {
make_file(
dir.path(),
&format!("file_{i}.rs"),
&format!("fn func_{i}() {{}}"),
Some(Language::Rust),
)
})
.collect();
let result = parallel_parse(&files, "proj");
assert_eq!(result.files_parsed, count);
assert_eq!(result.files_failed, 0);
assert_eq!(result.results.len(), count);
assert!(result.errors.is_empty());
let mut all_names: Vec<String> = result
.results
.iter()
.flat_map(|r| r.nodes.iter().map(|n| n.name.clone()))
.collect();
all_names.sort();
for i in 0..count {
assert!(
all_names.contains(&format!("func_{i}")),
"missing func_{i} in extracted nodes"
);
}
}
#[test]
fn test_parallel_parse_results_contain_nodes_and_edges() {
let dir = tempfile::tempdir().unwrap();
let files = vec![make_file(
dir.path(),
"a.rs",
"fn foo() {}\nfn bar() {}\n",
Some(Language::Rust),
)];
let result = parallel_parse(&files, "proj");
assert_eq!(result.files_parsed, 1);
assert_eq!(result.results.len(), 1);
let extract = &result.results[0];
let names: Vec<&str> = extract.nodes.iter().map(|n| n.name.as_str()).collect();
assert!(names.contains(&"foo"), "should extract foo: {names:?}");
assert!(names.contains(&"bar"), "should extract bar: {names:?}");
assert!(
!extract.edges.is_empty(),
"should extract edges for definitions"
);
for edge in &extract.edges {
assert_eq!(edge.project, "proj", "edge project should match");
}
}
#[test]
fn test_parallel_parse_preserves_total_count() {
let dir = tempfile::tempdir().unwrap();
let files = vec![
make_file(dir.path(), "a.rs", "fn a() {}", Some(Language::Rust)),
make_file(dir.path(), "b.py", "def b(): pass", Some(Language::Python)),
make_file(
dir.path(),
"c.c",
"int c(void) { return 0; }",
Some(Language::C),
),
];
let result = parallel_parse(&files, "proj");
assert_eq!(result.results.len() + result.errors.len(), files.len());
assert_eq!(result.files_parsed + result.files_failed, files.len());
}
thread_local! {
static TRACING_GUARD: RefCell<Option<tracing::subscriber::DefaultGuard>> =
const { RefCell::new(None) };
}
fn capture_tracing_debug<R: Send>(f: impl FnOnce() -> R + Send) -> String {
let (console_tx, console_rx) = unbounded::<Arc<LogRecord>>();
let (async_tx, _async_rx) = unbounded::<Arc<LogRecord>>();
let metrics = Arc::new(Metrics::new());
let main_layer =
LoggerSubscriber::new(console_tx.clone(), async_tx.clone(), metrics.clone())
.with_filter(LevelFilter::DEBUG);
let main_registry = tracing_subscriber::registry().with(main_layer);
let console_tx_for_handler = console_tx.clone();
let async_tx_for_handler = async_tx.clone();
let metrics_for_handler = metrics.clone();
let pool = ThreadPoolBuilder::new()
.start_handler(move |_idx| {
let worker_layer = LoggerSubscriber::new(
console_tx_for_handler.clone(),
async_tx_for_handler.clone(),
metrics_for_handler.clone(),
)
.with_filter(LevelFilter::DEBUG);
let worker_registry = tracing_subscriber::registry().with(worker_layer);
let guard = tracing::subscriber::set_default(worker_registry);
TRACING_GUARD.with(|g| *g.borrow_mut() = Some(guard));
})
.build()
.expect("rayon thread pool");
tracing::subscriber::with_default(main_registry, || {
pool.install(f);
});
drain_to_string(&console_rx)
}
#[test]
fn log_002_file_parsed_event_emitted_on_success() {
let dir = tempfile::tempdir().unwrap();
let files = vec![
make_file(dir.path(), "a.rs", "fn foo() {}", Some(Language::Rust)),
make_file(dir.path(), "b.rs", "fn bar() {}", Some(Language::Rust)),
];
let captured = capture_tracing_debug(|| {
let _ = parallel_parse(&files, "proj");
});
assert!(
captured.contains("file_parsed"),
"LOG-002: file_parsed event should be emitted, got: {captured:?}"
);
let count = captured.matches("file_parsed").count();
assert_eq!(
count, 2,
"LOG-002: one file_parsed event per parsed file, got {count}"
);
assert!(
captured.contains("a.rs"),
"file_parsed event should carry the file path"
);
assert!(
captured.contains("nodes"),
"file_parsed event should carry the nodes field"
);
}
#[test]
fn log_002_file_parsed_not_emitted_on_failure() {
let dir = tempfile::tempdir().unwrap();
let files = vec![FileInfo {
path: dir.path().join("does_not_exist.rs"),
relative_path: "does_not_exist.rs".to_string(),
language: Some(Language::Rust),
size: 0,
}];
let captured = capture_tracing_debug(|| {
let _ = parallel_parse(&files, "proj");
});
assert!(
!captured.contains("file_parsed"),
"LOG-002: file_parsed should NOT be emitted on parse failure, got: {captured:?}"
);
}
fn compress_source(source: &str) -> Vec<u8> {
let compressed = lz4_flex::compress(source.as_bytes());
let mut result = (source.len() as u32).to_le_bytes().to_vec();
result.extend(compressed);
result
}
#[test]
fn ram_first_parses_from_compressed_buffer() {
let dir = tempfile::tempdir().unwrap();
let source = "fn foo() {}\nfn bar() {}\n";
let file = make_file(dir.path(), "a.rs", source, Some(Language::Rust));
let mut compressed = RamFirstSources::new();
compressed.insert(file.path.clone(), compress_source(source));
let result = parallel_parse_ram_first(&[file], &compressed, "proj");
assert_eq!(result.files_parsed, 1, "got errors: {:?}", result.errors);
assert_eq!(result.files_failed, 0);
assert_eq!(result.results.len(), 1);
let names: Vec<&str> = result.results[0]
.nodes
.iter()
.map(|n| n.name.as_str())
.collect();
assert!(names.contains(&"foo"), "should extract foo: {names:?}");
assert!(names.contains(&"bar"), "should extract bar: {names:?}");
}
#[test]
fn ram_first_falls_back_to_disk_when_not_in_map() {
let dir = tempfile::tempdir().unwrap();
let file = make_file(dir.path(), "a.rs", "fn foo() {}", Some(Language::Rust));
let compressed = RamFirstSources::new();
let result = parallel_parse_ram_first(&[file], &compressed, "proj");
assert_eq!(result.files_parsed, 1, "got errors: {:?}", result.errors);
assert_eq!(result.files_failed, 0);
assert_eq!(result.results.len(), 1);
}
#[test]
fn ram_first_lz4_decompress_failure_is_collected() {
let dir = tempfile::tempdir().unwrap();
let file = make_file(dir.path(), "a.rs", "fn foo() {}", Some(Language::Rust));
let mut compressed = RamFirstSources::new();
compressed.insert(file.path.clone(), vec![0xFF; 8]);
let result = parallel_parse_ram_first(&[file], &compressed, "proj");
assert_eq!(result.files_parsed, 0);
assert_eq!(result.files_failed, 1);
assert_eq!(result.errors.len(), 1);
assert_eq!(result.errors[0].0, "a.rs");
assert!(
result.errors[0].1.contains("LZ4 decompress failed"),
"got: {}",
result.errors[0].1
);
}
#[test]
fn ram_first_utf8_decode_failure_is_collected() {
let dir = tempfile::tempdir().unwrap();
let file = make_file(dir.path(), "a.rs", "fn foo() {}", Some(Language::Rust));
let invalid_utf8: Vec<u8> = vec![0xFF, 0xFE, 0xFD, 0x80];
let mut compressed = RamFirstSources::new();
let mut prepared = (invalid_utf8.len() as u32).to_le_bytes().to_vec();
prepared.extend(lz4_flex::compress(&invalid_utf8));
compressed.insert(file.path.clone(), prepared);
let result = parallel_parse_ram_first(&[file], &compressed, "proj");
assert_eq!(result.files_parsed, 0);
assert_eq!(result.files_failed, 1);
assert!(
result.errors[0].1.contains("UTF-8 decode failed"),
"got: {}",
result.errors[0].1
);
}
#[test]
fn ram_first_unknown_language_is_error() {
let dir = tempfile::tempdir().unwrap();
let file = make_file(dir.path(), "a.rs", "fn foo() {}", None);
let compressed = RamFirstSources::new();
let result = parallel_parse_ram_first(&[file], &compressed, "proj");
assert_eq!(result.files_parsed, 0);
assert_eq!(result.files_failed, 1);
assert!(result.errors[0].1.contains("unknown language"));
}
#[test]
fn ram_first_mixed_success_fallback_and_failure() {
let dir = tempfile::tempdir().unwrap();
let src_a = "fn a() {}";
let file_a = make_file(dir.path(), "a.rs", src_a, Some(Language::Rust));
let file_b = make_file(dir.path(), "b.rs", "fn b() {}", Some(Language::Rust));
let file_c = make_file(dir.path(), "c.rs", "fn c() {}", Some(Language::Rust));
let mut compressed = RamFirstSources::new();
compressed.insert(file_a.path.clone(), compress_source(src_a));
compressed.insert(file_c.path.clone(), vec![0xFF; 4]);
let files = vec![file_a, file_b, file_c];
let result = parallel_parse_ram_first(&files, &compressed, "proj");
assert_eq!(
result.files_parsed, 2,
"a + b should parse; got: {:?}",
result.errors
);
assert_eq!(result.files_failed, 1);
assert_eq!(result.results.len(), 2);
assert_eq!(result.errors.len(), 1);
assert_eq!(result.errors[0].0, "c.rs");
}
#[test]
fn ram_first_empty_file_list_returns_empty() {
let compressed = RamFirstSources::new();
let result = parallel_parse_ram_first(&[], &compressed, "proj");
assert_eq!(result.files_parsed, 0);
assert_eq!(result.files_failed, 0);
assert!(result.results.is_empty());
assert!(result.errors.is_empty());
}
#[test]
fn parse_single_io_error_for_missing_file() {
let file = FileInfo {
path: std::path::PathBuf::from("/nonexistent/path/missing.rs"),
relative_path: "missing.rs".to_string(),
language: Some(Language::Rust),
size: 0,
};
let result = parse_single(&file, "proj");
assert!(
result.is_err(),
"expected error for missing file, got: {result:?}"
);
match result.unwrap_err() {
ParseError::Io { file_path, .. } => {
assert!(
file_path.contains("missing.rs"),
"error should carry file path, got: {file_path}"
);
}
other => panic!("expected ParseError::Io, got: {other:?}"),
}
}
#[test]
fn log_002_ram_first_emits_file_parsed_ram_first_event() {
let dir = tempfile::tempdir().unwrap();
let source = "fn foo() {}\nfn bar() {}\n";
let file = make_file(dir.path(), "a.rs", source, Some(Language::Rust));
let mut compressed = RamFirstSources::new();
compressed.insert(file.path.clone(), compress_source(source));
let captured = capture_tracing_debug(|| {
let _ = parallel_parse_ram_first(&[file], &compressed, "proj");
});
assert!(
captured.contains("file_parsed_ram_first"),
"LOG-002: file_parsed_ram_first event should be emitted, got: {captured:?}"
);
assert!(
captured.contains("a.rs"),
"event should carry the file path, got: {captured:?}"
);
assert!(
captured.contains("nodes"),
"event should carry the nodes field, got: {captured:?}"
);
}
#[test]
fn log_002_ram_first_warns_on_disk_fallback() {
let dir = tempfile::tempdir().unwrap();
let file = make_file(dir.path(), "a.rs", "fn foo() {}", Some(Language::Rust));
let compressed = RamFirstSources::new();
let captured = capture_tracing_debug(|| {
let _ = parallel_parse_ram_first(&[file], &compressed, "proj");
});
assert!(
captured.contains("RAM-first"),
"warn event should mention RAM-first, got: {captured:?}"
);
assert!(
captured.contains("falling back to disk read"),
"warn event should mention disk fallback, got: {captured:?}"
);
assert!(
captured.contains("a.rs"),
"warn event should carry the file path, got: {captured:?}"
);
}
#[test]
fn log_002_ram_first_no_event_on_decompress_failure() {
let dir = tempfile::tempdir().unwrap();
let file = make_file(dir.path(), "a.rs", "fn foo() {}", Some(Language::Rust));
let mut compressed = RamFirstSources::new();
compressed.insert(file.path.clone(), vec![0xFF; 8]);
let captured = capture_tracing_debug(|| {
let _ = parallel_parse_ram_first(&[file], &compressed, "proj");
});
assert!(
!captured.contains("file_parsed_ram_first"),
"file_parsed_ram_first should NOT be emitted on decompress failure, got: {captured:?}"
);
}
#[test]
fn ram_first_multiple_compressed_files_all_parsed() {
let dir = tempfile::tempdir().unwrap();
let src_a = "fn a() {}";
let src_b = "fn b() {}\nfn c() {}\n";
let file_a = make_file(dir.path(), "a.rs", src_a, Some(Language::Rust));
let file_b = make_file(dir.path(), "b.rs", src_b, Some(Language::Rust));
let mut compressed = RamFirstSources::new();
compressed.insert(file_a.path.clone(), compress_source(src_a));
compressed.insert(file_b.path.clone(), compress_source(src_b));
let result = parallel_parse_ram_first(&[file_a, file_b], &compressed, "proj");
assert_eq!(result.files_parsed, 2, "got errors: {:?}", result.errors);
assert_eq!(result.files_failed, 0);
assert_eq!(result.results.len(), 2);
let all_names: Vec<&str> = result
.results
.iter()
.flat_map(|r| r.nodes.iter().map(|n| n.name.as_str()))
.collect();
assert!(all_names.contains(&"a"), "should extract a: {all_names:?}");
assert!(all_names.contains(&"b"), "should extract b: {all_names:?}");
assert!(all_names.contains(&"c"), "should extract c: {all_names:?}");
}
#[test]
fn ram_first_all_files_failed_returns_empty_results() {
let dir = tempfile::tempdir().unwrap();
let file_a = make_file(dir.path(), "a.rs", "fn foo() {}", Some(Language::Rust));
let file_b = make_file(dir.path(), "b.rs", "fn bar() {}", Some(Language::Rust));
let mut compressed = RamFirstSources::new();
compressed.insert(file_a.path.clone(), vec![0xFF; 4]);
compressed.insert(file_b.path.clone(), vec![0xFF; 4]);
let result = parallel_parse_ram_first(&[file_a, file_b], &compressed, "proj");
assert_eq!(result.files_parsed, 0);
assert_eq!(result.files_failed, 2);
assert!(result.results.is_empty());
assert_eq!(result.errors.len(), 2);
for (_, msg) in &result.errors {
assert!(
msg.contains("LZ4 decompress failed"),
"expected LZ4 failure, got: {msg}"
);
}
}
}