The switch now reaches the running server
The join: the policy decides, features owns the listener objects, ListenerControl owns the running sockets, and only Server has both. Server::set_protocol_policy is what a click performs. It applies the locks to what was asked before storing anything (LP-21), so what is recorded is what the server allows. Closing removes each listener object and then stops its socket; opening puts the object back and then spawns it. The order is the point in both directions -- a socket stopped while its object remains returns on the next restart, and a socket spawned before its object exists has nothing to come back to. saved_listeners is carried over from the stored policy rather than taken from the request. A client never sets it, and a /set that omitted it would otherwise lose the listeners still waiting to come back. Putting a listener back has to bind a fresh socket, so it re-parses from the registry -- the objects are already back by then -- rather than trying to revive the saved one. Only main knows which session manager a protocol wants, so it leaves a spawner behind at startup and spawn_listener is now shared between that and the initial spawn. Without a spawner a restored listener is reported as pending a restart rather than promised, which is what the test servers will see. A listener that cannot be put back does not stop the others and stays saved for another try (LP-5). Still nothing an operator can reach: no JMAP method calls this yet, and no sign-in is refused. What it does do is close and reopen a port on a running server, which is the part that did not exist this morning.
This commit is contained in:
@@ -23,11 +23,18 @@
|
||||
//! process stops answering; anything that still routes the port is the
|
||||
//! operator's to reconcile, and is deliberately left alone.
|
||||
|
||||
use crate::config::server::ServerProtocol;
|
||||
use crate::config::server::{Listener, ServerProtocol};
|
||||
use crate::network::TcpAcceptor;
|
||||
use ahash::AHashMap;
|
||||
use parking_lot::RwLock;
|
||||
use std::sync::OnceLock;
|
||||
use tokio::sync::watch;
|
||||
|
||||
/// How a listener is spawned. Only `main` knows how to build the session
|
||||
/// manager for a protocol, so it leaves this behind at startup and the policy
|
||||
/// uses it to put a listener back without a restart (LP-5).
|
||||
pub type SpawnListener = Box<dyn Fn(Listener, TcpAcceptor, watch::Receiver<bool>) + Send + Sync>;
|
||||
|
||||
/// A listener that is currently accepting, and the switch that stops it.
|
||||
struct Running {
|
||||
protocol: ServerProtocol,
|
||||
@@ -47,6 +54,7 @@ pub struct ListenerInfo {
|
||||
#[derive(Default)]
|
||||
pub struct ListenerControl {
|
||||
running: RwLock<AHashMap<String, Running>>,
|
||||
spawner: OnceLock<SpawnListener>,
|
||||
}
|
||||
|
||||
impl ListenerControl {
|
||||
@@ -70,6 +78,33 @@ impl ListenerControl {
|
||||
shutdown_rx
|
||||
}
|
||||
|
||||
/// Remembers how to spawn a listener, once, at startup. Later calls are
|
||||
/// ignored, so nothing can swap the spawner out from under a running
|
||||
/// server.
|
||||
pub fn set_spawner(&self, spawner: SpawnListener) {
|
||||
let _ = self.spawner.set(spawner);
|
||||
}
|
||||
|
||||
/// Whether a spawner has been left behind. Without one, a listener can be
|
||||
/// stopped but not started, and the caller has to say so rather than
|
||||
/// promise a port that will not open until a restart.
|
||||
pub fn can_spawn(&self) -> bool {
|
||||
self.spawner.get().is_some()
|
||||
}
|
||||
|
||||
/// Starts a listener and registers it, so it can be stopped again.
|
||||
/// Returns false when no spawner was left behind.
|
||||
pub fn spawn(&self, listener: Listener, acceptor: TcpAcceptor) -> bool {
|
||||
let Some(spawner) = self.spawner.get() else {
|
||||
return false;
|
||||
};
|
||||
|
||||
let ports = listener.listeners.iter().map(|l| l.addr.port()).collect();
|
||||
let shutdown_rx = self.register(listener.id.clone(), listener.protocol, ports);
|
||||
spawner(listener, acceptor, shutdown_rx);
|
||||
true
|
||||
}
|
||||
|
||||
/// Stops one listener by id. Returns what was stopped, or `None` when no
|
||||
/// listener of that id is running.
|
||||
pub fn stop(&self, id: &str) -> Option<ListenerInfo> {
|
||||
|
||||
@@ -0,0 +1,188 @@
|
||||
/*
|
||||
* SPDX-FileCopyrightText: 2026 Coffey Labs
|
||||
*
|
||||
* SPDX-License-Identifier: AGPL-3.0-only
|
||||
*/
|
||||
|
||||
//! Turning the legacy-protocols switch, and making it true of the running
|
||||
//! server (legacy-protocols spec, LP-1, LP-2 and LP-5).
|
||||
//!
|
||||
//! Two halves meet here. `inbuxa_features::security` decides what the policy
|
||||
//! means and owns the listener **objects**; [`ListenerControl`] owns the
|
||||
//! running **sockets**. Neither can do the job alone, and only `Server` has
|
||||
//! both, so the join lives here.
|
||||
//!
|
||||
//! Order matters in both directions. Closing removes the object first and then
|
||||
//! stops the socket: a socket stopped before its object is gone would come
|
||||
//! back on the next restart. Opening puts the object back first and then
|
||||
//! spawns, for the same reason in reverse.
|
||||
//!
|
||||
//! Nothing here touches the host's firewall, NAT port-forwards or any proxy
|
||||
//! (LP-20). The server stops answering; what still routes the port is the
|
||||
//! operator's to reconcile.
|
||||
|
||||
use crate::{Server, config::server::Listeners, network::TcpAcceptor};
|
||||
use inbuxa_features::security::{
|
||||
listeners,
|
||||
protocol_policy::{self, ProtocolPolicy, SavedListener},
|
||||
};
|
||||
use store::registry::bootstrap::Bootstrap;
|
||||
|
||||
/// What turning the switch actually did.
|
||||
#[derive(Debug, Default)]
|
||||
pub struct PolicyChange {
|
||||
/// Listeners removed and stopped (LP-1).
|
||||
pub closed: Vec<SavedListener>,
|
||||
/// Listeners put back and started again (LP-5).
|
||||
pub reopened: Vec<SavedListener>,
|
||||
/// Listeners that could not be put back, with the reason. Each stays
|
||||
/// saved for another try (LP-5).
|
||||
pub failed: Vec<(SavedListener, String)>,
|
||||
/// Properties the locks overruled (LP-21).
|
||||
pub overruled: Vec<&'static str>,
|
||||
/// Listeners whose object is right but whose socket needs a restart,
|
||||
/// because no spawner was left behind. Empty on a normally booted server.
|
||||
pub pending_restart: Vec<String>,
|
||||
}
|
||||
|
||||
impl PolicyChange {
|
||||
/// Whether anything at all happened, for the caller deciding to emit
|
||||
/// `security.legacy-protocols-changed` (LP-8).
|
||||
pub fn is_empty(&self) -> bool {
|
||||
self.closed.is_empty()
|
||||
&& self.reopened.is_empty()
|
||||
&& self.failed.is_empty()
|
||||
&& self.overruled.is_empty()
|
||||
}
|
||||
}
|
||||
|
||||
impl Server {
|
||||
/// The policy in force.
|
||||
pub async fn protocol_policy(&self) -> trc::Result<ProtocolPolicy> {
|
||||
protocol_policy::get(&self.core.storage.data).await
|
||||
}
|
||||
|
||||
/// Turns the switch, and makes it true of the running server.
|
||||
///
|
||||
/// `requested` is what the client asked for; the locks are applied to it
|
||||
/// first (LP-21), so what gets stored is what the server allows, not what
|
||||
/// was asked. Returns what actually happened, for the response and the
|
||||
/// event.
|
||||
pub async fn set_protocol_policy(
|
||||
&self,
|
||||
requested: ProtocolPolicy,
|
||||
changed_by: Option<String>,
|
||||
) -> trc::Result<PolicyChange> {
|
||||
let mut policy = requested;
|
||||
let mut change = PolicyChange {
|
||||
overruled: policy.apply_locks(),
|
||||
..Default::default()
|
||||
};
|
||||
|
||||
// Carry forward what earlier changes saved: the client never sets
|
||||
// this, and a /set that omitted it must not lose the listeners still
|
||||
// waiting to come back.
|
||||
let previous = self.protocol_policy().await?;
|
||||
policy.saved_listeners = previous.saved_listeners;
|
||||
policy.changed_at = Some(store::write::now() * 1000);
|
||||
policy.changed_by = changed_by;
|
||||
|
||||
if policy.legacy_protocols.is_disabled() {
|
||||
self.close_legacy_listeners(&mut policy, &mut change).await?;
|
||||
} else {
|
||||
self.reopen_legacy_listeners(&mut policy, &mut change)
|
||||
.await?;
|
||||
}
|
||||
|
||||
protocol_policy::set(&self.core.storage.data, &policy).await?;
|
||||
|
||||
Ok(change)
|
||||
}
|
||||
|
||||
/// Removes the listener objects the policy closes, then stops their
|
||||
/// sockets (LP-1, LP-2).
|
||||
async fn close_legacy_listeners(
|
||||
&self,
|
||||
policy: &mut ProtocolPolicy,
|
||||
change: &mut PolicyChange,
|
||||
) -> trc::Result<()> {
|
||||
let removed = listeners::close(self.registry(), policy).await?;
|
||||
|
||||
for saved in &removed {
|
||||
// The runtime registry is keyed by the listener's name, which is
|
||||
// what `close` returns as the saved listener's id.
|
||||
self.inner.data.listener_control.stop(&saved.id);
|
||||
}
|
||||
|
||||
policy.saved_listeners.extend(removed.iter().cloned());
|
||||
change.closed = removed;
|
||||
|
||||
Ok(())
|
||||
}
|
||||
|
||||
/// Puts back every saved listener and starts it again (LP-5).
|
||||
async fn reopen_legacy_listeners(
|
||||
&self,
|
||||
policy: &mut ProtocolPolicy,
|
||||
change: &mut PolicyChange,
|
||||
) -> trc::Result<()> {
|
||||
if policy.saved_listeners.is_empty() {
|
||||
return Ok(());
|
||||
}
|
||||
|
||||
let saved = std::mem::take(&mut policy.saved_listeners);
|
||||
let (restored, failed) = listeners::reopen(self.registry(), &saved).await?;
|
||||
|
||||
// A listener that could not be put back stays saved for another try.
|
||||
policy.saved_listeners = failed.iter().map(|(listener, _)| listener.clone()).collect();
|
||||
change.failed = failed;
|
||||
|
||||
if !restored.is_empty() {
|
||||
change.pending_restart = self.spawn_restored_listeners(&restored).await?;
|
||||
}
|
||||
change.reopened = restored;
|
||||
|
||||
Ok(())
|
||||
}
|
||||
|
||||
/// Binds and spawns the listeners just put back, so a port opens without a
|
||||
/// restart. Returns the names that still need one.
|
||||
async fn spawn_restored_listeners(&self, restored: &[SavedListener]) -> trc::Result<Vec<String>> {
|
||||
let control = &self.inner.data.listener_control;
|
||||
if !control.can_spawn() {
|
||||
return Ok(restored.iter().map(|listener| listener.id.clone()).collect());
|
||||
}
|
||||
|
||||
// Re-parse from the registry rather than from the saved object: the
|
||||
// socket has to be created and bound afresh, and the parser is what
|
||||
// knows how. The objects are already back, so this sees them.
|
||||
let mut bootstrap = Bootstrap::new(self.registry().clone()).await;
|
||||
let mut parsed = Listeners::parse(&mut bootstrap).await;
|
||||
parsed
|
||||
.parse_tcp_acceptors(&mut bootstrap, self.inner.clone())
|
||||
.await;
|
||||
|
||||
let wanted: Vec<&str> = restored.iter().map(|l| l.id.as_str()).collect();
|
||||
let mut spawned = Vec::new();
|
||||
|
||||
let mut acceptors = std::mem::take(&mut parsed.tcp_acceptors);
|
||||
for listener in parsed.servers {
|
||||
if !wanted.contains(&listener.id.as_str()) || control.is_running(&listener.id) {
|
||||
continue;
|
||||
}
|
||||
let acceptor = acceptors
|
||||
.remove(&listener.id)
|
||||
.unwrap_or(TcpAcceptor::Plain);
|
||||
let id = listener.id.clone();
|
||||
if control.spawn(listener, acceptor) {
|
||||
spawned.push(id);
|
||||
}
|
||||
}
|
||||
|
||||
Ok(restored
|
||||
.iter()
|
||||
.map(|listener| listener.id.clone())
|
||||
.filter(|id| !spawned.contains(id))
|
||||
.collect())
|
||||
}
|
||||
}
|
||||
@@ -36,6 +36,7 @@ pub mod autoconfig;
|
||||
pub mod control;
|
||||
pub mod dkim;
|
||||
pub mod dns;
|
||||
pub mod legacy;
|
||||
pub mod limiter;
|
||||
pub mod listen;
|
||||
pub mod mta;
|
||||
|
||||
+65
-33
@@ -9,14 +9,21 @@
|
||||
#![warn(clippy::cast_possible_wrap)]
|
||||
#![warn(clippy::cast_sign_loss)]
|
||||
|
||||
use common::{BuildServer, config::server::ServerProtocol, manager::boot::BootManager};
|
||||
use common::{
|
||||
BuildServer, Inner,
|
||||
config::server::{Listener, ServerProtocol},
|
||||
manager::boot::BootManager,
|
||||
network::TcpAcceptor,
|
||||
};
|
||||
use http::HttpSessionManager;
|
||||
use imap::core::ImapSessionManager;
|
||||
use managesieve::core::ManageSieveSessionManager;
|
||||
use pop3::Pop3SessionManager;
|
||||
use services::{StartServices, broadcast::subscriber::spawn_broadcast_subscriber};
|
||||
use smtp::{StartQueueManager, core::SmtpSessionManager};
|
||||
use std::sync::Arc;
|
||||
use std::time::Duration;
|
||||
use tokio::sync::watch;
|
||||
use trc::Collector;
|
||||
use utils::wait_for_shutdown;
|
||||
|
||||
@@ -75,43 +82,24 @@ async fn main() -> std::io::Result<()> {
|
||||
// rest accepting (legacy-protocols LP-2). The registry lives in `Data` and
|
||||
// so outlives the listeners, which it must: it owns the sending ends.
|
||||
let listener_control = &init.inner.data.listener_control;
|
||||
let spawn_inner = init.inner.clone();
|
||||
let (shutdown_tx, shutdown_rx) =
|
||||
init.servers
|
||||
.spawn_with_control(listener_control, |server, acceptor, shutdown_rx| {
|
||||
match &server.protocol {
|
||||
ServerProtocol::Smtp | ServerProtocol::Lmtp => server.spawn(
|
||||
SmtpSessionManager::new(init.inner.clone()),
|
||||
init.inner.clone(),
|
||||
acceptor,
|
||||
shutdown_rx,
|
||||
),
|
||||
ServerProtocol::Http => server.spawn(
|
||||
HttpSessionManager::new(init.inner.clone()),
|
||||
init.inner.clone(),
|
||||
acceptor,
|
||||
shutdown_rx,
|
||||
),
|
||||
ServerProtocol::Imap => server.spawn(
|
||||
ImapSessionManager::new(init.inner.clone()),
|
||||
init.inner.clone(),
|
||||
acceptor,
|
||||
shutdown_rx,
|
||||
),
|
||||
ServerProtocol::Pop3 => server.spawn(
|
||||
Pop3SessionManager::new(init.inner.clone()),
|
||||
init.inner.clone(),
|
||||
acceptor,
|
||||
shutdown_rx,
|
||||
),
|
||||
ServerProtocol::ManageSieve => server.spawn(
|
||||
ManageSieveSessionManager::new(init.inner.clone()),
|
||||
init.inner.clone(),
|
||||
acceptor,
|
||||
shutdown_rx,
|
||||
),
|
||||
};
|
||||
spawn_listener(&spawn_inner, server, acceptor, shutdown_rx);
|
||||
});
|
||||
|
||||
// Leave behind how to spawn a listener, so putting one back opens its port
|
||||
// without a restart (LP-5). Only this file knows the session manager for a
|
||||
// protocol, so only this file can say.
|
||||
let spawn_inner = init.inner.clone();
|
||||
init.inner
|
||||
.data
|
||||
.listener_control
|
||||
.set_spawner(Box::new(move |server, acceptor, shutdown_rx| {
|
||||
spawn_listener(&spawn_inner, server, acceptor, shutdown_rx);
|
||||
}));
|
||||
|
||||
// Start broadcast subscriber
|
||||
let inner = init.inner.clone();
|
||||
spawn_broadcast_subscriber(init.inner, shutdown_rx);
|
||||
@@ -132,3 +120,47 @@ async fn main() -> std::io::Result<()> {
|
||||
|
||||
Ok(())
|
||||
}
|
||||
|
||||
/// Starts one listener under the session manager its protocol calls for.
|
||||
///
|
||||
/// Used twice: once for every listener at startup, and again whenever the
|
||||
/// legacy-protocols switch puts a listener back (LP-5).
|
||||
fn spawn_listener(
|
||||
inner: &Arc<Inner>,
|
||||
server: Listener,
|
||||
acceptor: TcpAcceptor,
|
||||
shutdown_rx: watch::Receiver<bool>,
|
||||
) {
|
||||
match &server.protocol {
|
||||
ServerProtocol::Smtp | ServerProtocol::Lmtp => server.spawn(
|
||||
SmtpSessionManager::new(inner.clone()),
|
||||
inner.clone(),
|
||||
acceptor,
|
||||
shutdown_rx,
|
||||
),
|
||||
ServerProtocol::Http => server.spawn(
|
||||
HttpSessionManager::new(inner.clone()),
|
||||
inner.clone(),
|
||||
acceptor,
|
||||
shutdown_rx,
|
||||
),
|
||||
ServerProtocol::Imap => server.spawn(
|
||||
ImapSessionManager::new(inner.clone()),
|
||||
inner.clone(),
|
||||
acceptor,
|
||||
shutdown_rx,
|
||||
),
|
||||
ServerProtocol::Pop3 => server.spawn(
|
||||
Pop3SessionManager::new(inner.clone()),
|
||||
inner.clone(),
|
||||
acceptor,
|
||||
shutdown_rx,
|
||||
),
|
||||
ServerProtocol::ManageSieve => server.spawn(
|
||||
ManageSieveSessionManager::new(inner.clone()),
|
||||
inner.clone(),
|
||||
acceptor,
|
||||
shutdown_rx,
|
||||
),
|
||||
}
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user