liminal-server 0.23.1

#[cfg(test)]
use super::Sender;
use super::{
    AdmissionReadiness, AdmissionReservation, Arc, AtomTable, CONNECTION_READY_ATOM,
    CONNECTION_SCHEDULER_THREADS, CONNECTION_SHUTDOWN_CONTROL_ATOM, CandidateLaneListener,
    ConnectionControl, ConnectionHandle, ConnectionIncarnation, ConnectionIncarnationAuthority,
    ConnectionNotifier, ConnectionProcess, ConnectionRuntime, ConnectionRuntimeInstallation,
    ConnectionServiceInstallation, ConnectionServices, ExitReason, LimitsConfig, LysInstall,
    ModuleRegistry, MountKind, Mutex, NativeBifs, NativeHandlerFactory, NativeRetirements,
    PassVerifier, Scheduler, SchedulerConfig, SchedulerServices, ServerError, ShutdownHandle,
    SocketAddr, TcpStream, Weak, install_lys, lifecycle, reclaim,
};

pub(super) struct SupervisorInner {
    pub(super) scheduler: Arc<Scheduler>,
    pub(super) runtime: Arc<ConnectionRuntime>,
    pub(super) reclaim: reclaim::ReclaimReactor,
    pub(super) retirements: Arc<NativeRetirements>,
    pub(super) background: Mutex<lifecycle::BackgroundStart>,
    pub(super) incarnations: Option<Arc<ConnectionIncarnationAuthority>>,
    /// P0 #56 R4: the readiness probe's view of whether this server can admit.
    /// Shared with the incarnation authority when one is installed.
    pub(super) admission_readiness: AdmissionReadiness,
    #[cfg(test)]
    pub(super) reclaim_stop_observer: Mutex<Option<Sender<bool>>>,
}

impl std::fmt::Debug for SupervisorInner {
    fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        formatter
            .debug_struct("SupervisorInner")
            .field("runtime", &self.runtime)
            .finish_non_exhaustive()
    }
}
impl SupervisorInner {
    pub(super) fn new(
        services: Arc<dyn ConnectionServices>,
        notifier: Option<Arc<dyn ConnectionNotifier>>,
        auth_token: Option<Vec<u8>>,
        pass_verifier: Option<PassVerifier>,
        lys: Option<LysInstall>,
        limits: LimitsConfig,
        fatal_shutdown: Option<ShutdownHandle>,
    ) -> Result<Self, ServerError> {
        let installed_services = ConnectionServiceInstallation::capture(services);
        // P0 #56 R4. Owned by the supervisor so the readiness probe has a stable
        // handle whether or not a participant service (and therefore an
        // incarnation authority) is installed: a server with no authority has
        // nothing that can hold admission, and reports available.
        let admission_readiness = AdmissionReadiness::available();
        let incarnations = installed_services
            .participant_service
            .as_ref()
            .map(
                |service| -> Result<Arc<ConnectionIncarnationAuthority>, ServerError> {
                    ConnectionIncarnationAuthority::startup(
                        service.durable_store(),
                        limits.max_connections,
                        service.publication_conversation_limit(),
                        service,
                        admission_readiness.clone(),
                    )
                    .map(Arc::new)
                },
            )
            .transpose()?;
        let atoms = AtomTable::with_common_atoms();
        let control_atom = atoms.intern(CONNECTION_SHUTDOWN_CONTROL_ATOM);
        let registry = Arc::new(ModuleRegistry::new());

        let scheduler = Scheduler::with_services(
            SchedulerConfig {
                thread_count: Some(CONNECTION_SCHEDULER_THREADS),
                ..SchedulerConfig::default()
            },
            SchedulerServices::from_config().owned_readiness(),
            registry,
            // The strongest of the three cases: `registry` here is a bare
            // `ModuleRegistry::new()` with NO module ever inserted, so this
            // scheduler cannot execute a single bytecode instruction, let alone
            // a guard BIF. Connections run as native processes via
            // `NativeHandlerFactory`. beamr 0.19 requires the answer be written
            // down rather than inherited; the answer is none.
            NativeBifs::none(),
        )
        .map_err(|message| ServerError::ListenerAccept {
            message: format!("failed to start connection scheduler: {message}"),
        })?;
        let ready_atom = atoms.intern(CONNECTION_READY_ATOM);
        let scheduler = Arc::new(scheduler);
        // The runtime captures a WEAK handle to the connection scheduler so
        // notifier wakes (R3/R1(vi)) can be fired from another actor's slice
        // without a strong scheduler↔process↔runtime cycle that would leak the
        // whole connection scheduler.
        let runtime = Arc::new(ConnectionRuntime::new(
            ConnectionRuntimeInstallation {
                services: installed_services,
                incarnations: incarnations.clone(),
                fatal_shutdown,
            },
            control_atom,
            ready_atom,
            Arc::downgrade(&scheduler),
            notifier,
            (auth_token, pass_verifier),
            limits,
        ));
        if let Some(lys) = lys {
            install_lys(&runtime, lys)?;
        }
        // A connection fate parked on candidate-lane `Precedence` is retried by
        // the drain that clears it, so the runtime is told about every drain.
        if let (Some(service), Some(notifier)) = (runtime.participant_service(), runtime.notifier())
        {
            service.install_checkpoint_crossing_observer(Arc::clone(notifier));
        }
        if let Some(service) = runtime.participant_service() {
            let runtime_weak: Weak<ConnectionRuntime> = Arc::downgrade(&runtime);
            let listener: Weak<dyn CandidateLaneListener> = runtime_weak;
            service
                .install_candidate_lane_listener(listener)
                .map_err(|error| ServerError::ParticipantIncarnation {
                    phase: "candidate-lane listener installation",
                    message: error.to_string(),
                })?;
        }
        let reclaim = reclaim::ReclaimReactor::start(&scheduler, &runtime)?;
        Ok(Self {
            scheduler,
            runtime,
            reclaim,
            retirements: Arc::new(NativeRetirements::default()),
            background: Mutex::new(lifecycle::BackgroundStart::default()),
            incarnations,
            admission_readiness,
            #[cfg(test)]
            reclaim_stop_observer: Mutex::new(None),
        })
    }

