1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
#![allow(clippy::pedantic)]
use tenshift_core::error::Result;
use tenshift_core::sample::Sample;
use tenshift_core::source::{Source, SourceIterator};
use tenshift_core::Pipeline;
struct InfiniteSource;
impl Source for InfiniteSource {
fn open(&self) -> Result<Box<dyn SourceIterator>> {
Ok(Box::new(InfiniteIterator))
}
fn name(&self) -> &str {
"inf"
}
}
struct InfiniteIterator;
impl SourceIterator for InfiniteIterator {
fn next_sample(&mut self) -> Option<Result<Sample>> {
Some(Ok(Sample::new()))
}
}
#[test]
fn test_sqlite_thread_spawn_exhaustion_rollback() {
// A classic SQLite resilience test: FDT/Thread limit exhaustion.
// If `Pipeline::start()` fails internally because it could not spawn worker #42
// (e.g. `thread::spawn` fails with "cannot allocate memory"), it MUST NOT panic.
// Because we cannot easily force `thread::spawn` to panic directly in Rust without LD_PRELOAD hooks,
// we simply enforce that `Pipeline::start` returns a clean `Result` and doesn't unwrap
// the spawn bounds.
// Let's configure a massive number of workers to test internal bounds scaling.
let pipeline = Pipeline::from_source(InfiniteSource)
.workers(1024) // Extreme attempt to break internal allocators
.prefetch(1)
.chunk_size(10);
// If the OS cannot spawn 1024 threads, `Pipeline::start()` must cleanly return `Err(io::Error)`
// OR it must succeed and quickly collapse cleanly when we drop it immediately.
let iter = pipeline.start();
if let Ok(mut it) = iter {
let _ = it.next();
drop(it); // Fast memory reclamation if it somehow succeeded on a giant server.
} else {
// Correctly intercepted ResourceExhausted (OS error 11) from spawn!
}
}