liminal-server 0.23.1

use super::{
    Arc, Atom, AtomicBool, AtomicU64, Condvar, ConnectionIncarnationAuthority, ConnectionNotifier,
    ConnectionServices, HashMap, InstalledParticipantService, LimitsConfig, Mutex, OnceLock,
    ParkedConnectionFates, PassVerifier, PendingPush, QueuedConnectionControl, Registry, Scheduler,
    ServerSignals, ShutdownHandle, SocketAddr, Weak,
};
#[cfg(test)]
use super::{Barrier, VecDeque};
#[cfg(test)]
use super::{RawFd, Sender};

/// RAII guard for one §5 `max_connections` admission reservation.
///
/// Acquired (via [`ConnectionRuntime::try_reserve_admission`]) before any process
/// construction in `spawn_connection`; every early-return failure path releases
/// it through `Drop`, and a successful `register` converts it into the
/// connection record (whose removal releases the admission instead). RAII means
/// no failure path — present or future — can leak a reservation.
pub(super) struct AdmissionReservation<'a> {
    pub(super) runtime: &'a ConnectionRuntime,
    pub(super) armed: bool,
}

impl AdmissionReservation<'_> {
    /// Converts the reservation into record ownership: `Drop` no longer releases
    /// it, because the registered record's removal will.
    pub(super) fn convert(mut self) {
        self.armed = false;
    }
}

impl Drop for AdmissionReservation<'_> {
    fn drop(&mut self) {
        if self.armed {
            self.runtime.release_admission();
        }
    }
}

#[derive(Debug, Clone, PartialEq, Eq)]
pub(in crate::server::connection) enum ConnectionControl {
    NotifyShutdown,
    ForceClose,
    /// Server-initiated push of an opaque payload, correlated by `correlation_id`,
    /// to be written out as a [`Frame::Push`] by the receiving connection process.
    Push {
        correlation_id: u64,
        payload: Vec<u8>,
    },
    /// ACCESS-005 R4/R6: this connection's rights changed under a grant
    /// revocation. The connection sweeps its own `ConnectionGrants`: it
    /// retires withdrawn rights' work and, when its admitting right is lost,
    /// closes with the reason. The denied state is already in force before
    /// this control is read, so a refused publication only delays the sweep
    /// to the connection's own wake.
    Revocation,
}

#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub(super) struct ActiveConnection {
    pub(super) pid: u64,
    pub(super) peer_addr: Option<SocketAddr>,
}

#[cfg(test)]
#[derive(Debug, Clone)]
pub(super) struct PreWaitBarrier {
    pub(super) armed: Arc<Barrier>,
    pub(super) release: Arc<Barrier>,
}

/// Pid-specific, one-use deterministic gate on the reclamation delivery path
/// (oracle 26). The exit-event reactor stages it only for the targeted pid, so
/// unrelated exits still reclaim immediately; for the target it rendezvouses
/// `reached` (proving the pid is dead but its record still tracked — the S8
/// reclamation window), then `release` (the harness lets the reclaim proceed),
/// then `done` (the `remove()` funnel completed). It changes no production
/// semantics — the whole type and its call sites are `#[cfg(test)]`.
#[cfg(test)]
#[derive(Debug, Clone)]
pub(super) struct ReclaimBarrier {
    pub(super) pid: u64,
    pub(super) reached: Arc<Barrier>,
    pub(super) release: Arc<Barrier>,
    pub(super) done: Arc<Barrier>,
}

#[derive(Debug)]
pub(super) struct ConnectionServiceInstallation {
    pub(super) services: Arc<dyn ConnectionServices>,
    pub(super) participant_service: Option<InstalledParticipantService>,
}

impl ConnectionServiceInstallation {
    /// Captures the service adapter's capability posture exactly once, before
    /// participant incarnation startup or connection process construction.
    pub(super) fn capture(services: Arc<dyn ConnectionServices>) -> Self {
        let participant_service = services.participant_service();
        Self {
            services,
            participant_service,
        }
    }
}

#[derive(Debug)]
pub(super) struct ConnectionRuntimeInstallation {
    pub(super) services: ConnectionServiceInstallation,
    pub(super) incarnations: Option<Arc<ConnectionIncarnationAuthority>>,
    pub(super) fatal_shutdown: Option<ShutdownHandle>,
}

#[cfg(test)]
#[derive(Debug)]
pub(super) enum CloseEventFdProbe {
    Off,
    Armed,
    Observed(Result<bool, std::io::Error>),
}

