aion-server 0.31.0

//! Test-only filesystem and tracing helpers.

use std::cell::RefCell;
use std::io::{self, Write};
use std::sync::{Arc, Mutex, Once};

use tempfile::TempDir;

/// A `tracing` writer that accumulates formatted events in memory, so a test
/// can assert on what the server actually told the operator.
#[derive(Clone, Default)]
pub(crate) struct CapturedLogs(Arc<Mutex<Vec<u8>>>);

thread_local! {
    /// The buffer the CURRENT thread's capture routes events into, when one
    /// is active. Events on threads with no active capture are dropped.
    static ACTIVE_SINK: RefCell<Option<Arc<Mutex<Vec<u8>>>>> = const { RefCell::new(None) };
}

/// One-time installer for the process-wide capture subscriber.
static INSTALL_CAPTURE_SUBSCRIBER: Once = Once::new();

/// Why the capture subscriber could not be installed, when it could not.
/// Every subsequent capture writes this into its own buffer so the failure
/// is loud in the assertion that would otherwise see silence.
static INSTALL_ERROR: Mutex<Option<String>> = Mutex::new(None);

/// The per-event writer: forwards to the current thread's active sink.
pub(crate) struct CapturedWriter;

impl Write for CapturedWriter {
    fn write(&mut self, buffer: &[u8]) -> io::Result<usize> {
        ACTIVE_SINK.with(|sink| {
            if let Some(bytes) = sink.borrow().as_ref() {
                let mut bytes = bytes
                    .lock()
                    .map_err(|_| io::Error::other("captured log lock poisoned"))?;
                bytes.extend_from_slice(buffer);
            }
            Ok(buffer.len())
        })
    }

    fn flush(&mut self) -> io::Result<()> {
        Ok(())
    }
}

/// Clears the thread's active sink when a capture ends, even on panic.
struct SinkGuard;

impl Drop for SinkGuard {
    fn drop(&mut self) {
        ACTIVE_SINK.with(|sink| sink.borrow_mut().take());
    }
}

impl CapturedLogs {
    /// Run `body` with every `tracing` event emitted ON THIS THREAD routed
    /// into this buffer.
    ///
    /// Mechanically this is ONE process-wide subscriber (installed on first
    /// use) writing through a thread-local sink, not a scoped `with_default`
    /// per capture. The scoped form registers and drops a dispatcher per
    /// capture, and every registration rebuilds `tracing`'s global
    /// per-callsite interest cache — a rebuild landing on another thread
    /// mid-event silently drops that event, which surfaced as intermittent,
    /// schedule-dependent misses of log assertions across the parallel test
    /// harness. A single permanent subscriber keeps callsite interest stable
    /// forever, so no capture can lose an event to another capture's
    /// lifecycle. Events on OTHER threads (including ones `body` spawns) are
    /// not captured — the same visibility the scoped form had.
    pub(crate) fn capture<T>(body: impl FnOnce() -> T) -> (Self, T) {
        INSTALL_CAPTURE_SUBSCRIBER.call_once(|| {
            let subscriber = tracing_subscriber::fmt()
                .without_time()
                .with_ansi(false)
                .with_writer(|| CapturedWriter)
                .finish();
            if let Err(error) = tracing::subscriber::set_global_default(subscriber) {
                // Loud, not silent: a foreign global subscriber means every
                // capture in this process would assert against an empty
                // buffer. No test process installs one before a capture; if
                // that ever changes, every capture assertion must fail
                // naming the cause, so record it for every capture to carry.
                if let Ok(mut slot) = INSTALL_ERROR.lock() {
                    *slot = Some(format!("capture subscriber not installed: {error}"));
                }
            }
        });
        let captured = Self::default();
        if let Some(reason) = INSTALL_ERROR
            .lock()
            .unwrap_or_else(std::sync::PoisonError::into_inner)
            .as_deref()
            && let Ok(mut bytes) = captured.0.lock()
        {
            bytes.extend_from_slice(reason.as_bytes());
        }
        ACTIVE_SINK.with(|sink| {
            *sink.borrow_mut() = Some(Arc::clone(&captured.0));
        });
        let guard = SinkGuard;
        let value = body();
        drop(guard);
        (captured, value)
    }

    /// The captured events as text, or an error if a writer panicked mid-write.
    pub(crate) fn text(&self) -> Result<String, Box<dyn std::error::Error>> {
        let bytes = self.0.lock().map_err(|_| "captured log lock poisoned")?;
        Ok(String::from_utf8(bytes.clone())?)
    }
}