    pub(super) fn spawn_connection(
        self: &Arc<Self>,
        stream: TcpStream,
    ) -> Result<ConnectionHandle, ServerError> {
        // §5 `max_connections`: ATOMIC admission reservation acquired BEFORE any
        // process construction (review round 1 item 7 — a signed bound must not
        // be exceedable by concurrent callers; check-then-spawn across an
        // unlocked window was). The CAS reservation is released on every failure
        // path below and converts into the connection record at `register`;
        // thereafter the single record-removal path (`remove`) releases it. An
        // over-cap accept therefore costs nothing and the bound holds under any
        // concurrency.
        self.runtime.try_reserve_admission()?;
        let reservation = AdmissionReservation {
            runtime: &self.runtime,
            armed: true,
        };
        stream
            .set_nonblocking(true)
            .map_err(|error| ServerError::ListenerAccept {
                message: format!("failed to configure connection stream: {error}"),
            })?;
        // A failed `peer_addr` is folded to `None` rather than refusing the
        // accept: an unidentifiable peer is still a serviceable connection. The
        // cost is paid in triage, not in service — every later log line for this
        // connection reports `peer_addr=None`, so nothing that happens on it can
        // be tied back to a remote endpoint.
        let peer_addr = stream.peer_addr().ok();
        // The host-held duplicate keeps the fd alive until the single record-removal
        // path has synchronously deregistered readiness. External process death can
        // therefore never let fd reuse overtake host-side deregistration.
        let fd_guard = stream
            .try_clone()
            .map_err(|error| ServerError::ListenerAccept {
                message: format!("failed to retain connection fd for teardown: {error}"),
            })?;
        let connection_incarnation = self.allocate_connection_incarnation()?;
        let holder = Arc::new(Mutex::new(Some(stream)));
        let runtime = Arc::clone(&self.runtime);
        let process_holder = Arc::clone(&holder);
        let factory: NativeHandlerFactory = Box::new(move || {
            Box::new(ConnectionProcess::from_holder(
                Arc::clone(&runtime),
                peer_addr,
                &process_holder,
                connection_incarnation,
            ))
        });
        let pid = self.spawn_native_connection(factory, "failed to spawn connection process")?;
        if let Err(error) = self.runtime.register_connection(
            pid,
            peer_addr,
            connection_incarnation,
            MountKind::Tcp,
            Some(fd_guard),
        ) {
            // Registration failure leaves no host record to reap. Terminate the
            // just-spawned process explicitly so neither its stream nor admission
            // reservation can escape this failed spawn.
            self.scheduler.terminate_process(pid, ExitReason::Error);
            return Err(error);
        }
        // The reservation is now owned by the registered record: `remove` (the
        // single record-removal path — finish/mark_crashed/reap all funnel
        // through it) releases the admission when the record goes away.
        reservation.convert();
        Ok(ConnectionHandle {
            pid,
            peer_addr,
            connection_incarnation,
            supervisor: Arc::clone(self),
        })
    }

