use std::{
collections::VecDeque,
fmt,
sync::{Arc, Mutex as StdMutex, RwLock},
time::Duration,
};
use async_trait::async_trait;
use iroh_tickets::endpoint::EndpointTicket;
use serde::Serialize;
use tokio::{
sync::{Mutex, broadcast},
task::JoinHandle,
};
use tokio_util::sync::CancellationToken;
use url::{Url, form_urlencoded};
use uuid::Uuid;
use super::lock::RuntimeLock;
use crate::{
config::paths::AppPaths,
domain::errors::{AgentError, AgentResult, ErrorCode},
presentation::ui_handshake::UiPairingService,
transport::iroh_endpoint::{IrohEndpointFactory, IrohEndpointStatus, RunningIrohEndpoint},
};
const STATUS_CHANNEL_CAPACITY: usize = 64;
const EVENT_CHANNEL_CAPACITY: usize = 64;
const EVENT_HISTORY_CAPACITY: usize = 20;
const MAX_PUBLIC_RETRY_DELAY: Duration = Duration::from_secs(30);
const PUBLIC_READINESS_DEADLINE: Duration = Duration::from_secs(10);
#[derive(Clone, PartialEq, Eq)]
pub(crate) struct RuntimeOptions {
pub frontend_origin: String,
}
#[derive(Debug, Clone, Copy, Serialize, PartialEq, Eq)]
#[serde(rename_all = "snake_case")]
pub(crate) enum RuntimePhase {
Starting,
Ready,
Degraded,
Stopping,
Stopped,
}
#[derive(Debug, Clone, Copy, Serialize, PartialEq, Eq)]
#[serde(rename_all = "snake_case")]
pub(crate) enum RuntimeTransport {
Iroh,
}
#[derive(Clone, Serialize, PartialEq, Eq)]
pub(crate) struct RuntimeStatus {
pub phase: RuntimePhase,
pub transport: RuntimeTransport,
pub server_id: String,
pub iroh_endpoint_id: Option<String>,
#[serde(skip_serializing)]
pub iroh_endpoint_ticket: Option<String>,
pub frontend_deep_link: Option<String>,
pub pairing_code: Option<String>,
pub pairing_expires_at: Option<String>,
pub message: Option<String>,
}
#[derive(Clone, PartialEq, Eq)]
pub(crate) enum RuntimeEvent {
Starting,
Ready,
Degraded,
Stopping,
Stopped,
}
impl fmt::Debug for RuntimeEvent {
fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
formatter.write_str(match self {
Self::Starting => "Starting",
Self::Ready => "Ready",
Self::Degraded => "Degraded",
Self::Stopping => "Stopping",
Self::Stopped => "Stopped",
})
}
}
pub(crate) struct FrontendDeepLink(String);
struct IrohFragment {
endpoint: String,
ticket: String,
server: String,
pair: String,
}
impl FrontendDeepLink {
pub(crate) fn iroh(
frontend_origin: &str,
endpoint: &str,
ticket: &str,
server_id: &str,
pairing_code: &str,
) -> AgentResult<Self> {
validate_exact_origin(frontend_origin)?;
let fragment = form_urlencoded::Serializer::new(String::new())
.append_pair("transport", "iroh")
.append_pair("endpoint", endpoint)
.append_pair("ticket", ticket)
.append_pair("server", server_id)
.append_pair("pair", pairing_code)
.finish();
Self::from_iroh_fragment(frontend_origin, &fragment)
}
pub(crate) fn from_iroh_fragment(frontend_origin: &str, fragment: &str) -> AgentResult<Self> {
validate_exact_origin(frontend_origin)?;
let IrohFragment {
endpoint,
ticket,
server,
pair,
} = parse_iroh_fragment(fragment)?;
let canonical = form_urlencoded::Serializer::new(String::new())
.append_pair("transport", "iroh")
.append_pair("endpoint", &endpoint)
.append_pair("ticket", &ticket)
.append_pair("server", &server)
.append_pair("pair", &pair)
.finish();
Ok(Self(format!("{frontend_origin}/#{canonical}")))
}
pub(crate) fn as_str(&self) -> &str {
&self.0
}
}
impl fmt::Debug for FrontendDeepLink {
fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
formatter.write_str("FrontendDeepLink(<redacted>)")
}
}
fn parse_iroh_fragment(fragment: &str) -> AgentResult<IrohFragment> {
let mut transport = None;
let mut endpoint = None;
let mut ticket = None;
let mut server = None;
let mut pair = None;
for (key, value) in form_urlencoded::parse(fragment.as_bytes()) {
let target = match key.as_ref() {
"transport" => &mut transport,
"endpoint" => &mut endpoint,
"ticket" => &mut ticket,
"server" => &mut server,
"pair" => &mut pair,
_ => return Err(invalid_runtime_configuration()),
};
if target.replace(value.into_owned()).is_some() {
return Err(invalid_runtime_configuration());
}
}
if transport.as_deref() != Some("iroh") {
return Err(invalid_runtime_configuration());
}
let (Some(endpoint), Some(ticket), Some(server), Some(pair)) = (endpoint, ticket, server, pair)
else {
return Err(invalid_runtime_configuration());
};
if server.is_empty() || pair.is_empty() {
return Err(invalid_runtime_configuration());
}
let parsed_endpoint_id = endpoint
.parse::<iroh::EndpointId>()
.map_err(|_| invalid_runtime_configuration())?;
let ticket_value = ticket
.parse::<EndpointTicket>()
.map_err(|_| invalid_runtime_configuration())?;
if ticket_value.endpoint_addr().id != parsed_endpoint_id {
return Err(invalid_runtime_configuration());
}
Ok(IrohFragment {
endpoint: parsed_endpoint_id.to_string(),
ticket,
server,
pair,
})
}
#[async_trait]
pub(crate) trait RuntimeApplication: Send + Sync {
async fn run(&self, cancellation: CancellationToken) -> AgentResult<()>;
}
#[async_trait]
pub(crate) trait RuntimeShutdown: Send + Sync {
async fn shutdown(&self);
}
#[async_trait]
pub(crate) trait IrohEndpointStarter: Send + Sync {
async fn start(&self, cancellation: CancellationToken) -> AgentResult<RunningIrohEndpoint>;
}
#[async_trait]
impl IrohEndpointStarter for IrohEndpointFactory {
async fn start(&self, cancellation: CancellationToken) -> AgentResult<RunningIrohEndpoint> {
IrohEndpointFactory::start(self, cancellation).await
}
}
pub(crate) struct RuntimeDependencies {
pub application: Arc<dyn RuntimeApplication>,
pub iroh_endpoint_factory: Arc<dyn IrohEndpointStarter>,
pub shutdown: Arc<dyn RuntimeShutdown>,
}
pub(crate) struct RuntimeSupervisor {
options: RuntimeOptions,
paths: AppPaths,
pairing: Arc<UiPairingService>,
dependencies: RuntimeDependencies,
}
impl fmt::Debug for RuntimeSupervisor {
fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
formatter.write_str("RuntimeSupervisor { redacted }")
}
}
impl RuntimeSupervisor {
pub(crate) fn with_dependencies(
options: RuntimeOptions,
paths: AppPaths,
pairing: Arc<UiPairingService>,
dependencies: RuntimeDependencies,
) -> Self {
Self {
options,
paths,
pairing,
dependencies,
}
}
pub(crate) async fn start(self) -> AgentResult<RunningRuntime> {
validate_options(&self.options)?;
let runtime_lock = RuntimeLock::acquire(&self.paths.server_lock_file())?;
let initial = RuntimeStatus {
phase: RuntimePhase::Starting,
transport: RuntimeTransport::Iroh,
server_id: self.pairing.server_id().as_str().to_owned(),
iroh_endpoint_id: None,
iroh_endpoint_ticket: None,
frontend_deep_link: None,
pairing_code: None,
pairing_expires_at: None,
message: None,
};
let shared = Arc::new(RuntimeShared::new(initial));
shared.publish_event(RuntimeEvent::Starting);
let cancellation = CancellationToken::new();
let application_cancellation = cancellation.child_token();
let application = self.dependencies.application.clone();
let application_task =
tokio::spawn(async move { application.run(application_cancellation).await });
let endpoint_cancellation = cancellation.child_token();
let endpoint = match self
.dependencies
.iroh_endpoint_factory
.start(endpoint_cancellation.clone())
.await
{
Ok(endpoint) => endpoint,
Err(error) => {
cancellation.cancel();
let _ = application_task.await;
self.dependencies.shutdown.shutdown().await;
return Err(error);
}
};
if let Err(error) = publish_iroh_status(
&shared,
&self.pairing,
&self.options.frontend_origin,
endpoint.status.borrow().clone(),
true,
) {
cancellation.cancel();
let _ = endpoint.task.await;
let _ = application_task.await;
self.dependencies.shutdown.shutdown().await;
return Err(error);
}
let coordinator = tokio::spawn(run_public(PublicRuntime {
_runtime_lock: runtime_lock,
pairing: self.pairing,
frontend_origin: self.options.frontend_origin,
endpoint_factory: self.dependencies.iroh_endpoint_factory,
shutdown: self.dependencies.shutdown,
shared: shared.clone(),
cancellation: cancellation.clone(),
server_task: Some(application_task),
endpoint: Some(endpoint),
endpoint_cancellation: Some(endpoint_cancellation),
retry_attempt: 0,
}));
*shared.coordinator.lock().await = Some(coordinator);
Ok(RunningRuntime {
shared,
cancellation,
})
}
}
pub(crate) struct RunningRuntime {
shared: Arc<RuntimeShared>,
cancellation: CancellationToken,
}
impl fmt::Debug for RunningRuntime {
fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
formatter.write_str("RunningRuntime { redacted }")
}
}
impl RunningRuntime {
pub(crate) fn status(&self) -> RuntimeStatus {
self.shared.status()
}
pub(crate) fn subscribe_statuses(&self) -> broadcast::Receiver<RuntimeStatus> {
self.shared.statuses.subscribe()
}
pub(crate) fn subscribe_events_with_history(
&self,
) -> (Vec<RuntimeEvent>, broadcast::Receiver<RuntimeEvent>) {
let history = self
.shared
.event_history
.lock()
.unwrap_or_else(|poisoned| poisoned.into_inner());
let receiver = self.shared.events.subscribe();
(history.iter().cloned().collect(), receiver)
}
pub(crate) async fn stop(&self) -> AgentResult<()> {
self.cancellation.cancel();
let task = self.shared.coordinator.lock().await.take();
match task {
Some(task) => task.await.map_err(|_| runtime_task_failed())?,
None => Ok(()),
}
}
}
impl Drop for RunningRuntime {
fn drop(&mut self) {
self.cancellation.cancel();
}
}
struct RuntimeShared {
current: RwLock<RuntimeStatus>,
statuses: broadcast::Sender<RuntimeStatus>,
events: broadcast::Sender<RuntimeEvent>,
event_history: StdMutex<VecDeque<RuntimeEvent>>,
coordinator: Mutex<Option<JoinHandle<AgentResult<()>>>>,
}
impl RuntimeShared {
fn new(initial: RuntimeStatus) -> Self {
let (statuses, _) = broadcast::channel(STATUS_CHANNEL_CAPACITY);
let (events, _) = broadcast::channel(EVENT_CHANNEL_CAPACITY);
Self {
current: RwLock::new(initial),
statuses,
events,
event_history: StdMutex::new(VecDeque::with_capacity(EVENT_HISTORY_CAPACITY)),
coordinator: Mutex::new(None),
}
}
fn status(&self) -> RuntimeStatus {
self.current
.read()
.unwrap_or_else(|poisoned| poisoned.into_inner())
.clone()
}
fn publish_status(&self, status: RuntimeStatus) {
let mut current = self
.current
.write()
.unwrap_or_else(|poisoned| poisoned.into_inner());
if *current == status {
return;
}
*current = status.clone();
let _ = self.statuses.send(status);
}
fn publish_event(&self, event: RuntimeEvent) {
tracing::info!(event = ?event, "runtime lifecycle changed");
let mut history = self
.event_history
.lock()
.unwrap_or_else(|poisoned| poisoned.into_inner());
if history.len() == EVENT_HISTORY_CAPACITY {
history.pop_front();
}
history.push_back(event.clone());
let _ = self.events.send(event);
}
}
struct PublicRuntime {
_runtime_lock: RuntimeLock,
pairing: Arc<UiPairingService>,
frontend_origin: String,
endpoint_factory: Arc<dyn IrohEndpointStarter>,
shutdown: Arc<dyn RuntimeShutdown>,
shared: Arc<RuntimeShared>,
cancellation: CancellationToken,
server_task: Option<JoinHandle<AgentResult<()>>>,
endpoint: Option<RunningIrohEndpoint>,
endpoint_cancellation: Option<CancellationToken>,
retry_attempt: u32,
}
enum PublicMonitorResult {
Cancelled,
Server(AgentResult<()>),
EndpointFailed,
ReadinessTimedOut,
RelayLost,
}
async fn run_public(mut runtime: PublicRuntime) -> AgentResult<()> {
let mut result = Ok(());
loop {
if runtime.endpoint.is_none() {
let delay = public_retry_delay(runtime.retry_attempt, random_jitter());
runtime.retry_attempt = runtime.retry_attempt.saturating_add(1);
let should_stop = tokio::select! {
biased;
_ = runtime.cancellation.cancelled() => true,
_ = tokio::time::sleep(delay) => false,
};
if should_stop {
break;
}
let endpoint_cancellation = runtime.cancellation.child_token();
match runtime
.endpoint_factory
.start(endpoint_cancellation.clone())
.await
{
Ok(endpoint) => {
let endpoint_status = endpoint.status.borrow().clone();
if let Err(error) = publish_iroh_status(
&runtime.shared,
&runtime.pairing,
&runtime.frontend_origin,
endpoint_status,
false,
) {
endpoint_cancellation.cancel();
let _ = endpoint.task.await;
result = Err(error);
break;
}
runtime.endpoint = Some(endpoint);
runtime.endpoint_cancellation = Some(endpoint_cancellation);
}
Err(_) => {
publish_degraded(
&runtime.shared,
&runtime.pairing,
"Iroh endpoint restart failed",
);
continue;
}
}
}
let endpoint = runtime.endpoint.as_mut().expect("endpoint was started");
let monitor = monitor_public_endpoint(
&mut runtime.server_task,
endpoint,
&runtime.cancellation,
&runtime.shared,
&runtime.pairing,
&runtime.frontend_origin,
)
.await;
let endpoint_cancellation = runtime
.endpoint_cancellation
.take()
.expect("endpoint cancellation is present");
endpoint_cancellation.cancel();
let endpoint = runtime.endpoint.take().expect("endpoint is present");
if !endpoint.task.is_finished() {
let _ = endpoint.task.await;
}
match monitor {
PublicMonitorResult::Cancelled => break,
PublicMonitorResult::Server(server_result) => {
result = server_result;
break;
}
PublicMonitorResult::EndpointFailed => {
publish_degraded(&runtime.shared, &runtime.pairing, "Iroh endpoint stopped");
}
PublicMonitorResult::ReadinessTimedOut => {
publish_degraded(
&runtime.shared,
&runtime.pairing,
"Iroh relay readiness timed out",
);
}
PublicMonitorResult::RelayLost => {
publish_degraded(
&runtime.shared,
&runtime.pairing,
"Iroh relay is unavailable",
);
}
}
}
publish_stopping(&runtime.shared, &runtime.pairing);
runtime.cancellation.cancel();
if let Some(server_task) = runtime.server_task.take()
&& !server_task.is_finished()
{
let _ = server_task.await;
}
runtime.shutdown.shutdown().await;
publish_stopped(&runtime.shared);
result
}
async fn monitor_public_endpoint(
server_task: &mut Option<JoinHandle<AgentResult<()>>>,
endpoint: &mut RunningIrohEndpoint,
cancellation: &CancellationToken,
shared: &RuntimeShared,
pairing: &UiPairingService,
frontend_origin: &str,
) -> PublicMonitorResult {
let readiness_deadline = tokio::time::Instant::now() + PUBLIC_READINESS_DEADLINE;
let mut awaiting_relay = {
let status = endpoint.status.borrow();
!status.relay_ready || status.endpoint_ticket.is_none()
};
loop {
tokio::select! {
biased;
_ = cancellation.cancelled() => return PublicMonitorResult::Cancelled,
_ = tokio::time::sleep_until(readiness_deadline), if awaiting_relay => {
return PublicMonitorResult::ReadinessTimedOut;
}
server = async {
match server_task.as_mut() {
Some(task) => Some(task.await),
None => std::future::pending().await,
}
} => {
*server_task = None;
return PublicMonitorResult::Server(flatten_task(server.expect("server task is present")));
}
task = &mut endpoint.task => {
return match flatten_task(task) {
Ok(()) => PublicMonitorResult::EndpointFailed,
Err(_) => PublicMonitorResult::EndpointFailed,
};
}
changed = endpoint.status.changed() => {
if changed.is_err() {
return PublicMonitorResult::EndpointFailed;
}
let endpoint_status = endpoint.status.borrow().clone();
if !endpoint_status.relay_ready || endpoint_status.endpoint_ticket.is_none() {
if awaiting_relay {
continue;
}
return PublicMonitorResult::RelayLost;
}
if publish_iroh_status(
shared,
pairing,
frontend_origin,
endpoint_status,
false,
)
.is_err()
{
return PublicMonitorResult::EndpointFailed;
}
awaiting_relay = false;
}
}
}
}
fn publish_iroh_status(
shared: &RuntimeShared,
pairing: &UiPairingService,
frontend_origin: &str,
endpoint: IrohEndpointStatus,
initial: bool,
) -> AgentResult<()> {
let mut status = shared.status();
status.transport = RuntimeTransport::Iroh;
status.iroh_endpoint_id = Some(endpoint.endpoint_id.clone());
if endpoint.relay_ready
&& let Some(ticket) = endpoint.endpoint_ticket
{
apply_iroh_pairing(
&mut status,
pairing,
frontend_origin,
&endpoint.endpoint_id,
&ticket,
)?;
status.phase = RuntimePhase::Ready;
status.message = Some("Iroh endpoint ready".to_owned());
shared.publish_status(status);
shared.publish_event(RuntimeEvent::Ready);
return Ok(());
}
status.iroh_endpoint_ticket = None;
status.frontend_deep_link = None;
status.pairing_code = None;
status.pairing_expires_at = None;
if initial {
status.phase = RuntimePhase::Starting;
status.message = Some("Iroh endpoint bound; waiting for relay".to_owned());
shared.publish_status(status);
} else {
publish_degraded_status(shared, status, "Iroh relay is unavailable");
}
Ok(())
}
fn apply_iroh_pairing(
status: &mut RuntimeStatus,
pairing: &UiPairingService,
frontend_origin: &str,
endpoint: &str,
ticket: &str,
) -> AgentResult<()> {
pairing.invalidate_code();
let issued = pairing
.issue_code_with_expiry()
.map_err(|_| pairing_failed())?;
let code = issued.code().expose_for_display().to_owned();
let link = FrontendDeepLink::iroh(
frontend_origin,
endpoint,
ticket,
pairing.server_id().as_str(),
&code,
)?;
status.iroh_endpoint_ticket = Some(ticket.to_owned());
status.frontend_deep_link = Some(link.as_str().to_owned());
status.pairing_code = Some(code);
status.pairing_expires_at = Some(issued.expires_at().to_rfc3339());
Ok(())
}
fn publish_degraded(shared: &RuntimeShared, pairing: &UiPairingService, message: &str) {
pairing.invalidate_code();
let status = shared.status();
publish_degraded_status(shared, status, message);
}
fn publish_degraded_status(shared: &RuntimeShared, mut status: RuntimeStatus, message: &str) {
status.phase = RuntimePhase::Degraded;
status.iroh_endpoint_ticket = None;
status.frontend_deep_link = None;
status.pairing_code = None;
status.pairing_expires_at = None;
status.message = Some(message.to_owned());
shared.publish_status(status);
shared.publish_event(RuntimeEvent::Degraded);
}
fn publish_stopping(shared: &RuntimeShared, pairing: &UiPairingService) {
pairing.invalidate_code();
let mut status = shared.status();
status.phase = RuntimePhase::Stopping;
status.iroh_endpoint_ticket = None;
status.frontend_deep_link = None;
status.pairing_code = None;
status.pairing_expires_at = None;
status.message = Some("runtime stopping".to_owned());
shared.publish_status(status);
shared.publish_event(RuntimeEvent::Stopping);
}
fn publish_stopped(shared: &RuntimeShared) {
let mut status = shared.status();
status.phase = RuntimePhase::Stopped;
status.iroh_endpoint_ticket = None;
status.frontend_deep_link = None;
status.pairing_code = None;
status.pairing_expires_at = None;
status.message = Some("runtime stopped".to_owned());
shared.publish_status(status);
shared.publish_event(RuntimeEvent::Stopped);
}
fn public_retry_delay(attempt: u32, jitter: u8) -> Duration {
let base = Duration::from_secs(1_u64 << attempt.min(5));
let jitter = base.mul_f64(f64::from(jitter) / (f64::from(u8::MAX) * 4.0));
base.saturating_add(jitter).min(MAX_PUBLIC_RETRY_DELAY)
}
#[cfg(test)]
pub(crate) fn public_retry_delay_for_test(attempt: u32, jitter: u8) -> Duration {
public_retry_delay(attempt, jitter)
}
fn random_jitter() -> u8 {
Uuid::new_v4().as_bytes()[0]
}
fn validate_options(options: &RuntimeOptions) -> AgentResult<()> {
validate_exact_origin(&options.frontend_origin)?;
if options.frontend_origin.starts_with("https://") {
Ok(())
} else {
Err(invalid_runtime_configuration())
}
}
fn validate_exact_origin(origin: &str) -> AgentResult<()> {
let parsed = Url::parse(origin).map_err(|_| invalid_runtime_configuration())?;
if parsed.origin().ascii_serialization() != origin
|| !parsed.username().is_empty()
|| parsed.password().is_some()
|| parsed.query().is_some()
|| parsed.fragment().is_some()
|| parsed.path() != "/"
|| parsed.scheme() != "https"
{
return Err(invalid_runtime_configuration());
}
Ok(())
}
fn flatten_task(result: Result<AgentResult<()>, tokio::task::JoinError>) -> AgentResult<()> {
result.map_err(|_| runtime_task_failed())?
}
fn invalid_runtime_configuration() -> AgentError {
AgentError::new(
ErrorCode::InvalidMessage,
"invalid integrated runtime configuration",
)
}
fn pairing_failed() -> AgentError {
AgentError::new(
ErrorCode::PairingStorageFailed,
"pairing code could not be issued",
)
}
fn runtime_task_failed() -> AgentError {
AgentError::new(
ErrorCode::BackendDisconnected,
"integrated runtime task failed",
)
}