tenshift-core 0.1.3

#![allow(clippy::pedantic)]

use std::time::Duration;

use tenshift_core::error::{Error, Result};
use tenshift_core::pipeline::ErrorPolicy;
use tenshift_core::sample::{Sample, Tensor};
use tenshift_core::source::{Source, SourceIterator};
use tenshift_core::Pipeline;

/// A source that specifically simulates Linux kernel behavior under severe I/O load,
/// mimicking the exact semantics of `EAGAIN` (WouldBlock) and `EINTR` (Interrupted System Call).
/// In standard read() loops, these must be handled silently without bubbling out to the user,
/// simply stalling until data is ready. If tenshift crashes on sporadic source delays, it fails.
struct PosixInterruptSource {
    target_samples: usize,
    eagain_probability: u8,
    eintr_probability: u8,
    rng_state: u64,
}

impl Source for PosixInterruptSource {
    fn open(&self) -> Result<Box<dyn SourceIterator>> {
        Ok(Box::new(PosixInterruptIterator {
            yielded_count: 0,
            target_samples: self.target_samples,
            eagain_prob: self.eagain_probability,
            eintr_prob: self.eintr_probability,
            state: self.rng_state,
        }))
    }

    fn name(&self) -> &str {
        "posix-interrupt-sim"
    }
}

struct PosixInterruptIterator {
    yielded_count: usize,
    target_samples: usize,
    eagain_prob: u8,
    eintr_prob: u8,
    state: u64,
}

impl PosixInterruptIterator {
    fn next_rand(&mut self) -> u8 {
        self.state ^= self.state << 13;
        self.state ^= self.state >> 17;
        self.state ^= self.state << 5;
        (self.state % 100) as u8
    }
}

impl SourceIterator for PosixInterruptIterator {
    fn next_sample(&mut self) -> Option<Result<Sample>> {
        if self.yielded_count >= self.target_samples {
            return None;
        }

        let roll = self.next_rand();

        // Simulate EAGAIN: The OS buffer is temporarily empty.
        // A true POSIX program must yield or wait. Tenshift should just sleep or handle it transparently.
        if roll < self.eagain_prob {
            std::thread::sleep(Duration::from_micros(10));
            // This does NOT return an error to the pipeline, it just delays the return.
            // If the timeout boundary is robust, the pipeline will stall gracefully.
        }

        // Simulate EINTR: The kernel interrupted the read() syscall for an OS signal,
        // but no data was corrupted. The program must explicitly retry.
        if roll >= self.eagain_prob && roll < self.eagain_prob + self.eintr_prob {
            std::thread::sleep(Duration::from_micros(5));
            self.yielded_count += 1;
            return Some(Err(Error::TransformFailed {
                index: 0,
                reason: "System call interrupted".into(),
            }));
        }

        let mut s = Sample::new();
        s.insert("data", Tensor::i64(&[self.yielded_count as i64], vec![1]));
        self.yielded_count += 1;
        Some(Ok(s))
    }
}

#[test]
fn test_sqlite_eagain_and_eintr_starvation_resistance() {
    // Ensure the collector doesn't deadlock gaps entirely, and mathematical completion matches probability exactly.

    let pipeline = Pipeline::from_source(PosixInterruptSource {
        target_samples: 50_000,
        eagain_probability: 30, // 30% chance each sample causes a thread stall
        eintr_probability: 40,  // 40% chance each sample generates a recoverable read error
        rng_state: 0xDEAD_BEEF,
    })
    .workers(16)
    .chunk_size(4)
    .prefetch(128)
    .on_error(ErrorPolicy::Skip)
    .sequence_gap_timeout(Duration::from_secs(5))
    .source_timeout(Duration::from_secs(5))
    .batch(100);

    let mut successes = 0;

    let iter = pipeline.start().expect("init");
    for batch in iter {
        let n = batch[0].get("data").unwrap().shape()[0];
        successes += n;
    }

    // Since EINTR probability is exactly 40%, we expect roughly exactly 30,000 successful samples.
    // The pipeline MUST NOT deadlock or fail.
    assert!(
        successes > 28_000 && successes < 32_000,
        "Pipeline corrupted data recovery rate under EINTR fragmentation"
    );
}