    /// LP-WS-TRANSPORT R1.3: the sibling-transport spawn body. Mirrors
    /// [`Self::spawn_connection`]'s admission → incarnation → spawn → register →
    /// convert sequence exactly (same reservation guard, same failure rollback,
    /// same single record-removal ownership), differing only in that the caller
    /// supplies the native handler factory and the host-held fd guard instead of
    /// a raw `TcpStream`. Purely additive; the TCP path never calls this.
    pub(super) fn spawn_transport_connection(
        self: &Arc<Self>,
        peer_addr: Option<SocketAddr>,
        fd_guard: Option<TcpStream>,
        mount: MountKind,
        build_factory: &dyn Fn(
            Arc<ConnectionRuntime>,
            Option<ConnectionIncarnation>,
        ) -> NativeHandlerFactory,
    ) -> Result<ConnectionHandle, ServerError> {
        self.runtime.try_reserve_admission()?;
        let reservation = AdmissionReservation {
            runtime: &self.runtime,
            armed: true,
        };
        let connection_incarnation = self.allocate_connection_incarnation()?;
        let factory = build_factory(Arc::clone(&self.runtime), connection_incarnation);
        let pid = self.spawn_native_connection(factory, "failed to spawn connection process")?;
        if let Err(error) = self.runtime.register_connection(
            pid,
            peer_addr,
            connection_incarnation,
            mount,
            fd_guard,
        ) {
            self.scheduler.terminate_process(pid, ExitReason::Error);
            return Err(error);
        }
        reservation.convert();
        Ok(ConnectionHandle {
            pid,
            peer_addr,
            connection_incarnation,
            supervisor: Arc::clone(self),
        })
    }

    pub(super) fn allocate_connection_incarnation(
        &self,
    ) -> Result<Option<ConnectionIncarnation>, ServerError> {
        let Some(authority) = self.incarnations.as_ref() else {
            return Ok(None);
        };
        // Production-era uniqueness invariant: every published incarnation is
        // unique against ALL durable references — binding epochs committed
        // into conversation logs included — by allocator-log monotonicity
        // alone, not by the completeness of the reference set below.
        //
        //   1. Startup replays the durable allocator stream and STRICTLY
        //      increments the server incarnation, fsyncing the Startup event
        //      before any listener becomes ready
        //      (`IncarnationStream::startup`); a server value is never wrapped
        //      or reused, so no two process lifetimes share one.
        //   2. Within a lifetime, allocations are serialized under this
        //      authority's mutex, candidates start strictly above the durable
        //      `last_examined_connection_ordinal`
        //      (`allocate_connection_incarnation`), and every allocation's
        //      event is appended and flushed BEFORE its pair is published
        //      (`StartedIncarnationStream::allocate`), so ordinals never
        //      repeat within a lifetime and replay restores a head at or
        //      above every published ordinal.
        //   3. A durable reference can only name a pair this allocator
        //      previously PUBLISHED (binding epochs are committed only after
        //      their connection was admitted), and the same store's flush
        //      barrier orders the allocator event before any conversation-log
        //      entry that references it.
        //
        // The live-connection reference set below is therefore defense in
        // depth — a bounded collision skip against a rolled-back or divergent
        // allocator stream — never the uniqueness foundation, and never a raw
        // caller-supplied matrix.
        let references = self.runtime.complete_active_incarnation_references()?;
        authority.allocate(&references).map(Some)
    }

    pub(super) fn broadcast_control(&self, control: &ConnectionControl) {
        for connection in self.runtime.active_connections() {
            if !self.enqueue_control(connection.pid, control.clone()) {
                // Observation, not diagnosis: the routine cause is a process
                // that is no longer live, but a refused control queue reaches
                // here too and has already warned with its own error.
                tracing::debug!(
                    connection_pid = connection.pid,
                    peer_addr = ?connection.peer_addr,
                    ?control,
                    "connection control message was not published to the process"
                );
            }
        }
    }

    /// Queues `control` for `pid` and wakes the process. Returns whether the
    /// control was PUBLISHED (left in the queue with a successful wake, or
    /// already consumed by a drain) — `false` guarantees no consumer ever saw
    /// it.
    ///
    /// S8: a failed wake does NOT prove the queued control was never consumed.
    /// The insert releases the queue lock before the wake attempt, and a
    /// process already executing a control drain (each control atom drains ALL
    /// queued controls for the pid) can pop the just-inserted entry in that
    /// window, then exit before the wake check. Publication is therefore
    /// disambiguated BY OBSERVATION on the failed-wake path: `remove_control`
    /// finding and removing the entry proves no consumer saw it (truly
    /// unpublished — `false`); finding nothing proves a drain consumed it
    /// (`pop_control` is the only other remover of queue entries, and the
    /// removal key embeds the push's runtime-unique correlation id, so it can
    /// never match a different entry) — the control was published and the
    /// caller's slot lifecycle carries the delivery truth (`true`).
    pub(super) fn enqueue_control(&self, pid: u64, control: ConnectionControl) -> bool {
        self.runtime.publish_control(&self.scheduler, pid, control)
    }
}