use tokio::net::TcpListener;
use super::ShutdownOutcome;
use super::transports::{shutdown_signal, transport_bind, transport_result};
use crate::error::ServerError;
use crate::shutdown;
use crate::state::ServerState;
pub(super) async fn serve_until_shutdown(
state: &ServerState,
shutdown_tx: &tokio::sync::watch::Sender<bool>,
grpc: &mut tokio::task::JoinHandle<Result<(), ServerError>>,
http: &mut tokio::task::JoinHandle<Result<(), ServerError>>,
) -> Result<shutdown::ShutdownReport, ServerError> {
Ok(tokio::select! {
result = &mut *grpc => {
transport_result("gRPC", result)?;
state.shutdown()?;
transport_exit_report(state)
},
result = &mut *http => {
transport_result("HTTP", result)?;
state.shutdown()?;
transport_exit_report(state)
},
result = shutdown_signal() => {
result?;
let _receiver_count = shutdown_tx.send(true);
let report = shutdown::drain_after_first_signal(state.clone(), async {
let _ = shutdown_signal().await;
}).await?;
if !matches!(report.outcome, ShutdownOutcome::Forced) {
transport_result("gRPC", grpc.await)?;
transport_result("HTTP", http.await)?;
}
report
},
})
}
pub(super) struct OpenedDoors {
pub(super) grpc_listener: TcpListener,
pub(super) http_listener: TcpListener,
pub(super) bound_grpc: std::net::SocketAddr,
pub(super) bound_http: std::net::SocketAddr,
pub(super) identity_pid: u32,
pub(super) pid_file_guard: crate::control::pid_file::PidFileGuard,
}
pub(super) async fn bind_and_claim(
home: &std::path::Path,
grpc_address: std::net::SocketAddr,
http_address: std::net::SocketAddr,
commit: &str,
drain_timeout: std::time::Duration,
) -> Result<OpenedDoors, ServerError> {
let grpc_listener = TcpListener::bind(grpc_address)
.await
.map_err(|source| transport_bind("grpc", grpc_address, source))?;
let http_listener = TcpListener::bind(http_address)
.await
.map_err(|source| transport_bind("http", http_address, source))?;
let bound_grpc = grpc_listener
.local_addr()
.map_err(|source| transport_bind("grpc", grpc_address, source))?;
let bound_http = http_listener
.local_addr()
.map_err(|source| transport_bind("http", http_address, source))?;
let identity = crate::control::incarnation::self_identity()?;
let pid_record = crate::control::pid_file::PidRecord {
pid: identity.pid,
started_at_unix_secs: identity.started_at_unix_secs,
binary_sha256: identity.binary_sha256,
version: env!("CARGO_PKG_VERSION").to_owned(),
commit: commit.to_owned(),
http_address: bound_http,
grpc_address: bound_grpc,
drain_timeout_seconds: drain_timeout.as_secs(),
};
let pid_file_guard = crate::control::pid_file::claim(home, &pid_record)?;
match crate::control::incarnation::probe(&pid_record) {
crate::control::incarnation::IncarnationProbe::Verified { .. } => {}
other => {
tracing::error!(
pid = pid_record.pid,
probe = ?other,
"the pid record this server just wrote does not verify against its \
own live process: the incarnation instrument is unstable on this \
host, and `aion server stop`/`status` will refuse to trust the \
record. The server runs; the control verbs will not address it"
);
}
}
tracing::info!(
pid = identity.pid,
home_claimed = pid_file_guard.holds_claim(),
"home claim resolved"
);
Ok(OpenedDoors {
grpc_listener,
http_listener,
bound_grpc,
bound_http,
identity_pid: identity.pid,
pid_file_guard,
})
}
fn transport_exit_report(state: &ServerState) -> shutdown::ShutdownReport {
shutdown::ShutdownReport {
outcome: ShutdownOutcome::Clean,
drain_timeout: state.runtime_config().drain_timeout,
delivered_drain_requests: 0,
parked: Vec::new(),
parked_declared: Vec::new(),
managed_workers_stopped: Vec::new(),
managed_workers_unstopped: Vec::new(),
}
}