liminal-server 0.25.4

//! Delivery flush completion must be told by removal, not inferred at timeout.

use std::sync::Arc;
use std::sync::atomic::Ordering;
use std::sync::mpsc::{Receiver, channel};
use std::thread::JoinHandle;
use std::time::{Duration, Instant};

use super::{ConnectionSupervisor, recover_lock};

type TestResult = Result<(), Box<dyn std::error::Error>>;
const EXITED_PID: u64 = u64::MAX;

fn waiting_flush(
    supervisor: &ConnectionSupervisor,
    budget: Duration,
) -> Result<(JoinHandle<()>, Receiver<bool>), Box<dyn std::error::Error>> {
    // A scheduler-exited connection awaiting reclamation still owns a record.
    // It cannot run another slice or park; removal is its only completion edge.
    supervisor.inner.runtime.register(EXITED_PID, None)?;
    let (entered, observed_entry) = channel();
    *recover_lock(&supervisor.inner.runtime.settle_wait_observer) = Some(entered);
    let runtime = Arc::clone(&supervisor.inner.runtime);
    let (finished, observed_finish) = channel();
    let worker = std::thread::spawn(move || {
        let outcome = runtime.wait_for_delivery_quiesced(Instant::now() + budget);
        let _sent = finished.send(outcome);
    });
    observed_entry.recv_timeout(Duration::from_secs(2))?;
    Ok((worker, observed_finish))
}

#[test]
fn last_connection_removal_wakes_delivery_flush_before_its_deadline() -> TestResult {
    let supervisor = ConnectionSupervisor::new()?;
    let (worker, completion) = waiting_flush(&supervisor, Duration::from_secs(5))?;
    let removed = supervisor.inner.runtime.remove(EXITED_PID);
    let outcome = completion.recv_timeout(Duration::from_secs(1));
    worker.join().map_err(|_| "flush waiter panicked")?;
    supervisor.shutdown();
    assert!(removed.is_some());
    assert!(
        outcome?,
        "last removal must tell the flush barrier it is complete"
    );
    Ok(())
}

#[test]
fn delivery_flush_deadline_is_failure_even_if_a_silent_state_change_looks_quiescent() -> TestResult
{
    let supervisor = ConnectionSupervisor::new()?;
    let (worker, completion) = waiting_flush(&supervisor, Duration::from_millis(200))?;
    // Negative control: deliberately change the predicate WITHOUT its wake.
    // This reproduces the missed-notification state; the deadline must not
    // disguise it as observed completion. Production uses mark_parked/remove.
    {
        let records = recover_lock(&supervisor.inner.runtime.records);
        let record = records
            .connections
            .get(&EXITED_PID)
            .ok_or("missing held connection")?;
        record.parked.store(true, Ordering::Release);
        record.ready_pending.store(false, Ordering::Release);
        drop(records);
    }
    let outcome = completion.recv_timeout(Duration::from_secs(2));
    worker.join().map_err(|_| "flush waiter panicked")?;
    drop(supervisor.inner.runtime.remove(EXITED_PID));
    supervisor.shutdown();
    assert!(!outcome?, "a deadline cannot certify a silent completion");
    Ok(())
}