liminal_server/server/connection/

supervisor.rs

#[path = "supervisor/force_close.rs"]
mod force_close;
#[cfg(test)]
#[path = "process/force_close_tests.rs"]
mod force_close_tests;
#[path = "supervisor/retirement.rs"]
mod retirement;

use retirement::NativeRetirements;

#[path = "supervisor/routed_push.rs"]
mod routed_push;
#[path = "supervisor/tap_placements.rs"]
mod tap_placements;
use std::collections::HashMap;
#[cfg(test)]
use std::collections::VecDeque;
use std::collections::hash_map::Entry;
use std::net::{SocketAddr, TcpStream};
use std::os::fd::RawFd;
#[cfg(test)]
use std::sync::Barrier;
use std::sync::atomic::{AtomicBool, AtomicU64, Ordering};
use std::sync::mpsc::{Receiver, RecvTimeoutError, Sender, TryRecvError, channel};
use std::sync::{Arc, Condvar, Mutex, MutexGuard, OnceLock, PoisonError, Weak};
use std::time::{Duration, Instant};

use beamr::atom::{Atom, AtomTable};
use beamr::module::ModuleRegistry;
use beamr::native::native_process::NativeHandlerFactory;
use beamr::process::ExitReason;
use beamr::scheduler::{NativeBifs, ReadinessToken, Scheduler, SchedulerConfig, SchedulerServices};
use beamr::timer::TimerRef;

use liminal::protocol::{Frame, WorkerRefusal, WorkerRegistration};
use liminal_protocol::wire::ConnectionIncarnation;

use super::incarnation::ConnectionIncarnationAuthority;
use super::loopback::{LoopbackConnectionProcess, LoopbackServerEnd};
use super::notifier::ConnectionNotifier;
use super::outbound::OutboundError;
use super::process::ConnectionProcess;
use super::refusal::AdmissionRefusal;
#[cfg(any(test, feature = "test-support"))]
use super::services::LiminalConnectionServices;
use super::services::{ConnectionServices, build_connection_services};
#[cfg(test)]
use super::services::{SubsystemFactory, build_connection_services_via};
use crate::ServerError;
use crate::auth_pass::{PassPrincipal, PassVerifier};
use crate::config::types::{AuthConfig, LimitsConfig, ServerConfig};
use crate::health::AdmissionReadiness;
use crate::server::mount::MountKind;
use crate::server::participant::{
    CandidateLaneListener, ConnectionFateClass, InstalledParticipantService,
    ParticipantSemanticHandler, ParticipantServiceFatal,
};
use crate::server::shutdown::ShutdownHandle;
use crate::server::signals::ServerSignals;

const CONNECTION_SCHEDULER_THREADS: usize = 4;
const CONNECTION_SHUTDOWN_CONTROL_ATOM: &str = "liminal_server_connection_shutdown_control";
/// R6 (§1.2(4)): the single `READY` wake vocabulary for a connection. One atom;
/// any marker (or N coalesced) triggers one full slice servicing all sources.
const CONNECTION_READY_ATOM: &str = "liminal_server_connection_ready";

#[path = "supervisor_fate.rs"]
mod fate;
use fate::ParkedConnectionFates;
#[cfg(test)]
#[path = "supervisor_fate_tests.rs"]
mod fate_tests;
#[path = "supervisor_lifecycle.rs"]
mod lifecycle;
#[path = "supervisor_reclaim.rs"]
mod reclaim;
#[cfg(test)]
use reclaim::run_for_test as run_reclaim_reactor;
#[path = "relay_pump.rs"]
pub(super) mod relay_pump;
#[cfg(test)]
#[path = "../shutdown_store_tests.rs"]
mod shutdown_store_tests;
#[cfg(test)]
#[path = "supervisor_shutdown_wake_tests.rs"]
mod shutdown_wake_tests;
#[cfg(test)]
#[path = "supervisor_tests.rs"]
mod tests;

#[cfg(test)]
#[path = "worker_refusal_tests.rs"]
mod worker_refusal_tests;

/// Supervisor that owns the beamr scheduler for per-connection processes.
#[derive(Clone, Debug)]
pub struct ConnectionSupervisor {
    inner: Arc<SupervisorInner>,
}

use super::{service_directory, services, state, wake};
use builder::{
    ConnectionSupervisorBuilder, LysInstall, configured_authentication, configured_lys, install_lys,
};
#[cfg(test)]
use handle::PushReplyFailure;
use handle::{ConnectionHandle, PushReplyAwaiter};
use inner::SupervisorInner;
use push_types::{
    PendingPush, PushBinding, PushReplyDisposition, PushSlotDisposition, PushSlotResolution,
};
use registry::{
    ConnectionRecord, QueuedConnectionControl, ReadinessRegistration, Registry, lock, recover_lock,
};
use runtime::{
    ActiveConnection, AdmissionReservation, ConnectionControl, ConnectionRuntime,
    ConnectionRuntimeInstallation, ConnectionServiceInstallation,
};
#[cfg(test)]
use runtime::{CloseEventFdProbe, PreWaitBarrier, ReclaimBarrier};