#[derive(Debug)]
pub(in crate::server::connection) struct ConnectionRuntime {
    pub(in crate::server::connection) signals: OnceLock<ServerSignals>,
    pub(in crate::server::connection) link_life: crate::link::LinkRegistry,
    /// The Lys pass admitter and the Lys network id liminal's subjects are
    /// placed under, installed once before the first connection (ACCESS-001
    /// R2). Absent, a Lys pass record is refused at `Connect` by name.
    pub(super) lys_admitter: OnceLock<(Arc<dyn crate::auth_lys::session::LysAdmitter>, String)>,
    /// The ACCESS-003 service bindings, installed once from
    /// `[auth.lys.services]`; a service request without them is refused by
    /// name.
    pub(super) service_bindings: OnceLock<Arc<crate::services::production::ServiceBindings>>,
    /// The services registry and lifecycle (SRV-028), installed once after
    /// the listener binds; absent, every lifecycle request is refused by
    /// name and no service process credential is admitted.
    pub(super) lifecycle: OnceLock<Arc<crate::services::lifecycle::Lifecycle>>,
    /// The channel membership gate, installed once with the admitter
    /// (ACCESS-006 R1). Absent, no channel is membership-protected.
    pub(super) membership_gate: OnceLock<Arc<crate::auth_membership::MembershipGate>>,
    /// This liminal as a machine Lys knows, installed once with the admitter
    /// (ACCESS-002 R1). Absent, no link opens here, by name.
    pub(super) link_host: OnceLock<Arc<crate::link::LinkHost>>,
    /// The grant-revocation policy, installed once before the first
    /// connection (ACCESS-005). Present, every `Connect` must carry a bound
    /// pass the authority admits and negotiate the reason-bearing close;
    /// absent, admission is exactly as before.
    pub(super) revocation: OnceLock<Arc<crate::auth_revocation::RevocationAuthority>>,
    #[cfg(test)]
    pub(super) close_event_fd_probe: Mutex<CloseEventFdProbe>,
    pub(super) services: Arc<dyn ConnectionServices>,
    /// Live retry index woken by candidate-lane drains; durable Open rows remain
    /// the recovery authority.
    pub(super) parked_connection_fates: Mutex<ParkedConnectionFates>,
    /// Complete participant handler/store bundle captured at supervisor startup.
    /// `Some` is paired with an incarnation authority on `SupervisorInner`.
    pub(super) participant_service: Option<InstalledParticipantService>,
    /// Same started authority used for allocation, shared for terminal Open/Complete.
    pub(super) incarnations: Option<Arc<ConnectionIncarnationAuthority>>,
    /// Existing runtime shutdown activation notified by the first post-Open fatal.
    /// Test-only/runtime-less constructors intentionally carry `None`.
    pub(super) fatal_shutdown: Option<ShutdownHandle>,
    pub(super) records: Mutex<Registry>,
    pub(super) controls: Mutex<Vec<QueuedConnectionControl>>,
    pub(super) control_atom: Atom,
    /// R6 single `READY` wake atom for this connection scheduler. Fired by every
    /// wake source's notifier (R3/R1(vi)); coalescing and duplicates are harmless.
    pub(super) ready_atom: Atom,
    /// Weak handle to the connection scheduler, used to build [`ReadyWaker`]s a
    /// notifier fires from another actor's slice. Weak so it never keeps the
    /// scheduler alive (the scheduler owns the processes that own this runtime).
    pub(super) scheduler: Weak<Scheduler>,
    /// W4 leg 3 (§4.3) TOLD drain-completion primitive. Reuses the
    /// [`ShutdownHandle`] `Condvar` shape (`shutdown.rs` reuse candidate (c)): a
    /// monotonic connection-removal generation guarded by [`Self::drain_removed`]'s
    /// mutex, bumped once whenever [`Self::remove`] actually drops a record — the
    /// single removal funnel every exit route (in-slice `mark_crashed`/`finish`,
    /// the reclaim reactor, and the reconciliation scan) reaches. The
    /// shutdown-sequence drain/settle waiter parks on the `Condvar` and wakes only
    /// on a delivered exit (a generation bump + `notify_all`) or the one admitted
    /// deadline it passes to `wait_timeout`. No periodic reap or count scan.
    pub(super) drain_generation: Mutex<u64>,
    /// Woken on every connection-record removal; the drain/settle waiter parks
    /// here. Paired with [`Self::drain_generation`] under the same mutex so an
    /// exit delivered between the waiter's arm-before-observe snapshot and its
    /// park cannot be lost (oracle 18).
    pub(super) drain_removed: Condvar,
    /// FIX A-ii shutdown flush barrier: a delivery-quiescence generation of the
    /// exact TOLD `drain_generation` shape. Bumped whenever a connection parks
    /// with every accepted publish already fanned out to its socket — but only
    /// while [`Self::settle_armed`] is set (the flush barrier is waiting) — so
    /// normal operation pays nothing. Guards [`Self::settle_changed`]'s mutex.
    pub(super) settle_generation: Mutex<u64>,
    /// Woken when a connection reaches delivery quiescence during shutdown; the
    /// flush barrier parks here, paired with [`Self::settle_generation`] for the
    /// same arm-before-observe safety as the drain waiter.
    pub(super) settle_changed: Condvar,
    /// Set only while the flush barrier is actively waiting, so a park bumps the
    /// settle generation and wakes the barrier ONLY when someone is listening.
    pub(super) settle_armed: AtomicBool,
    /// Server `Push` frames this runtime's connections have queued for their
    /// sockets, counted once per frame accepted into an outbound buffer. A push
    /// that fails, is cancelled or is rolled back is never counted.
    pub(super) pushes_queued: AtomicU64,
    /// Observes the real condvar predicate while its generation lock is held.
    #[cfg(test)]
    pub(super) settle_wait_observer: Mutex<Option<Sender<()>>>,
    /// Test-only count of drain waiter wakes that observed a real removal
    /// (generation advanced across the park). A quiet drain records zero — it
    /// wakes only for the single deadline (oracle 12).
    #[cfg(test)]
    pub(super) drain_exit_wakes: AtomicU64,
    /// Test-only count of drain waiter deadline expirations. A quiet drain that
    /// times out records exactly one — one arming, one delivery, no helper tick
    /// (oracles 12, 16).
    #[cfg(test)]
    pub(super) drain_deadline_hits: AtomicU64,
    /// Test-only one-use gate in the drain waiter's observe->park window, so a
    /// harness can deliver an exit strictly after the completion observation and
    /// before the park to pin the arm-before-observe barrier (oracles 18, 19).
    #[cfg(test)]
    pub(super) drain_park_barrier: Mutex<Option<PreWaitBarrier>>,
    /// R7 (§1.2(6)) test-only per-connection slice counter, keyed by pid. Bumped
    /// once at the head of every serviced slice. The park-flip's permanent rule-1
    /// assertion (a parked connection's counter must not advance without an event)
    /// reads this; the instrument lands now with a test proving it counts slices.
    #[cfg(test)]
    pub(super) slice_counts: Mutex<HashMap<u64, u64>>,
    /// One-use readiness markers for the next serviced slice of a process.
    #[cfg(test)]
    pub(super) slice_observers: Mutex<HashMap<u64, Sender<u64>>>,
    /// One-use readiness markers emitted only after the real final probe selects
    /// `Wait`, immediately before the native process returns to the scheduler.
    #[cfg(test)]
    pub(super) park_observers: Mutex<HashMap<u64, Sender<u64>>>,
    /// Most recent slice count whose real final probe selected `Wait`.
    #[cfg(test)]
    pub(super) park_counts: Mutex<HashMap<u64, u64>>,
    /// Where each report of frames written to a connection's transport goes,
    /// by pid (R3b F8, ruling 2a).
    #[cfg(test)]
    pub(super) frame_observers: Mutex<HashMap<u64, Sender<Vec<liminal::protocol::FrameType>>>>,
    /// Explicit capacities consumed in TCP process construction order.
    #[cfg(test)]
    pub(super) next_outbound_capacities: Mutex<VecDeque<usize>>,
    #[cfg(test)]
    pub(super) participant_holdback_pauses: Mutex<HashMap<u64, Sender<()>>>,
    /// Deterministic test gate placed after arm and before the final probe.
    #[cfg(test)]
    pub(super) pre_wait_barrier: Mutex<Option<PreWaitBarrier>>,
    /// Deterministic test gate in `enqueue_control`'s insert->wake window (S8).
    #[cfg(test)]
    pub(super) pre_wake_barrier: Mutex<Option<PreWaitBarrier>>,
    /// Pid-specific one-use gate held on the reclamation delivery path so a test
    /// can pin the dead-but-tracked S8 window deterministically (oracle 26).
    #[cfg(test)]
    pub(super) reclaim_barrier: Mutex<Option<ReclaimBarrier>>,
    /// Barrier-staged slices where the final probe found newly arrived work.
    #[cfg(test)]
    pub(super) pre_wait_probe_hits: AtomicU64,
    /// One-use observers for process-owned streams reaching their actual drop
    /// boundary after external scheduler termination.
    #[cfg(test)]
    pub(super) process_stream_drop_observers: Mutex<HashMap<RawFd, Sender<()>>>,
    /// One-shot reply slots for in-flight server pushes, keyed by correlation id.
    /// The supervisor registers a slot in `push_to_connection`; the connection
    /// process resolves it when the matching `PushReply` frame arrives. Each slot
    /// records the owning connection pid so the close path can drop a connection's
    /// outstanding slots and wake their awaiters with a prompt disconnected error.
    pub(super) push_replies: Mutex<HashMap<u64, PendingPush>>,
    /// Monotonic source of push correlation ids. Server-allocated, so it never
    /// collides with a client-chosen id on this connection.
    pub(super) next_push_id: AtomicU64,
    /// §5 `max_connections` admission counter. Incremented atomically (CAS
    /// against the limit) BEFORE a connection process is constructed and
    /// decremented on every spawn-failure path and on final record removal, so
    /// the signed bound holds under concurrent spawns — admission is never
    /// derived from the records-map length across an unlocked window.
    pub(super) admissions: AtomicU64,
    /// Optional application hook invoked on worker registration and on the close
    /// of a connection that had registered, and on the attach and close of a
    /// pass-stamped connection. `None` keeps liminal standalone: a
    /// `WorkerRegister` is accepted with no callback, a pass attaches silently.
    pub(super) notifier: Option<Arc<dyn ConnectionNotifier>>,
    /// Configured connection auth token (the `[auth]` section's token as opaque
    /// bytes). `Some` gates the `Connect` handshake — the frame's `auth_token` must
    /// match under a constant-time comparison; `None` leaves the server open-access,
    /// byte-identical to the pre-auth behaviour.
    pub(super) auth_token: Option<Vec<u8>>,
    pub(super) pass_verifier: Option<PassVerifier>,
    /// Operational caps (§5). Enforced with typed refusals at admission:
    /// per-connection subscription, conversation, push, and pending-reply counts,
    /// plus the shared inbox byte budget. Non-config constructors carry the signed
    /// defaults ([`LimitsConfig::default`]).
    pub(super) limits: LimitsConfig,
}