/// Tightens an already-created directory to mode `0700` so a test-built
/// workspace root starts from the mode the server would give it, rather than
/// whatever the harness umask happened to produce.
pub(crate) fn make_private(path: &std::path::Path) -> std::io::Result<()> {
    #[cfg(unix)]
    {
        use std::os::unix::fs::PermissionsExt;
        std::fs::set_permissions(path, std::fs::Permissions::from_mode(0o700))?;
    }
    #[cfg(not(unix))]
    {
        let _ = path;
    }
    Ok(())
}

/// Creates a temporary directory whose mode is `0700` regardless of the
/// process umask.
///
/// `tempfile::tempdir` inherits the umask, so under a conventional `022` a
/// fresh directory is `0755`. The server would now tighten such a directory on
/// sight, which would silently mutate the fixture a test is asserting against —
/// so tests that hand a temporary directory to any sensitive-root surface go
/// through this helper and start from a known mode under every umask.
pub(crate) fn private_tempdir() -> std::io::Result<TempDir> {
    let dir = tempfile::tempdir()?;
    #[cfg(unix)]
    {
        use std::os::unix::fs::PermissionsExt;
        std::fs::set_permissions(dir.path(), std::fs::Permissions::from_mode(0o700))?;
    }
    Ok(dir)
}

/// What a fixture teardown failure is called in the harness output, so the
/// line is greppable whichever way it was reported, and whichever target
/// reported it: the integration targets' own guards use this same wording.
const TEARDOWN_FAILED: &str = "aion-server test fixture teardown failed";

/// An engine a fixture built, stopped when the fixture's binding ends.
///
/// # Why a fixture needs this
///
/// `Drop for Engine` deliberately leaves the beamr scheduler and the engine's
/// NIF seams alive: it closes the engine-task epoch and stops the timer wheel,
/// but stopping the scheduler and clearing the seams is
/// [`aion::Engine::shutdown`]'s job and nothing else's. That is the engine's
/// contract — shutdown is the caller's act — so a fixture that builds an engine
/// and merely drops it leaks the scheduler's threads and, through the
/// uncleared seams, every store clone the seams reach. A test target that
/// builds hundreds of engines in one process therefore climbs towards the
/// per-process thread cap and starts failing later spawns with an operating
/// system error, in whichever test happens to be running when the cap is hit.
///
/// Holding the engine in this guard makes the fixture keep the engine's side
/// of that contract. [`Self::shutdown`] is idempotent, so a test may stop the
/// engine explicitly at its end and still leave the guard in place as the net
/// that covers every early return.
pub(crate) struct EngineUnderTest {
    /// The engine this guard stops.
    engine: std::sync::Arc<aion::Engine>,
    /// Whether the engine has already been stopped through this guard.
    stopped: std::sync::atomic::AtomicBool,
}

impl EngineUnderTest {
    /// Take ownership of a fixture-built engine.
    pub(crate) fn new(engine: std::sync::Arc<aion::Engine>) -> Self {
        Self {
            engine,
            stopped: std::sync::atomic::AtomicBool::new(false),
        }
    }

    /// A counted handle on the guarded engine, for a fixture that hands the
    /// engine to something which takes ownership of it. The guard keeps its
    /// own handle, so the engine is still stopped when the guard ends.
    pub(crate) fn handle(&self) -> std::sync::Arc<aion::Engine> {
        std::sync::Arc::clone(&self.engine)
    }

    /// Give up responsibility for stopping the engine, because another owner
    /// has taken it and has already stopped it.
    ///
    /// This is how an engine passes from the guard that owned it at creation
    /// to the [`StateUnderTest`] built over it WITHOUT the engine being
    /// stopped twice: the state's shutdown is the one that runs, and this
    /// leaves the engine guard's own `Drop` a no-op. It is deliberately NOT
    /// called when the state's shutdown failed, so an engine whose production
    /// teardown path refused is still stopped by the guard that created it.
    pub(crate) fn disarm(&self) {
        self.stopped
            .store(true, std::sync::atomic::Ordering::SeqCst);
    }

    /// Stop the engine and hand the outcome to the test. Idempotent: a second
    /// call (including the one `Drop` makes) is a no-op that reports success.
    pub(crate) fn shutdown(&self) -> Result<(), aion::EngineError> {
        if self.stopped.swap(true, std::sync::atomic::Ordering::SeqCst) {
            return Ok(());
        }
        self.engine.shutdown()
    }
}