#[path = "supervisor/builder.rs"]
pub(super) mod builder;
#[path = "supervisor/context_watch.rs"]
pub(super) mod context_watch;
#[path = "supervisor/handle.rs"]
pub(super) mod handle;
#[path = "supervisor/inner.rs"]
mod inner;
#[path = "supervisor/push_types.rs"]
pub(super) mod push_types;
#[path = "supervisor/registry.rs"]
mod registry;
#[path = "supervisor/runtime.rs"]
pub(super) mod runtime;
#[path = "supervisor/runtime_admission.rs"]
mod runtime_admission;
#[path = "supervisor/runtime_authority.rs"]
mod runtime_authority;
#[path = "supervisor/runtime_close.rs"]
mod runtime_close;
#[path = "supervisor/runtime_control.rs"]
mod runtime_control;
#[path = "supervisor/runtime_fate.rs"]
mod runtime_fate;
#[path = "supervisor/runtime_notifier.rs"]
mod runtime_notifier;
#[path = "supervisor/runtime_push_admission.rs"]
mod runtime_push_admission;
#[path = "supervisor/runtime_push_resolution.rs"]
mod runtime_push_resolution;
#[path = "supervisor/runtime_readiness.rs"]
mod runtime_readiness;
#[path = "supervisor/runtime_registry.rs"]
mod runtime_registry;
#[path = "supervisor/runtime_settlement.rs"]
mod runtime_settlement;
#[path = "supervisor/runtime_shutdown.rs"]
mod runtime_shutdown;
#[cfg(test)]
#[path = "supervisor/runtime_test_barriers.rs"]
mod runtime_test_barriers;
#[cfg(test)]
#[path = "supervisor/runtime_test_construction.rs"]
mod runtime_test_construction;
#[cfg(test)]
#[path = "supervisor/runtime_test_observation.rs"]
mod runtime_test_observation;
#[cfg(test)]
#[path = "supervisor/shutdown_notifier_tests.rs"]
mod shutdown_notifier_tests;
#[path = "supervisor/supervisor_authority.rs"]
mod supervisor_authority;
#[path = "supervisor/supervisor_config.rs"]
mod supervisor_config;
#[path = "supervisor/supervisor_connections.rs"]
mod supervisor_connections;
#[path = "supervisor/supervisor_push.rs"]
mod supervisor_push;
#[path = "supervisor/supervisor_shutdown.rs"]
mod supervisor_shutdown;
#[cfg(test)]
#[path = "supervisor/supervisor_test_support.rs"]
mod supervisor_test_support;
#[cfg(test)]
#[path = "supervisor/worker_refusal_ticket_fixture.rs"]
mod worker_refusal_ticket_fixture;

#[test]
fn host_fd_is_closed_before_connection_close_event_handoff()
-> Result<(), Box<dyn std::error::Error>> {
    use std::os::fd::AsRawFd;
    let fixture = crate::server::self_streams::tests::R3Fixture::new()?;
    let services: Arc<dyn ConnectionServices> = fixture.services;
    let runtime = ConnectionRuntime::for_tests(services);
    runtime
        .signals
        .set(ServerSignals::default())
        .map_err(|_| "close-event signals already installed")?;
    let listener = std::net::TcpListener::bind("127.0.0.1:0")?;
    let client = TcpStream::connect(listener.local_addr()?)?;
    let (server, address) = listener.accept()?;
    let host_fd = server.as_raw_fd();
    let positive_control = std::fs::File::open(format!("/dev/fd/{host_fd}"))?;
    drop(positive_control);
    runtime.register_record(1, Some(address), None, MountKind::Tcp, Some(server))?;
    *runtime
        .close_event_fd_probe
        .lock()
        .map_err(|error| error.to_string())? = CloseEventFdProbe::Armed;
    let removed = runtime
        .remove(1)
        .ok_or("fd-owning connection was not removed")?;
    let mut probe = runtime
        .close_event_fd_probe
        .lock()
        .map_err(|error| error.to_string())?;
    let CloseEventFdProbe::Observed(observed) =
        std::mem::replace(&mut *probe, CloseEventFdProbe::Off)
    else {
        return Err("close-event handoff was not observed".into());
    };
    drop(probe);
    assert!(
        observed?,
        "connection-close event was handed off while its host fd remained open"
    );
    assert!(removed.fd_guard.is_none());
    drop(removed);
    drop(client);
    drop(listener);
    Ok(())
}