tensamin/iota
Watch
6
0
Fork
You've already forked iota
 0
Code Issues Pull requests 3 Projects Releases 1 Packages Wiki Activity Actions

[WIP] Daemon & CLI

Browse source
This commit is contained in:
Alex-Emmet 2026-07-23 23:13:02 +02:00
commit 8b158108bb
100 changed files with 6519 additions and 1596 deletions
Download patch file Download diff file Expand all files Collapse all files

4
iota-daemon-lib/Cargo.toml
View file

@ -4,6 +4,7 @@ version = "0.1.0"
edition = "2024"
[dependencies]
async-trait = "0.1.89"
iota-ipc = { path = "../iota-ipc" }
iota-logger = { path = "../iota-logger" }
iota-state = { path = "../iota-state" }
@ -17,3 +18,6 @@ sysinfo = "0.38.3"
tokio = { version = "1.50.0", features = ["full"] }
tokio-util = { version = "0.7", features = ["rt"] }
uuid = { version = "*", features = ["v4"] }
[dev-dependencies]
tempfile = "3"

101
iota-daemon-lib/src/command_router.rs
View file

@ -1,42 +1,34 @@
use crate::DaemonRuntime;
use iota_ipc::{
IpcErrorCode, LocalRequest, ResponseEnvelope, ResponseResult,
};
use crate::{DaemonRuntime, DaemonServices};
use iota_ipc::{ExitIntent, IpcErrorCode, LocalRequest, ResponseEnvelope, ResponseResult};
use iota_logger::{log, log_command};
use iota_storage::users::user_manager;
use iota_storage::util::config_util::modify_config;
use mtp::codec::{CommunicationType, CommunicationValue};
use omikron_connector::omikron_connection::OMIKRON_CONNECTION;
use std::sync::Arc;
use std::time::Duration;
use crate::daemon_state::ShutdownReason;
use crate::daemon_state::{ShutdownReason, StartupPhase};
#[derive(Clone)]
pub struct CommandRouter {
runtime: Arc<DaemonRuntime>,
services: Arc<DaemonServices>,
}
impl CommandRouter {
pub fn new(runtime: Arc<DaemonRuntime>) -> Self {
Self { runtime }
pub fn new(runtime: Arc<DaemonRuntime>, services: Arc<DaemonServices>) -> Self {
Self { runtime, services }
}
pub async fn route(&self, request_id: u64, request: LocalRequest) -> ResponseEnvelope {
log_command!("{:?}", request);
let result = self.execute(request).await;
ResponseEnvelope {
request_id,
result,
}
ResponseEnvelope { request_id, result }
}
/// Parse a legacy console command string into a typed request.
pub fn parse_console_command(line: &str) -> Option<LocalRequest> {
let parts: Vec<&str> = line
.trim_start_matches('/')
.split_whitespace()
.collect();
let parts: Vec<&str> = line.trim_start_matches('/').split_whitespace().collect();
match parts.as_slice() {
["help"] => None,
["tasks"] => Some(LocalRequest::ListTasks),
@ -53,13 +45,33 @@ impl CommandRouter {
["user", "list"] => Some(LocalRequest::ListUsers),
["reconnect"] => Some(LocalRequest::ReconnectOmikron),
["regenerate", "keys"] => Some(LocalRequest::RotateIotaIdentity),
["reload"] | ["restart"] => Some(LocalRequest::RestartDaemon),
["shutdown"] | ["stop"] => Some(LocalRequest::StopDaemon),
["reload"] | ["restart"] => Some(LocalRequest::RequestProcessExit {
intent: ExitIntent::Restart,
}),
["shutdown"] | ["stop"] => Some(LocalRequest::RequestProcessExit {
intent: ExitIntent::Stop,
}),
_ => None,
}
}
async fn execute(&self, request: LocalRequest) -> ResponseResult {
let needs_omikron = matches!(
request,
LocalRequest::CreateUser { .. }
| LocalRequest::RemoveUser { .. }
| LocalRequest::ReconnectOmikron
| LocalRequest::RotateIotaIdentity
);
if needs_omikron && !self.services.omikron.is_connected().await {
return ResponseResult::Error(
if self.runtime.current_startup_phase() != StartupPhase::Ready {
IpcErrorCode::NotReady
} else {
IpcErrorCode::OmikronUnavailable
},
);
}
match request {
LocalRequest::GetStatus => {
let phase = self.runtime.current_startup_phase();
@ -95,10 +107,13 @@ impl CommandRouter {
ResponseResult::Ok(users.join("\n"))
}
LocalRequest::CreateUser { username } => {
match omikron_connector::user_ops::create_user(&username).await {
(Some(user), _) => {
ResponseResult::Ok(format!("Created user {}", user.user_id))
}
match omikron_connector::user_ops::create_user(
self.services.omikron.as_ref(),
&username,
)
.await
{
(Some(user), _) => ResponseResult::Ok(format!("Created user {}", user.user_id)),
_ => ResponseResult::Error(IpcErrorCode::StorageFailure),
}
}
@ -109,24 +124,46 @@ impl CommandRouter {
};
let message = CommunicationValue::new(CommunicationType::DeleteUser)
.with_sender(user.user_id as u64);
if let Err(_e) = OMIKRON_CONNECTION.send_message(&message).await {
if let Err(_e) = self.services.omikron.send_message(&message).await {
return ResponseResult::Error(IpcErrorCode::OmikronUnavailable);
}
user_manager::remove_user(user.user_id);
ResponseResult::Ok(format!("Removed user {}", user.user_id))
}
LocalRequest::ReconnectOmikron => {
OMIKRON_CONNECTION.reconnect().await;
ResponseResult::Ok("Reconnected to Omikron server".into())
}
LocalRequest::ReconnectOmikron => match self.services.omikron.reconnect().await {
Ok(()) => ResponseResult::Ok("Reconnected to Omikron server".into()),
Err(_) => ResponseResult::Error(IpcErrorCode::OmikronUnavailable),
},
LocalRequest::RotateIotaIdentity => {
modify_config(|config| {
config.public_key = None;
config.private_key = None;
config.iota_id = None;
});
OMIKRON_CONNECTION.reconnect().await;
ResponseResult::Ok("Key pair regenerated and Omikron reconnection requested".into())
match self.services.omikron.reconnect().await {
Ok(()) => ResponseResult::Ok(
"Key pair regenerated and Omikron reconnection requested".into(),
),
Err(_) => ResponseResult::Error(IpcErrorCode::OmikronUnavailable),
}
}
LocalRequest::RequestProcessExit { intent } => {
if matches!(intent, ExitIntent::Restart)
&& !matches!(
crate::deployment::from_environment().supervisor,
iota_ipc::SupervisorKind::Systemd | iota_ipc::SupervisorKind::IotaUi
)
{
return ResponseResult::Error(IpcErrorCode::Conflict);
}
self.runtime.shutdown(match intent {
ExitIntent::Stop => ShutdownReason::Stop,
ExitIntent::Restart => ShutdownReason::Restart,
});
ResponseResult::Ok("process exit accepted".into())
}
LocalRequest::GetDaemonStatus => {
ResponseResult::Ok(format!("{:?}", self.runtime.snapshot()))
}
LocalRequest::RestartDaemon => {
self.runtime.shutdown(ShutdownReason::Restart);
@ -140,10 +177,12 @@ impl CommandRouter {
}
pub async fn ping(&self, seconds: u64) -> Result<String, String> {
let response = OMIKRON_CONNECTION
let response = self
.services
.omikron
.await_response(
&CommunicationValue::new(CommunicationType::Ping),
Some(Duration::from_secs(seconds)),
Duration::from_secs(seconds),
)
.await;
match response {

199
iota-daemon-lib/src/daemon_state.rs
View file

@ -1,7 +1,10 @@
use crate::TaskRegistry;
use iota_ipc::StateSnapshot;
use iota_state::DaemonState;
use std::collections::BTreeMap;
use std::sync::Arc;
use std::time::Duration;
use std::time::{SystemTime, UNIX_EPOCH};
use sysinfo::{RefreshKind, System};
use tokio::sync::watch;
use tokio_util::sync::CancellationToken;
@ -58,8 +61,18 @@ pub struct DaemonRuntime {
pub state: Arc<DaemonState>,
pub cancellation: CancellationToken,
pub shutdown_tx: watch::Sender<Option<ShutdownReason>>,
shutdown_rx: watch::Receiver<Option<ShutdownReason>>,
pub startup_phase: watch::Sender<StartupPhase>,
pub degraded_reason: watch::Sender<Option<String>>,
startup_phase_rx: watch::Receiver<StartupPhase>,
degraded_reason_rx: watch::Receiver<Option<String>>,
pub lifecycle: watch::Sender<iota_ipc::LifecyclePhase>,
pub startup_step: watch::Sender<Option<String>>,
pub components: watch::Sender<BTreeMap<iota_ipc::ComponentId, iota_ipc::ComponentHealth>>,
lifecycle_rx: watch::Receiver<iota_ipc::LifecyclePhase>,
startup_step_rx: watch::Receiver<Option<String>>,
components_rx: watch::Receiver<BTreeMap<iota_ipc::ComponentId, iota_ipc::ComponentHealth>>,
pub tasks: TaskRegistry,
}
impl Clone for DaemonRuntime {
@ -68,8 +81,18 @@ impl Clone for DaemonRuntime {
state: self.state.clone(),
cancellation: self.cancellation.clone(),
shutdown_tx: self.shutdown_tx.clone(),
shutdown_rx: self.shutdown_rx.clone(),
startup_phase: self.startup_phase.clone(),
degraded_reason: self.degraded_reason.clone(),
startup_phase_rx: self.startup_phase_rx.clone(),
degraded_reason_rx: self.degraded_reason_rx.clone(),
lifecycle: self.lifecycle.clone(),
startup_step: self.startup_step.clone(),
components: self.components.clone(),
lifecycle_rx: self.lifecycle_rx.clone(),
startup_step_rx: self.startup_step_rx.clone(),
components_rx: self.components_rx.clone(),
tasks: self.tasks.clone(),
}
}
}
@ -82,21 +105,36 @@ impl Default for DaemonRuntime {
impl DaemonRuntime {
pub fn new() -> Self {
let (shutdown_tx, _) = watch::channel(None);
let (startup_phase, _) = watch::channel(StartupPhase::Starting);
let (degraded_reason, _) = watch::channel(None);
let (shutdown_tx, shutdown_rx) = watch::channel(None);
let (startup_phase, startup_phase_rx) = watch::channel(StartupPhase::Starting);
let (degraded_reason, degraded_reason_rx) = watch::channel(None);
let (lifecycle, lifecycle_rx) = watch::channel(iota_ipc::LifecyclePhase::Starting);
let (startup_step, startup_step_rx) = watch::channel(Some("starting".to_string()));
let (components, components_rx) = watch::channel(BTreeMap::new());
Self {
state: Arc::new(DaemonState::new()),
cancellation: CancellationToken::new(),
shutdown_tx,
shutdown_rx,
startup_phase,
degraded_reason,
startup_phase_rx,
degraded_reason_rx,
lifecycle,
startup_step,
components,
lifecycle_rx,
startup_step_rx,
components_rx,
tasks: TaskRegistry::default(),
}
}
pub fn shutdown(&self, reason: ShutdownReason) {
self.cancellation.cancel();
let _ = self.shutdown_tx.send(Some(reason));
if self.shutdown_tx.borrow().is_none() {
let _ = self.shutdown_tx.send(Some(reason));
self.cancellation.cancel();
}
}
pub fn shutdown_reason(&self) -> Option<ShutdownReason> {
@ -109,6 +147,27 @@ impl DaemonRuntime {
pub fn set_startup_phase(&self, phase: StartupPhase) {
let _ = self.startup_phase.send(phase);
let (lifecycle, step) = match phase {
StartupPhase::Ready => (iota_ipc::LifecyclePhase::Ready, None),
StartupPhase::Stopping => (iota_ipc::LifecyclePhase::Stopping, Some("stopping".into())),
StartupPhase::MigratingStorage => (
iota_ipc::LifecyclePhase::Starting,
Some("migrating_storage".into()),
),
StartupPhase::LoadingUsers => (
iota_ipc::LifecyclePhase::Starting,
Some("loading_users".into()),
),
StartupPhase::StartingServices => (
iota_ipc::LifecyclePhase::Starting,
Some("starting_services".into()),
),
StartupPhase::Starting | StartupPhase::Degraded => {
(iota_ipc::LifecyclePhase::Starting, Some("starting".into()))
}
};
let _ = self.lifecycle.send(lifecycle);
let _ = self.startup_step.send(step);
}
pub fn current_startup_phase(&self) -> StartupPhase {
@ -117,7 +176,62 @@ impl DaemonRuntime {
pub fn mark_degraded(&self, reason: String) {
let _ = self.degraded_reason.send(Some(reason.clone()));
let _ = self.startup_phase.send(StartupPhase::Degraded);
self.set_component_degraded(iota_ipc::ComponentId::Omikron, reason);
}
pub fn set_component_healthy(&self, component: iota_ipc::ComponentId, message: Option<String>) {
self.update_component(component, iota_ipc::HealthStatus::Healthy, message);
}
pub fn set_component_degraded(&self, component: iota_ipc::ComponentId, message: String) {
self.update_component(component, iota_ipc::HealthStatus::Degraded, Some(message));
}
pub fn set_component_failed(&self, component: iota_ipc::ComponentId, message: String) {
self.update_component(component, iota_ipc::HealthStatus::Failed, Some(message));
}
fn update_component(
&self,
component: iota_ipc::ComponentId,
status: iota_ipc::HealthStatus,
message: Option<String>,
) {
let mut components = self.components.borrow().clone();
components.insert(
component,
iota_ipc::ComponentHealth {
status,
message,
changed_at_ms: SystemTime::now()
.duration_since(UNIX_EPOCH)
.unwrap_or_default()
.as_millis(),
},
);
let _ = self.components.send(components);
}
pub fn overall_health(&self) -> iota_ipc::HealthStatus {
let components = self.components.borrow();
if [iota_ipc::ComponentId::Ipc, iota_ipc::ComponentId::Storage]
.iter()
.any(|id| {
components
.get(id)
.is_some_and(|v| v.status == iota_ipc::HealthStatus::Failed)
})
{
return iota_ipc::HealthStatus::Failed;
}
if components.values().any(|v| {
v.status == iota_ipc::HealthStatus::Degraded
|| v.status == iota_ipc::HealthStatus::Failed
}) {
iota_ipc::HealthStatus::Degraded
} else {
iota_ipc::HealthStatus::Healthy
}
}
pub fn snapshot(&self) -> StateSnapshot {
@ -133,42 +247,51 @@ impl DaemonRuntime {
net_up: state.net_up.clone(),
net_down: state.net_down.clone(),
sys_info: state.sys_info.clone(),
startup_phase: self.current_startup_phase().into(),
degraded_reason: self.degraded_reason.borrow().clone(),
lifecycle: *self.lifecycle.borrow(),
startup_step: self.startup_step.borrow().clone(),
overall_health: self.overall_health(),
components: self.components.borrow().clone(),
}
}
pub fn spawn_system_monitor(&self) {
pub async fn spawn_system_monitor(&self) {
let runtime = self.clone();
tokio::spawn(async move {
runtime.state.active_tasks.insert("System monitor".into());
let mut system = System::new_with_specifics(RefreshKind::everything());
let mut counter = 0.0;
loop {
if runtime.is_shutting_down() {
break;
self.tasks
.spawn_tracked("system-monitor", async move {
runtime.state.active_tasks.insert("System monitor".into());
let mut system = System::new_with_specifics(RefreshKind::everything());
let mut counter = 0.0;
loop {
if runtime.is_shutting_down() {
break;
}
system.refresh_cpu_all();
system.refresh_memory();
let cpu = system.global_cpu_usage() as f64;
let total_memory = system.total_memory();
let ram = if total_memory == 0 {
0.0
} else {
system.used_memory() as f64 / total_memory as f64 * 100.0
};
{
let mut state = runtime
.state
.app
.lock()
.unwrap_or_else(|error| error.into_inner());
state.push_cpu((counter, cpu));
state.push_ram((counter, ram));
state.sys_info = format!("CPU: {cpu:.1}% RAM: {ram:.1}%");
}
counter += 1.0;
tokio::time::sleep(Duration::from_millis(500)).await;
}
system.refresh_cpu_all();
system.refresh_memory();
let cpu = system.global_cpu_usage() as f64;
let total_memory = system.total_memory();
let ram = if total_memory == 0 {
0.0
} else {
system.used_memory() as f64 / total_memory as f64 * 100.0
};
{
let mut state = runtime
.state
.app
.lock()
.unwrap_or_else(|error| error.into_inner());
state.push_cpu((counter, cpu));
state.push_ram((counter, ram));
state.sys_info = format!("CPU: {cpu:.1}% RAM: {ram:.1}%");
}
counter += 1.0;
tokio::time::sleep(Duration::from_millis(500)).await;
}
runtime.state.active_tasks.remove("System monitor");
});
runtime.state.active_tasks.remove("System monitor");
Ok(())
})
.await;
}
}

37
iota-daemon-lib/src/deployment.rs Normal file
View file

@ -0,0 +1,37 @@
use iota_ipc::{DeploymentMode, SupervisorKind};
#[derive(Clone, Copy, Debug)]
pub struct DeploymentContext {
pub mode: DeploymentMode,
pub supervisor: SupervisorKind,
}
impl Default for DeploymentContext {
fn default() -> Self {
Self {
mode: DeploymentMode::External,
supervisor: SupervisorKind::None,
}
}
}
pub fn from_environment() -> DeploymentContext {
let mut mode = match std::env::var("IOTA_DEPLOYMENT_MODE").ok().as_deref() {
Some("session_child") => DeploymentMode::SessionChild,
Some("ui_auto_start") => DeploymentMode::UiAutoStart,
Some("user_service") => DeploymentMode::UserService,
Some("system_socket_activated") => DeploymentMode::SystemSocketActivated,
Some("system_always_on") => DeploymentMode::SystemAlwaysOn,
_ => DeploymentMode::External,
};
if std::env::var("LISTEN_FDS").ok().as_deref() == Some("1") {
mode = DeploymentMode::SystemSocketActivated;
}
let supervisor = match std::env::var("IOTA_SUPERVISOR").ok().as_deref() {
Some("iota_ui") => SupervisorKind::IotaUi,
Some("systemd") => SupervisorKind::Systemd,
Some("external") => SupervisorKind::External,
_ => SupervisorKind::None,
};
DeploymentContext { mode, supervisor }
}

371
iota-daemon-lib/src/ipc_server.rs
View file

@ -1,63 +1,113 @@
use crate::{CommandRouter, DaemonRuntime};
use crate::deployment::from_environment;
use crate::{CommandRouter, DaemonRuntime, DaemonServices};
use iota_ipc::{
ClientMessage, DaemonMessage, HelloAck, MIN_PROTOCOL_VERSION, PROTOCOL_VERSION, read_msg,
write_msg,
};
use iota_logger::log;
use std::io::Result;
use std::os::unix::fs::{FileTypeExt, MetadataExt, OpenOptionsExt};
use std::path::{Path, PathBuf};
use std::sync::Arc;
use std::{env, os::fd::FromRawFd};
use std::{env, fs::File, os::fd::FromRawFd, os::unix::net::UnixListener as StdUnixListener};
use tokio::net::{UnixListener, UnixStream};
use tokio::sync::{broadcast, mpsc, watch};
use tokio::time::timeout;
use uuid::Uuid;
/// Per-client outbound queue capacity.
const CLIENT_CHANNEL_SIZE: usize = 256;
/// Maximum handshake retries before giving up.
const MAX_HANDSHAKE_RETRIES: u32 = 10;
const MAX_HANDSHAKE_RETRIES: u32 = 1;
const CLIENT_IO_TIMEOUT: std::time::Duration = std::time::Duration::from_secs(15);
pub struct IpcServer {
path: PathBuf,
listener: UnixListener,
runtime: Arc<DaemonRuntime>,
services: Arc<DaemonServices>,
log_tx: broadcast::Sender<DaemonMessage>,
state_rx: watch::Sender<iota_ipc::StateSnapshot>,
state_rx: watch::Receiver<iota_ipc::StateSnapshot>,
instance_id: String,
_instance_lock: File,
}
impl IpcServer {
pub fn new(
pub async fn bind(
path: impl Into<PathBuf>,
runtime: Arc<DaemonRuntime>,
services: Arc<DaemonServices>,
log_tx: broadcast::Sender<DaemonMessage>,
state_rx: watch::Sender<iota_ipc::StateSnapshot>,
) -> Self {
Self {
path: path.into(),
runtime,
log_tx,
state_rx,
}
}
pub async fn run(self) -> Result<()> {
if let Some(parent) = self.path.parent() {
tokio::fs::create_dir_all(parent).await?;
}
state_rx: watch::Receiver<iota_ipc::StateSnapshot>,
) -> Result<Self> {
let path = path.into();
let listener = match activated_listener()? {
Some(listener) => listener,
None => {
remove_stale_socket(&self.path).await?;
UnixListener::bind(&self.path)?
if let Some(parent) = path.parent() {
tokio::fs::create_dir_all(parent).await?;
}
let lock_path = path.with_extension("sock.lock");
let lock = File::options()
.create(true)
.mode(0o600)
.read(true)
.write(true)
.open(lock_path)?;
let locked = unsafe {
libc::flock(
std::os::fd::AsRawFd::as_raw_fd(&lock),
libc::LOCK_EX | libc::LOCK_NB,
)
} == 0;
if !locked {
return Err(std::io::Error::new(
std::io::ErrorKind::AlreadyExists,
"another daemon instance is already running",
));
}
remove_stale_socket(&path).await?;
let listener = UnixListener::bind(&path)?;
let _ = tokio::fs::set_permissions(
&path,
std::os::unix::fs::PermissionsExt::from_mode(0o600),
)
.await;
return Ok(Self {
listener,
runtime,
services,
log_tx,
state_rx,
instance_id: Uuid::new_v4().to_string(),
_instance_lock: lock,
});
}
};
Ok(Self {
listener,
runtime,
services,
log_tx,
state_rx,
instance_id: Uuid::new_v4().to_string(),
_instance_lock: File::options().read(true).open("/dev/null")?,
})
}
pub async fn serve(self) -> Result<()> {
loop {
let (stream, _addr) = listener.accept().await?;
let (stream, _addr) = self.listener.accept().await?;
eprintln!("IPC client accepted");
let runtime = self.runtime.clone();
let services = self.services.clone();
let log_tx = self.log_tx.clone();
let state_rx = self.state_rx.clone();
let instance_id = self.instance_id.clone();
tokio::spawn(async move {
if let Err(error) = handle_client(stream, runtime, log_tx, state_rx).await {
if let Err(error) =
handle_client(stream, runtime, services, log_tx, state_rx, instance_id).await
{
eprintln!("IPC client error: {error}");
}
});
@ -77,13 +127,67 @@ fn activated_listener() -> Result<Option<UnixListener>> {
if listen_fds != Some(1) || listen_pid != Some(std::process::id()) {
return Ok(None);
}
let listener = unsafe { std::os::unix::net::UnixListener::from_raw_fd(3) };
UnixListener::from_std(listener).map(Some)
// SAFETY: systemd transfers ownership of the activated descriptor to us.
let listener = unsafe { StdUnixListener::from_raw_fd(3) };
into_tokio_listener(listener).map(Some)
}
fn into_tokio_listener(listener: StdUnixListener) -> Result<UnixListener> {
listener.set_nonblocking(true)?;
UnixListener::from_std(listener)
}
async fn write_client_message<W>(writer: &mut W, message: &DaemonMessage) -> Result<()>
where
W: tokio::io::AsyncWrite + Unpin,
{
timeout(CLIENT_IO_TIMEOUT, write_msg(writer, message))
.await
.map_err(|_| {
std::io::Error::new(std::io::ErrorKind::TimedOut, "IPC client write timed out")
})?
}
async fn remove_stale_socket(path: &Path) -> Result<()> {
match tokio::fs::symlink_metadata(path).await {
Ok(_) => tokio::fs::remove_file(path).await,
Ok(metadata) => {
if metadata.file_type().is_symlink() || !metadata.file_type().is_socket() {
return Err(std::io::Error::new(
std::io::ErrorKind::InvalidInput,
"IPC path exists but is not an owned Unix socket",
));
}
if metadata.uid() != unsafe { libc::geteuid() } as u32 {
return Err(std::io::Error::new(
std::io::ErrorKind::PermissionDenied,
"existing IPC socket is not owned by the current user",
));
}
match timeout(
std::time::Duration::from_millis(250),
UnixStream::connect(path),
)
.await
{
Ok(Ok(_)) => Err(std::io::Error::new(
std::io::ErrorKind::AddrInUse,
"an IPC daemon is already listening",
)),
Ok(Err(error))
if matches!(
error.kind(),
std::io::ErrorKind::ConnectionRefused | std::io::ErrorKind::NotFound
) =>
{
tokio::fs::remove_file(path).await
}
Ok(Err(error)) => Err(error),
Err(_) => Err(std::io::Error::new(
std::io::ErrorKind::TimedOut,
"could not determine whether the existing IPC socket is active",
)),
}
}
Err(error) if error.kind() == std::io::ErrorKind::NotFound => Ok(()),
Err(error) => Err(error),
}
@ -93,10 +197,10 @@ async fn remove_stale_socket(path: &Path) -> Result<()> {
struct PeerIdentity {
pid: i32,
uid: u32,
gid: u32,
_gid: u32,
}
fn peer_credentials(stream: &UnixStream) -> PeerIdentity {
fn peer_credentials(stream: &UnixStream) -> Result<PeerIdentity> {
#[cfg(target_os = "linux")]
{
use std::os::unix::io::AsRawFd;
@ -104,70 +208,109 @@ fn peer_credentials(stream: &UnixStream) -> PeerIdentity {
let mut cred: libc::ucred = std::mem::zeroed();
let mut len = std::mem::size_of::<libc::ucred>() as libc::socklen_t;
let fd = stream.as_raw_fd();
libc::getsockopt(
if libc::getsockopt(
fd,
libc::SOL_SOCKET,
libc::SO_PEERCRED,
&mut cred as *mut _ as *mut libc::c_void,
&mut len,
);
PeerIdentity {
) != 0
{
return Err(std::io::Error::last_os_error());
}
Ok(PeerIdentity {
pid: cred.pid,
uid: cred.uid,
gid: cred.gid,
}
_gid: cred.gid,
})
}
}
#[cfg(not(target_os = "linux"))]
{
PeerIdentity {
Ok(PeerIdentity {
pid: 0,
uid: 0,
gid: 0,
}
_gid: 0,
})
}
}
async fn handle_client(
stream: UnixStream,
runtime: Arc<DaemonRuntime>,
services: Arc<DaemonServices>,
log_tx: broadcast::Sender<DaemonMessage>,
_state_rx: watch::Sender<iota_ipc::StateSnapshot>,
mut state_rx: watch::Receiver<iota_ipc::StateSnapshot>,
instance_id: String,
) -> Result<()> {
let peer = peer_credentials(&stream);
let peer = peer_credentials(&stream)?;
// Access control belongs to the Unix socket. The systemd socket grants
// iota-operators group access (0660); rejecting every UID other than the
// service account here would make that authorization ineffective. Manual
// sockets remain owner-only (0600) at bind time.
let (mut reader, mut writer) = stream.into_split();
// A failed writer must stop the reader and any subsequent command work
// for this client; otherwise the reader can remain parked forever.
let session_cancellation = runtime.cancellation.child_token();
let (directed_tx, directed_rx) = mpsc::channel::<DaemonMessage>(CLIENT_CHANNEL_SIZE);
eprintln!("IPC handshake started (pid={}, uid={})", peer.pid, peer.uid);
// --- Handshake ---
let mut negotiated_version: Option<u16> = None;
for _ in 0..MAX_HANDSHAKE_RETRIES {
match read_msg::<_, ClientMessage>(&mut reader).await {
Ok(ClientMessage::Hello { supported_versions }) => {
let version = supported_versions
.iter()
.copied()
.find(|v| *v >= MIN_PROTOCOL_VERSION && *v <= PROTOCOL_VERSION)
.unwrap_or(PROTOCOL_VERSION);
negotiated_version = Some(version);
let instance_id = Uuid::new_v4().to_string();
let ack = DaemonMessage::HelloAck(HelloAck {
protocol_version: version,
daemon_version: env!("CARGO_PKG_VERSION").to_string(),
instance_id,
startup_phase: runtime.current_startup_phase().into(),
capabilities: vec!["commands".into(), "metrics".into(), "logs".into()],
});
write_msg(&mut writer, &ack).await?;
break;
}
Ok(_) => {
// Unexpected first message — send error and close.
match timeout(
std::time::Duration::from_secs(15),
read_msg::<_, ClientMessage>(&mut reader),
)
.await
{
Err(_) => {
return Err(std::io::Error::new(
std::io::ErrorKind::InvalidData,
"Expected Hello as first message",
std::io::ErrorKind::TimedOut,
"IPC Hello timed out",
));
}
Err(e) => return Err(e),
Ok(result) => match result {
Ok(ClientMessage::Hello { supported_versions }) => {
let version = supported_versions
.iter()
.copied()
.filter(|v| *v >= MIN_PROTOCOL_VERSION && *v <= PROTOCOL_VERSION)
.max()
.ok_or_else(|| {
std::io::Error::new(
std::io::ErrorKind::Unsupported,
"No compatible IPC protocol version",
)
})?;
negotiated_version = Some(version);
let ack = DaemonMessage::HelloAck(HelloAck {
protocol_version: version,
daemon_version: env!("CARGO_PKG_VERSION").to_string(),
instance_id: instance_id.clone(),
startup_phase: runtime.current_startup_phase().into(),
capabilities: vec!["commands".into(), "metrics".into(), "logs".into()],
lifecycle: *runtime.lifecycle.borrow(),
health: runtime.overall_health(),
deployment_mode: from_environment().mode,
supervisor: from_environment().supervisor,
});
write_client_message(&mut writer, &ack).await?;
eprintln!(
"IPC handshake acknowledged (pid={}, uid={})",
peer.pid, peer.uid
);
break;
}
Ok(_) => {
// Unexpected first message — send error and close.
return Err(std::io::Error::new(
std::io::ErrorKind::InvalidData,
"Expected Hello as first message",
));
}
Err(e) => return Err(e),
},
}
}
let _version = negotiated_version.ok_or_else(|| {
@ -182,9 +325,9 @@ async fn handle_client(
// --- Writer task: merge directed responses + shared log events ---
let mut log_rx = log_tx.subscribe();
let directed_for_writer = directed_tx.clone();
let writer_task = {
let runtime = runtime.clone();
let session_cancellation = session_cancellation.clone();
tokio::spawn(async move {
let mut directed_rx = directed_rx;
loop {
@ -193,7 +336,9 @@ async fn handle_client(
msg = directed_rx.recv() => {
match msg {
Some(message) => {
if write_msg(&mut writer, &message).await.is_err() {
if let Err(error) = write_client_message(&mut writer, &message).await {
eprintln!("IPC client writer stopped while sending directed message: {error}");
session_cancellation.cancel();
break;
}
}
@ -204,36 +349,91 @@ async fn handle_client(
result = log_rx.recv() => {
match result {
Ok(message) => {
if write_msg(&mut writer, &message).await.is_err() {
if let Err(error) = write_client_message(&mut writer, &message).await {
eprintln!("IPC client writer stopped while sending log message: {error}");
session_cancellation.cancel();
break;
}
}
Err(broadcast::error::RecvError::Lagged(skipped)) => {
let _ = directed_for_writer.send(DaemonMessage::Gap { skipped }).await;
// Then send current snapshot for resync
let _ = directed_for_writer.send(
DaemonMessage::StateUpdate(runtime.snapshot())
).await;
if write_client_message(&mut writer, &DaemonMessage::Gap { skipped }).await.is_err()
|| write_client_message(&mut writer, &DaemonMessage::StateUpdate(runtime.snapshot())).await.is_err()
{
session_cancellation.cancel();
break;
}
}
Err(broadcast::error::RecvError::Closed) => break,
}
}
changed = state_rx.changed() => {
if changed.is_err() {
break;
}
let snapshot = state_rx.borrow().clone();
if let Err(error) = write_client_message(&mut writer, &DaemonMessage::StateUpdate(snapshot)).await {
eprintln!("IPC client writer stopped while sending state update: {error}");
session_cancellation.cancel();
break;
}
}
}
}
})
};
// --- Reader loop ---
let router = CommandRouter::new(runtime.clone());
let router = CommandRouter::new(runtime.clone(), services);
loop {
match read_msg::<_, ClientMessage>(&mut reader).await {
let message = tokio::select! {
_ = session_cancellation.cancelled() => break,
result = read_msg::<_, ClientMessage>(&mut reader) => result,
};
match message {
Ok(ClientMessage::Request(envelope)) => {
let response = router.route(envelope.request_id, envelope.request).await;
let shutdown_reason = match &envelope.request {
iota_ipc::LocalRequest::RequestProcessExit {
intent: iota_ipc::ExitIntent::Restart,
}
| iota_ipc::LocalRequest::RestartDaemon => Some("restart requested"),
iota_ipc::LocalRequest::RequestProcessExit {
intent: iota_ipc::ExitIntent::Stop,
} => Some("shutdown requested"),
iota_ipc::LocalRequest::StopDaemon => Some("shutdown requested"),
_ => None,
};
let response = if envelope.protocol_version < MIN_PROTOCOL_VERSION
|| envelope.protocol_version > PROTOCOL_VERSION
{
iota_ipc::ResponseEnvelope {
request_id: envelope.request_id,
result: iota_ipc::ResponseResult::Error(
iota_ipc::IpcErrorCode::UnsupportedVersion,
),
}
} else {
router.route(envelope.request_id, envelope.request).await
};
let _ = directed_tx.send(DaemonMessage::Response(response)).await;
if let Some(reason) = shutdown_reason {
let _ = directed_tx
.send(DaemonMessage::LifecycleEvent(
iota_ipc::LifecycleEvent::Shutdown {
reason: reason.into(),
},
))
.await;
// The request itself initiates daemon cancellation. Give
// the dedicated writer a chance to flush the response
// and lifecycle event before this session is torn down.
tokio::time::sleep(std::time::Duration::from_millis(100)).await;
break;
}
}
Ok(ClientMessage::Subscribe { .. }) => {
let snapshot = DaemonMessage::StateUpdate(runtime.snapshot());
let _ = directed_tx.send(snapshot).await;
let _ = directed_tx.send(DaemonMessage::Subscribed).await;
}
Ok(ClientMessage::Ping { seq }) => {
let _ = directed_tx.send(DaemonMessage::Pong { seq }).await;
@ -250,6 +450,7 @@ async fn handle_client(
}
}
}
session_cancellation.cancel();
writer_task.abort();
log!(
"IPC client disconnected (pid={}, uid={})",
@ -258,3 +459,25 @@ async fn handle_client(
);
Ok(())
}
#[cfg(test)]
mod tests {
use super::*;
use std::time::Duration;
#[tokio::test(flavor = "current_thread")]
async fn converted_listener_does_not_block_the_runtime() {
let directory = tempfile::tempdir().unwrap();
let path = directory.path().join("ipc.sock");
let listener = match StdUnixListener::bind(path) {
Ok(listener) => into_tokio_listener(listener).unwrap(),
Err(error) if error.kind() == std::io::ErrorKind::PermissionDenied => return,
Err(error) => panic!("could not create test socket: {error}"),
};
assert!(
tokio::time::timeout(Duration::from_millis(50), listener.accept())
.await
.is_err()
);
}
}

5
iota-daemon-lib/src/lib.rs
View file

@ -1,8 +1,13 @@
pub mod command_router;
pub mod daemon_state;
pub mod deployment;
pub mod ipc_server;
pub mod log_broadcaster;
pub mod services;
pub mod task_registry;
pub use command_router::CommandRouter;
pub use daemon_state::{DaemonRuntime, ShutdownReason, StartupPhase};
pub use ipc_server::IpcServer;
pub use services::DaemonServices;
pub use task_registry::TaskRegistry;

23
iota-daemon-lib/src/services.rs Normal file
View file

@ -0,0 +1,23 @@
use omikron_connector::{OmikronClient, OmikronConnection};
use std::sync::Arc;
#[derive(Default)]
pub struct UserService;
#[derive(Default)]
pub struct ConfigService;
pub struct DaemonServices {
pub omikron: Arc<dyn OmikronClient>,
pub users: Arc<UserService>,
pub config: Arc<ConfigService>,
}
impl DaemonServices {
pub fn new(omikron: Arc<OmikronConnection>) -> Arc<Self> {
Arc::new(Self {
omikron,
users: Arc::new(UserService),
config: Arc::new(ConfigService),
})
}
}

40
iota-daemon-lib/src/task_registry.rs Normal file
View file

@ -0,0 +1,40 @@
use std::sync::Arc;
use std::time::Duration;
use tokio::sync::Mutex;
use tokio::task::JoinSet;
#[derive(Clone, Default)]
pub struct TaskRegistry {
tasks: Arc<Mutex<JoinSet<(String, Result<(), String>)>>>,
}
impl TaskRegistry {
pub async fn spawn_tracked<F>(&self, name: impl Into<String>, future: F)
where
F: std::future::Future<Output = Result<(), String>> + Send + 'static,
{
let name = name.into();
self.tasks
.lock()
.await
.spawn(async move { (name, future.await) });
}
pub async fn join_with_timeout(&self, timeout: Duration) -> Vec<String> {
let mut tasks = self.tasks.lock().await;
let mut failures = Vec::new();
let deadline = tokio::time::Instant::now() + timeout;
while !tasks.is_empty() {
match tokio::time::timeout_at(deadline, tasks.join_next()).await {
Ok(Some(Ok((name, Err(error))))) => failures.push(format!("{name}: {error}")),
Ok(Some(Ok((_, Ok(()))))) | Ok(Some(Err(_))) => {}
Ok(None) => break,
Err(_) => {
tasks.abort_all();
break;
}
}
}
failures
}
}

52
iota-daemon-lib/tests/command_router.rs Normal file
View file

@ -0,0 +1,52 @@
use async_trait::async_trait;
use iota_daemon_lib::{CommandRouter, DaemonRuntime, DaemonServices};
use iota_ipc::{LocalRequest, ResponseResult};
use mtp::codec::CommunicationValue;
use omikron_connector::{OmikronClient, OmikronError};
use std::sync::{
Arc,
atomic::{AtomicUsize, Ordering},
};
use std::time::Duration;
struct FakeOmikron {
reconnects: AtomicUsize,
}
#[async_trait]
impl OmikronClient for FakeOmikron {
async fn send_message(&self, _: &CommunicationValue) -> Result<(), OmikronError> {
Ok(())
}
async fn await_response(
&self,
_: &CommunicationValue,
_: Duration,
) -> Result<CommunicationValue, OmikronError> {
Err(OmikronError::Disconnected("fake".into()))
}
async fn reconnect(&self) -> Result<(), OmikronError> {
self.reconnects.fetch_add(1, Ordering::SeqCst);
Ok(())
}
async fn is_connected(&self) -> bool {
true
}
}
#[tokio::test]
async fn reconnect_uses_the_injected_client() {
let fake = Arc::new(FakeOmikron {
reconnects: AtomicUsize::new(0),
});
let services = Arc::new(DaemonServices {
omikron: fake.clone(),
users: Default::default(),
config: Default::default(),
});
let router = CommandRouter::new(Arc::new(DaemonRuntime::new()), services);
assert!(matches!(
router.route(1, LocalRequest::ReconnectOmikron).await.result,
ResponseResult::Ok(_)
));
assert_eq!(fake.reconnects.load(Ordering::SeqCst), 1);
}

29
iota-daemon-lib/tests/daemon_health.rs Normal file
View file

@ -0,0 +1,29 @@
use iota_daemon_lib::{DaemonRuntime, StartupPhase};
use iota_ipc::{ComponentId, HealthStatus, LifecyclePhase};
#[test]
fn component_failures_are_independent_and_recovery_is_scoped() {
let runtime = DaemonRuntime::new();
runtime.set_component_degraded(ComponentId::Omikron, "offline".into());
runtime.set_component_failed(ComponentId::Web, "bind failed".into());
runtime.set_startup_phase(StartupPhase::Ready);
let snapshot = runtime.snapshot();
assert_eq!(snapshot.lifecycle, LifecyclePhase::Ready);
assert_eq!(snapshot.overall_health, HealthStatus::Degraded);
assert_eq!(
snapshot.components[&ComponentId::Omikron].status,
HealthStatus::Degraded
);
runtime.set_component_healthy(ComponentId::Web, None);
assert_eq!(
runtime.snapshot().components[&ComponentId::Omikron].status,
HealthStatus::Degraded
);
}
#[test]
fn critical_failure_is_failed_but_optional_degradation_is_not() {
let runtime = DaemonRuntime::new();
runtime.set_component_failed(ComponentId::Storage, "database unavailable".into());
assert_eq!(runtime.snapshot().overall_health, HealthStatus::Failed);
}

40
iota-daemon-lib/tests/shutdown.rs Normal file
View file

@ -0,0 +1,40 @@
use iota_daemon_lib::{DaemonRuntime, ShutdownReason};
use std::time::Duration;
#[tokio::test]
async fn shutdown_reason_is_first_write_wins_and_tasks_join() {
let runtime = DaemonRuntime::new();
runtime.shutdown(ShutdownReason::Fatal("first".into()));
runtime.shutdown(ShutdownReason::Restart);
assert_eq!(
runtime.shutdown_reason(),
Some(ShutdownReason::Fatal("first".into()))
);
runtime.tasks.spawn_tracked("quick", async { Ok(()) }).await;
assert!(
runtime
.tasks
.join_with_timeout(Duration::from_millis(100))
.await
.is_empty()
);
}
#[tokio::test]
async fn long_task_is_aborted_at_join_timeout() {
let runtime = DaemonRuntime::new();
runtime
.tasks
.spawn_tracked("slow", async {
tokio::time::sleep(Duration::from_secs(10)).await;
Ok(())
})
.await;
assert!(
runtime
.tasks
.join_with_timeout(Duration::from_millis(10))
.await
.is_empty()
);
}