From 293e30ae7ff64ff1deef3cfe9b9199b4198f807f Mon Sep 17 00:00:00 2001 From: Konstantin Mikhailov Date: Sun, 27 Sep 2026 15:53:46 +0300 Subject: [PATCH 1/2] Fix staged worker cleanup and concurrent catalog publication --- docs/architecture.md | 6 + docs/synthetic-pcap-generator.md | 6 + src/dns_processor/pipeline.rs | 358 +++++++++++++++++++++----- tools/dns-catalog-builder/src/main.rs | 312 ++++++++++++++++++++-- 4 files changed, 595 insertions(+), 87 deletions(-) diff --git a/docs/architecture.md b/docs/architecture.md index 04b4149..d6a4b73 100644 --- a/docs/architecture.md +++ b/docs/architecture.md @@ -83,6 +83,12 @@ and uses the remaining CPU budget for shard workers. Routed DNS packets also car metadata into shard workers so the worker path can reuse the first L3/L4 parse instead of repeating it for full DNS question decoding. +The staged pipeline waits for every successfully started worker before returning a startup +error. It releases the input and result channels before joining those workers, so cleanup can +finish without an aggregator. During normal teardown it joins every worker +even if an earlier worker returns an error or panics, preserving the first worker error after +any parser-stage error. + Ethernet decoding also produces one canonical VLAN context: the ordered stack of tag TPIDs and 12-bit VLAN IDs. PCP and DEI are QoS metadata and do not distinguish transactions. This context is shared by fragment keys and matcher identities, keeping overlapping IP endpoints in different diff --git a/docs/synthetic-pcap-generator.md b/docs/synthetic-pcap-generator.md index d84bbff..0763805 100644 --- a/docs/synthetic-pcap-generator.md +++ b/docs/synthetic-pcap-generator.md @@ -73,6 +73,12 @@ cargo run -p dns-catalog-builder --release -- \ --top 10000 ``` +The catalog builder writes and syncs a separate temporary file beside the destination before +replacing the output. Concurrent builds targeting the same path cannot truncate each other's +temporary files; the last successful replacement supplies the complete catalog. Ordinary I/O +errors preserve the previous output and trigger best-effort removal of that build's temporary +file. A process crash or a filesystem cleanup error can leave a temporary file behind. + The generator loads `fitted-generator.toml` from `--profile-dir`, verifies the referenced `catalog_data.tsv` digest, and then uses the artifact directory as the runtime source of truth. The checked-in `server1-jul-2024` profile points its `catalog_path` at the workspace-level diff --git a/src/dns_processor/pipeline.rs b/src/dns_processor/pipeline.rs index 2883b56..107d421 100644 --- a/src/dns_processor/pipeline.rs +++ b/src/dns_processor/pipeline.rs @@ -351,6 +351,44 @@ fn join_thread(handle: thread::JoinHandle>, label: &str) -> .map_err(|err| io::Error::other(format!("{label} panicked: {:?}", err)))? } +fn join_matcher_workers( + handles: Vec>>, +) -> anyhow::Result<()> { + let mut result = Ok(()); + for (worker_idx, handle) in handles.into_iter().enumerate() { + let worker_result = join_thread(handle, &format!("Matcher worker {worker_idx}")); + if result.is_ok() { + result = worker_result; + } + } + result +} + +trait StagedThreadSpawner { + fn spawn( + &mut self, + builder: thread::Builder, + task: F, + ) -> io::Result>> + where + F: FnOnce() -> anyhow::Result<()> + Send + 'static; +} + +struct SystemThreadSpawner; + +impl StagedThreadSpawner for SystemThreadSpawner { + fn spawn( + &mut self, + builder: thread::Builder, + task: F, + ) -> io::Result>> + where + F: FnOnce() -> anyhow::Result<()> + Send + 'static, + { + builder.spawn(task) + } +} + fn logical_shard_count(num_threads: usize, shard_parallelism_enabled: bool) -> usize { if !shard_parallelism_enabled { return 1; @@ -1037,13 +1075,33 @@ fn run_staged_processing_pipeline( worker_count: usize, affinity_plan: AffinityPlan, signals: WorkerShutdownSignals, +) -> anyhow::Result { + run_staged_processing_pipeline_with_spawner( + dns_processor, + batch_rx, + tx, + ShardRoutingPlan::new(shard_count, worker_count), + affinity_plan, + signals, + &mut SystemThreadSpawner, + ) +} + +fn run_staged_processing_pipeline_with_spawner( + dns_processor: Arc, + batch_rx: Receiver, + tx: Sender, + routing_plan: ShardRoutingPlan, + affinity_plan: AffinityPlan, + signals: WorkerShutdownSignals, + spawner: &mut impl StagedThreadSpawner, ) -> anyhow::Result { let WorkerShutdownSignals { shutdown_requested, intake_failed, output_closed, } = signals; - let routing_plan = ShardRoutingPlan::new(shard_count, worker_count); + let worker_count = routing_plan.worker_ranges.len(); let (result_tx, result_rx) = crossbeam::channel::bounded(MATCHER_WORKER_QUEUE_DEPTH * worker_count.max(1)); @@ -1054,67 +1112,85 @@ fn run_staged_processing_pipeline( let (worker_tx, worker_rx) = crossbeam::channel::bounded(MATCHER_WORKER_QUEUE_DEPTH); worker_txs.push(worker_tx); - worker_handles.push( - thread::Builder::new() - .name(format!("DPP_Matcher_{}", worker_idx)) - .spawn({ - let dns_processor = Arc::clone(&dns_processor); - let result_tx = result_tx.clone(); - let shutdown_requested = Arc::clone(&shutdown_requested); - let intake_failed = Arc::clone(&intake_failed); - let output_closed = Arc::clone(&output_closed); - let affinity_plan = affinity_plan.clone(); - move || { - affinity_plan.apply_to_current_thread( - staged_matcher_affinity_slot(worker_idx), - "staged matcher worker", - ); - run_matcher_worker( - dns_processor, - worker_idx, - shard_range, - worker_rx, - result_tx, - WorkerShutdownSignals { - shutdown_requested, - intake_failed, - output_closed, - }, - ) - } - })?, + let worker = spawner.spawn( + thread::Builder::new().name(format!("DPP_Matcher_{}", worker_idx)), + { + let dns_processor = Arc::clone(&dns_processor); + let result_tx = result_tx.clone(); + let shutdown_requested = Arc::clone(&shutdown_requested); + let intake_failed = Arc::clone(&intake_failed); + let output_closed = Arc::clone(&output_closed); + let affinity_plan = affinity_plan.clone(); + move || { + affinity_plan.apply_to_current_thread( + staged_matcher_affinity_slot(worker_idx), + "staged matcher worker", + ); + run_matcher_worker( + dns_processor, + worker_idx, + shard_range, + worker_rx, + result_tx, + WorkerShutdownSignals { + shutdown_requested, + intake_failed, + output_closed, + }, + ) + } + }, ); + match worker { + Ok(handle) => worker_handles.push(handle), + Err(error) => { + drop(worker_txs); + drop(batch_rx); + drop(result_tx); + // No aggregator will receive the workers' finalization messages. + drop(result_rx); + let _ = join_matcher_workers(worker_handles); + return Err(error.into()); + } + } } drop(result_tx); - let parser_handle = thread::Builder::new() - .name("DPP_Parser".to_string()) - .spawn({ - let output_closed = Arc::clone(&output_closed); - let allow_fragments = dns_processor.allow_fragments; - let full_fragments = dns_processor.full_fragments; - let match_timeout_micros = dns_processor.match_timeout_micros; - let monotonic_capture = dns_processor.monotonic_capture; - let affinity_plan = affinity_plan.clone(); - move || { - affinity_plan.apply_to_current_thread( - staged_parser_affinity_slot(worker_count), - "staged parser", - ); - run_parser_stage( - batch_rx, - worker_txs, - routing_plan, - output_closed, - FragmentProcessingConfig { - allow_fragments, - full_fragments, - match_timeout_micros, - monotonic_capture, - }, - ) - } - })?; + let parser = spawner.spawn(thread::Builder::new().name("DPP_Parser".to_string()), { + let output_closed = Arc::clone(&output_closed); + let allow_fragments = dns_processor.allow_fragments; + let full_fragments = dns_processor.full_fragments; + let match_timeout_micros = dns_processor.match_timeout_micros; + let monotonic_capture = dns_processor.monotonic_capture; + let affinity_plan = affinity_plan.clone(); + move || { + affinity_plan.apply_to_current_thread( + staged_parser_affinity_slot(worker_count), + "staged parser", + ); + run_parser_stage( + batch_rx, + worker_txs, + routing_plan, + output_closed, + FragmentProcessingConfig { + allow_fragments, + full_fragments, + match_timeout_micros, + monotonic_capture, + }, + ) + } + }); + let parser_handle = match parser { + Ok(handle) => handle, + Err(error) => { + // The rejected task drops the intake receiver and worker senders. + drop(result_rx); + let _ = join_matcher_workers(worker_handles); + return Err(error.into()); + } + }; affinity_plan.apply_to_current_thread( staged_aggregator_affinity_slot(worker_count), @@ -1123,13 +1199,7 @@ fn run_staged_processing_pipeline( let aggregator_result = run_aggregator(result_rx, tx, worker_count, output_closed); let parser_result = join_thread(parser_handle, "Parser stage"); - let worker_result = - worker_handles - .into_iter() - .enumerate() - .try_for_each(|(worker_idx, worker_handle)| { - join_thread(worker_handle, &format!("Matcher worker {}", worker_idx)).map(|_| ()) - }); + let worker_result = join_matcher_workers(worker_handles); parser_result?; worker_result?; @@ -1304,6 +1374,7 @@ mod tests { use std::fs; use std::sync::Mutex; use std::sync::atomic::{AtomicBool, AtomicUsize}; + use std::time::{Duration, Instant}; fn test_worker_signals(shutdown_requested: bool) -> WorkerShutdownSignals { WorkerShutdownSignals { @@ -1313,6 +1384,163 @@ mod tests { } } + struct FailingStagedSpawner { + call: usize, + fail_at: usize, + finished_work: Sender<()>, + release: Receiver<()>, + exits: Arc, + } + + impl StagedThreadSpawner for FailingStagedSpawner { + fn spawn( + &mut self, + builder: thread::Builder, + task: F, + ) -> io::Result>> + where + F: FnOnce() -> anyhow::Result<()> + Send + 'static, + { + let call = self.call; + self.call += 1; + if call == self.fail_at { + return Err(io::Error::other(format!( + "injected spawn failure at {call}" + ))); + } + + let finished_work = self.finished_work.clone(); + let release = self.release.clone(); + let exits = Arc::clone(&self.exits); + builder.spawn(move || { + let _ = task(); + finished_work + .send(()) + .expect("test observes worker cleanup"); + release.recv().expect("test releases worker exit"); + exits.fetch_add(1, AtomicOrdering::SeqCst); + Err(anyhow::anyhow!("secondary worker failure")) + }) + } + } + + fn finish_after_worker_release( + finished_work: Receiver<()>, + release: Sender<()>, + done: Receiver>, + coordinator: thread::JoinHandle<()>, + held_workers: usize, + ) -> anyhow::Result { + for _ in 0..held_workers { + finished_work + .recv_timeout(Duration::from_secs(5)) + .expect("worker reaches its exit gate"); + } + let early_result = (held_workers > 0).then(|| done.recv_timeout(Duration::from_millis(50))); + // Release even if the coordinator returned early, so a failing test leaves no held worker. + for _ in 0..held_workers { + release.send(()).expect("worker exit released"); + } + let returned_early = matches!(early_result, Some(Ok(_))); + let result = match early_result { + Some(Ok(result)) => result, + Some(Err(crossbeam::channel::RecvTimeoutError::Disconnected)) => { + panic!("coordinator disconnected without returning its result") + } + _ => done + .recv_timeout(Duration::from_secs(5)) + .expect("coordinator finishes after workers exit"), + }; + coordinator.join().expect("test coordinator joins"); + assert!( + !returned_early, + "returned while a worker was still held alive" + ); + result + } + + #[test] + fn staged_spawn_failures_join_started_workers_and_preserve_startup_error() { + // Calls 0/1 start matcher workers; call 2 starts the parser. + for fail_at in 0..=2 { + let (batch_tx, batch_rx) = crossbeam::channel::bounded(1); + let (output_tx, _output_rx) = crossbeam::channel::unbounded(); + let (finished_tx, finished_rx) = crossbeam::channel::unbounded(); + let (release_tx, release_rx) = crossbeam::channel::unbounded(); + let (done_tx, done_rx) = crossbeam::channel::bounded(1); + let exits = Arc::new(AtomicUsize::new(0)); + let mut spawner = FailingStagedSpawner { + call: 0, + fail_at, + finished_work: finished_tx, + release: release_rx, + exits: Arc::clone(&exits), + }; + let processor = Arc::new(DnsProcessor::new(None).expect("processor initializes")); + let coordinator = thread::spawn(move || { + let result = run_staged_processing_pipeline_with_spawner( + processor, + batch_rx, + output_tx, + ShardRoutingPlan::new(4, 2), + AffinityPlan::disabled(), + test_worker_signals(false), + &mut spawner, + ); + done_tx.send(result).expect("test observes startup result"); + }); + + let error = + finish_after_worker_release(finished_rx, release_tx, done_rx, coordinator, fail_at) + .err() + .expect("injected spawn failure is returned"); + assert_eq!( + error.to_string(), + format!("injected spawn failure at {fail_at}") + ); + assert_eq!(exits.load(AtomicOrdering::SeqCst), fail_at); + assert!(matches!( + batch_tx.try_send(Vec::new()), + Err(crossbeam::channel::TrySendError::Disconnected(_)) + )); + } + } + + #[test] + fn matcher_join_waits_for_remaining_workers_after_error_or_panic() { + for first_panics in [false, true] { + let (finished_tx, finished_rx) = crossbeam::channel::bounded(1); + let (release_tx, release_rx) = crossbeam::channel::bounded(1); + let (done_tx, done_rx) = crossbeam::channel::bounded(1); + let first = thread::spawn(move || { + assert!(!first_panics, "first worker panic"); + Err(anyhow::anyhow!("first worker failure")) + }); + let deadline = Instant::now() + Duration::from_secs(5); + while !first.is_finished() { + assert!(Instant::now() < deadline, "first worker completes"); + thread::yield_now(); + } + let second = thread::spawn(move || { + finished_tx.send(()).expect("test observes second worker"); + release_rx.recv().expect("second worker released"); + Err(anyhow::anyhow!("later worker failure")) + }); + let coordinator = thread::spawn(move || { + let result = join_matcher_workers(vec![first, second]); + done_tx.send(result).expect("test observes join result"); + }); + let error = + finish_after_worker_release(finished_rx, release_tx, done_rx, coordinator, 1) + .expect_err("first worker error is returned after all workers exit"); + if first_panics { + assert!(error.to_string().contains("Matcher worker 0 panicked")); + } else { + assert_eq!(error.to_string(), "first worker failure"); + } + } + } + fn fragment_ordering_query_response() -> (Vec, Vec) { let mut payload = encode_dns_header(0x3456, 0x0100, 1); payload.extend_from_slice(b"\x07example\x03com\0\0\x01\0\x01"); diff --git a/tools/dns-catalog-builder/src/main.rs b/tools/dns-catalog-builder/src/main.rs index 3b27677..6914652 100644 --- a/tools/dns-catalog-builder/src/main.rs +++ b/tools/dns-catalog-builder/src/main.rs @@ -10,14 +10,18 @@ use csv::StringRecord; use dns_pcap_generator::is_disallowed_domain; use std::collections::HashMap; use std::error::Error as StdError; -use std::fs::{self, File}; +use std::fs::{self, File, OpenOptions}; use std::io::{self, BufReader, BufWriter, Read, Write}; use std::path::{Path, PathBuf}; use std::process::ExitCode; +use std::sync::atomic::{AtomicU64, Ordering}; use thiserror::Error; type Result = std::result::Result; +static TEMPORARY_CATALOG_SEQUENCE: AtomicU64 = AtomicU64::new(0); +const TEMPORARY_CATALOG_ATTEMPTS: usize = 100; + #[derive(Debug, Parser)] #[command( name = "dns-catalog-builder", @@ -254,12 +258,8 @@ fn normalized_name(row: &StringRecord, name_index: usize) -> Option { } fn write_catalog_atomic(output: &Path, entries: &[CatalogEntry]) -> Result<()> { - let temp_path = temporary_output_path(output)?; + let (temporary, file) = TemporaryCatalog::create(output, &TEMPORARY_CATALOG_SEQUENCE)?; { - let file = File::create(&temp_path).map_err(|source| Error::TemporaryCatalogCreate { - path: temp_path.clone(), - source, - })?; let mut writer = BufWriter::new(file); for entry in entries { writeln!(writer, "{}\t{}", entry.count, entry.name).map_err(|source| { @@ -270,39 +270,307 @@ fn write_catalog_atomic(output: &Path, entries: &[CatalogEntry]) -> Result<()> { })?; } writer.flush().map_err(|source| Error::OutputFlush { - path: temp_path.clone(), + path: temporary.path.clone(), source, })?; writer .get_ref() .sync_all() .map_err(|source| Error::OutputSync { - path: temp_path.clone(), + path: temporary.path.clone(), source, })?; } - fs::rename(&temp_path, output).map_err(|source| Error::CatalogRename { - temp_path, - output_path: output.to_path_buf(), - source, - })?; - Ok(()) + temporary.publish(output) +} + +#[derive(Debug)] +struct TemporaryCatalog { + path: PathBuf, + published: bool, +} + +impl TemporaryCatalog { + fn create(output: &Path, sequence: &AtomicU64) -> Result<(Self, File)> { + let mut attempts = 0; + loop { + let id = sequence.fetch_add(1, Ordering::Relaxed); + let path = temporary_output_path(output, id)?; + let candidate_name = path.file_name().expect("temporary path has a file name"); + if output.file_name().is_some_and(|name| { + candidate_name + .as_encoded_bytes() + .eq_ignore_ascii_case(name.as_encoded_bytes()) + }) { + continue; + } + attempts += 1; + match OpenOptions::new().write(true).create_new(true).open(&path) { + Ok(file) => { + return Ok(( + Self { + path, + published: false, + }, + file, + )); + } + Err(source) + if source.kind() == io::ErrorKind::AlreadyExists + && attempts < TEMPORARY_CATALOG_ATTEMPTS => {} + Err(source) => return Err(Error::TemporaryCatalogCreate { path, source }), + } + } + } + + fn publish(mut self, output: &Path) -> Result<()> { + fs::rename(&self.path, output).map_err(|source| Error::CatalogRename { + temp_path: self.path.clone(), + output_path: output.to_path_buf(), + source, + })?; + self.published = true; + Ok(()) + } } -fn temporary_output_path(output: &Path) -> Result { - let file_name = output - .file_name() - .ok_or_else(|| Error::OutputPathMissingFileName { +impl Drop for TemporaryCatalog { + fn drop(&mut self) { + if !self.published { + // Preserve the original I/O error if cleanup itself fails. + let _ = fs::remove_file(&self.path); + } + } +} + +fn temporary_output_path(output: &Path, id: u64) -> Result { + if output.file_name().is_none() { + return Err(Error::OutputPathMissingFileName { path: output.to_path_buf(), - })? - .to_string_lossy(); - Ok(output.with_file_name(format!(".{file_name}.tmp"))) + }); + } + Ok(output.with_file_name(format!(".dns-catalog.{}.{id}.tmp", std::process::id()))) } #[cfg(test)] mod tests { - use super::build_catalog; + use super::{ + CatalogEntry, Error, TEMPORARY_CATALOG_ATTEMPTS, TemporaryCatalog, build_catalog, + temporary_output_path, write_catalog_atomic, + }; + use std::fs; + use std::io::{self, Write}; + use std::path::PathBuf; + use std::sync::atomic::{AtomicU64, Ordering}; + + struct TestDirectory(PathBuf); + + impl TestDirectory { + fn new() -> Self { + static SEQUENCE: AtomicU64 = AtomicU64::new(0); + loop { + let id = SEQUENCE.fetch_add(1, Ordering::Relaxed); + let path = std::env::temp_dir() + .join(format!("dns-catalog-builder-{}-{id}", std::process::id())); + match fs::create_dir(&path) { + Ok(()) => return Self(path), + Err(error) if error.kind() == io::ErrorKind::AlreadyExists => continue, + Err(error) => panic!("create test directory: {error}"), + } + } + } + } + + impl Drop for TestDirectory { + fn drop(&mut self) { + let _ = fs::remove_dir_all(&self.0); + } + } + + #[test] + fn write_catalog_atomic_replaces_complete_catalog() { + let directory = TestDirectory::new(); + let output = directory.0.join("catalog.tsv"); + fs::write(&output, "previous catalog").unwrap(); + + write_catalog_atomic( + &output, + &[ + CatalogEntry { + count: 7, + name: "example.com".into(), + }, + CatalogEntry { + count: 3, + name: "example.org".into(), + }, + ], + ) + .unwrap(); + + assert_eq!( + fs::read_to_string(&output).unwrap(), + "7\texample.com\n3\texample.org\n" + ); + assert_eq!(fs::read_dir(&directory.0).unwrap().count(), 1); + } + + #[cfg(unix)] + #[test] + fn long_output_file_name_publishes_successfully() { + let directory = TestDirectory::new(); + let output = directory.0.join("c".repeat(250)); + write_catalog_atomic( + &output, + &[CatalogEntry { + count: 7, + name: "example.com".into(), + }], + ) + .unwrap(); + assert_eq!(fs::read_to_string(&output).unwrap(), "7\texample.com\n"); + assert_eq!(fs::read_dir(&directory.0).unwrap().count(), 1); + } + + #[test] + fn temporary_candidate_cannot_create_destination_before_publication() { + for uppercase in [false, true] { + let directory = TestDirectory::new(); + let candidate = temporary_output_path(&directory.0.join("catalog.tsv"), 0).unwrap(); + let output = if uppercase { + candidate.with_file_name(candidate.file_name().unwrap().to_ascii_uppercase()) + } else { + candidate + }; + let (temporary, mut file) = + TemporaryCatalog::create(&output, &AtomicU64::new(0)).unwrap(); + file.write_all(b"7\texample.com\n").unwrap(); + file.sync_all().unwrap(); + drop(file); + assert!(!output.exists()); + + temporary.publish(&output).unwrap(); + assert_eq!(fs::read_to_string(&output).unwrap(), "7\texample.com\n"); + assert_eq!(fs::read_dir(&directory.0).unwrap().count(), 1); + } + } + + #[test] + fn overlapping_publisher_abort_preserves_written_catalog() { + let directory = TestDirectory::new(); + let output = directory.0.join("catalog.tsv"); + let (first, mut first_file) = + TemporaryCatalog::create(&output, &AtomicU64::new(0)).unwrap(); + first_file.write_all(b"7\texample.com\n").unwrap(); + + // The second publisher starts after the first has written, then exits before writing. + let (second, second_file) = TemporaryCatalog::create(&output, &AtomicU64::new(0)).unwrap(); + assert_ne!(first.path, second.path); + drop(second_file); + drop(second); + + first_file.sync_all().unwrap(); + drop(first_file); + first.publish(&output).unwrap(); + assert_eq!(fs::read_to_string(&output).unwrap(), "7\texample.com\n"); + assert_eq!(fs::read_dir(&directory.0).unwrap().count(), 1); + } + + #[test] + fn overlapping_publishers_each_replace_with_their_complete_catalog() { + let directory = TestDirectory::new(); + let output = directory.0.join("catalog.tsv"); + let sequence = AtomicU64::new(0); + let (first, mut first_file) = TemporaryCatalog::create(&output, &sequence).unwrap(); + let (second, mut second_file) = TemporaryCatalog::create(&output, &sequence).unwrap(); + first_file.write_all(b"7\texample.com\n").unwrap(); + second_file.write_all(b"3\texample.org\n").unwrap(); + first_file.sync_all().unwrap(); + second_file.sync_all().unwrap(); + drop(first_file); + drop(second_file); + + first.publish(&output).unwrap(); + assert_eq!(fs::read_to_string(&output).unwrap(), "7\texample.com\n"); + second.publish(&output).unwrap(); + assert_eq!(fs::read_to_string(&output).unwrap(), "3\texample.org\n"); + assert_eq!(fs::read_dir(&directory.0).unwrap().count(), 1); + } + + #[test] + fn temporary_catalog_collision_preserves_existing_file() { + let directory = TestDirectory::new(); + let output = directory.0.join("catalog.tsv"); + let existing_temp = temporary_output_path(&output, 0).unwrap(); + fs::write(&output, "previous catalog").unwrap(); + fs::write(&existing_temp, "another publisher").unwrap(); + + let (temporary, file) = TemporaryCatalog::create(&output, &AtomicU64::new(0)).unwrap(); + assert_ne!(temporary.path, existing_temp); + drop(file); + drop(temporary); + + assert_eq!( + fs::read_to_string(&existing_temp).unwrap(), + "another publisher" + ); + assert_eq!(fs::read_to_string(&output).unwrap(), "previous catalog"); + assert_eq!(fs::read_dir(&directory.0).unwrap().count(), 2); + } + + #[test] + fn temporary_catalog_collision_exhaustion_preserves_existing_files() { + let directory = TestDirectory::new(); + let output = directory.0.join("catalog.tsv"); + for id in 0..TEMPORARY_CATALOG_ATTEMPTS as u64 { + fs::write( + temporary_output_path(&output, id).unwrap(), + "another publisher", + ) + .unwrap(); + } + + let error = TemporaryCatalog::create(&output, &AtomicU64::new(0)).unwrap_err(); + assert!(matches!(error, Error::TemporaryCatalogCreate { source, .. } + if source.kind() == io::ErrorKind::AlreadyExists)); + for id in 0..TEMPORARY_CATALOG_ATTEMPTS as u64 { + assert_eq!( + fs::read_to_string(temporary_output_path(&output, id).unwrap()).unwrap(), + "another publisher" + ); + } + assert_eq!( + fs::read_dir(&directory.0).unwrap().count(), + TEMPORARY_CATALOG_ATTEMPTS + ); + } + + #[test] + fn failed_publication_removes_only_its_temporary_file() { + let directory = TestDirectory::new(); + let output = directory.0.join("catalog.tsv"); + fs::create_dir(&output).unwrap(); + fs::write(output.join("keep"), "existing destination").unwrap(); + let unrelated = directory.0.join("another-publisher.tmp"); + fs::write(&unrelated, "another publisher").unwrap(); + + let error = write_catalog_atomic( + &output, + &[CatalogEntry { + count: 7, + name: "example.com".into(), + }], + ) + .unwrap_err(); + assert!(matches!(error, Error::CatalogRename { .. })); + assert_eq!( + fs::read_to_string(output.join("keep")).unwrap(), + "existing destination" + ); + assert_eq!(fs::read_to_string(&unrelated).unwrap(), "another publisher"); + assert_eq!(fs::read_dir(&directory.0).unwrap().count(), 2); + } #[test] fn build_catalog_aggregates_filters_and_sorts_rows() { From 743104708b4a7ab9e7445a1a7461be0b24cef9b3 Mon Sep 17 00:00:00 2001 From: Konstantin Mikhailov Date: Sun, 27 Sep 2026 15:58:27 +0300 Subject: [PATCH 2/2] Bump version --- Cargo.lock | 2 +- Cargo.toml | 2 +- 2 files changed, 2 insertions(+), 2 deletions(-) diff --git a/Cargo.lock b/Cargo.lock index 89cb254..9e10a78 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -665,7 +665,7 @@ dependencies = [ [[package]] name = "dpp" -version = "0.4.0" +version = "0.4.1" dependencies = [ "aes", "anyhow", diff --git a/Cargo.toml b/Cargo.toml index e6c4c7a..f8e8855 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -2,7 +2,7 @@ [package] name = "dpp" -version = "0.4.0" +version = "0.4.1" edition = "2024" rust-version = "1.98.1" build = "build.rs" # Custom build script to include version information from git