impl ConnectionRuntime {
    pub(super) fn new(
        installation: ConnectionRuntimeInstallation,
        control_atom: Atom,
        ready_atom: Atom,
        scheduler: Weak<Scheduler>,
        notifier: Option<Arc<dyn ConnectionNotifier>>,
        authentication: (Option<Vec<u8>>, Option<PassVerifier>),
        limits: LimitsConfig,
    ) -> Self {
        let (auth_token, pass_verifier) = authentication;
        let ConnectionRuntimeInstallation {
            services:
                ConnectionServiceInstallation {
                    services,
                    participant_service,
                },
            incarnations,
            fatal_shutdown,
        } = installation;
        Self {
            services,
            participant_service,
            incarnations,
            fatal_shutdown,
            signals: OnceLock::new(),
            link_life: crate::link::LinkRegistry::default(),
            lys_admitter: OnceLock::new(),
            service_bindings: OnceLock::new(),
            lifecycle: OnceLock::new(),
            membership_gate: OnceLock::new(),
            link_host: OnceLock::new(),
            revocation: OnceLock::new(),
            #[cfg(test)]
            close_event_fd_probe: Mutex::new(CloseEventFdProbe::Off),
            parked_connection_fates: Mutex::new(ParkedConnectionFates::default()),
            records: Mutex::new(Registry::default()),
            controls: Mutex::new(Vec::new()),
            control_atom,
            ready_atom,
            scheduler,
            drain_generation: Mutex::new(0),
            drain_removed: Condvar::new(),
            settle_generation: Mutex::new(0),
            settle_changed: Condvar::new(),
            settle_armed: AtomicBool::new(false),
            pushes_queued: AtomicU64::new(0),
            #[cfg(test)]
            settle_wait_observer: Mutex::new(None),
            #[cfg(test)]
            drain_exit_wakes: AtomicU64::new(0),
            #[cfg(test)]
            drain_deadline_hits: AtomicU64::new(0),
            #[cfg(test)]
            drain_park_barrier: Mutex::new(None),
            #[cfg(test)]
            slice_counts: Mutex::new(HashMap::new()),
            #[cfg(test)]
            slice_observers: Mutex::new(HashMap::new()),
            #[cfg(test)]
            park_observers: Mutex::new(HashMap::new()),
            #[cfg(test)]
            frame_observers: Mutex::new(HashMap::new()),
            #[cfg(test)]
            park_counts: Mutex::new(HashMap::new()),
            #[cfg(test)]
            next_outbound_capacities: Mutex::new(VecDeque::new()),
            #[cfg(test)]
            participant_holdback_pauses: Mutex::new(HashMap::new()),
            #[cfg(test)]
            pre_wait_barrier: Mutex::new(None),
            #[cfg(test)]
            pre_wake_barrier: Mutex::new(None),
            #[cfg(test)]
            reclaim_barrier: Mutex::new(None),
            #[cfg(test)]
            pre_wait_probe_hits: AtomicU64::new(0),
            #[cfg(test)]
            process_stream_drop_observers: Mutex::new(HashMap::new()),
            push_replies: Mutex::new(HashMap::new()),
            next_push_id: AtomicU64::new(1),
            admissions: AtomicU64::new(0),
            notifier,
            auth_token,
            pass_verifier,
            limits,
        }
    }
}