impl std::ops::Deref for EngineUnderTest {
    type Target = std::sync::Arc<aion::Engine>;

    fn deref(&self) -> &Self::Target {
        &self.engine
    }
}

impl Drop for EngineUnderTest {
    fn drop(&mut self) {
        report_teardown(self.shutdown().err().map(|error| error.to_string()));
    }
}

/// A [`ServerState`](crate::ServerState) a fixture built, shut down when the
/// fixture's binding ends.
///
/// Shutdown goes through `ServerState::shutdown`, the same call the running
/// server makes, so a test tears down exactly what production tears down: the
/// namespace resolver's engine handle, by way of [`aion::Engine::shutdown`].
/// The state derefs to the wrapped `ServerState`, so a fixture hands a clone
/// to a router or a handler and keeps the guard itself for the test's length.
///
/// A state whose resolver carries no engine has nothing to stop and must not
/// be wrapped: `ServerState::shutdown` reports the missing handle as an error,
/// which this guard would correctly turn into a test failure.
///
/// # Why it can also hold the engine guard
///
/// An engine is guarded on the line that BUILDS it, not on the line that
/// finally wraps a state around it: everything in between — recording
/// ownership, seeding history, fetching a fixture JWKS inside
/// [`crate::api::http::test_support::server_state`] — is fallible, and an `?`
/// there would otherwise drop a live engine that nobody had undertaken to
/// stop. [`Self::over`] takes that creation-time guard by value, so the engine
/// is owned for every instant of its life and exactly one owner stops it.
pub(crate) struct StateUnderTest {
    /// The state this guard shuts down.
    state: crate::ServerState,
    /// The guard the engine was created under, when the caller had one. Held
    /// so a state-shutdown failure still leaves the engine stopped, and
    /// disarmed on success so the engine is never stopped twice.
    engine: Option<EngineUnderTest>,
    /// Whether the state has already been shut down through this guard.
    stopped: std::sync::atomic::AtomicBool,
}

impl StateUnderTest {
    /// Take ownership of a state whose engine was built INSIDE it (the
    /// `ServerState::build*` paths), so there is no separate engine handle.
    pub(crate) fn new(state: crate::ServerState) -> Self {
        Self {
            state,
            engine: None,
            stopped: std::sync::atomic::AtomicBool::new(false),
        }
    }

    /// Take ownership of a state AND of the guard its engine was created
    /// under, so the engine has a single owner from creation to teardown.
    pub(crate) fn over(engine: EngineUnderTest, state: crate::ServerState) -> Self {
        Self {
            state,
            engine: Some(engine),
            stopped: std::sync::atomic::AtomicBool::new(false),
        }
    }

    /// Shut the state's engine down and hand the outcome to the test.
    /// Idempotent, for the same reason [`EngineUnderTest::shutdown`] is.
    pub(crate) fn shutdown(&self) -> Result<(), crate::ServerError> {
        if self.stopped.swap(true, std::sync::atomic::Ordering::SeqCst) {
            return Ok(());
        }
        // The production path is the one that runs: `ServerState::shutdown`
        // reaches `Engine::shutdown` through the namespace resolver, exactly as
        // the running server does. Only once it has SUCCEEDED is the
        // creation-time guard disarmed; if it refused, that guard stays armed
        // and stops the engine itself rather than leaving it running.
        let outcome = self.state.shutdown();
        if outcome.is_ok()
            && let Some(engine) = &self.engine
        {
            engine.disarm();
        }
        outcome
    }
}

impl std::ops::Deref for StateUnderTest {
    type Target = crate::ServerState;

    fn deref(&self) -> &Self::Target {
        &self.state
    }
}

impl Drop for StateUnderTest {
    fn drop(&mut self) {
        report_teardown(self.shutdown().err().map(|error| error.to_string()));
    }
}

/// Report a fixture teardown outcome from a `Drop`.
///
/// A teardown failure is a test failure, and an assertion is how a test fails,
/// so the assertion carries the error's own words. The one exception is a drop
/// running while the thread is ALREADY unwinding: a panic raised during unwind
/// aborts the process, which would erase the failure the test was in the
/// middle of reporting. There the error goes to stderr beside that failure
/// instead.
fn report_teardown(failure: Option<String>) {
    if std::thread::panicking() {
        if let Some(message) = failure {
            eprintln!("{TEARDOWN_FAILED}: {message}");
        }
        return;
    }
    assert!(
        failure.is_none(),
        "{TEARDOWN_FAILED}: {}",
        failure.as_deref().unwrap_or_default()
    );
}