Update Rust crate sqlx to 0.9.0

Cargo.toml

@@ -4,9 +4,10 @@ version = "0.1.0"
 edition = "2024"
 
 [dependencies]
-mtp = { git = "https://git.methanium.net/methanium/mtp", features = [
+mtp = { git = "https://git.methanium.net/Methanium/mtp.git",  features = [
     "crypto",
     "files",
+    "raw",
     "web-server",
 ] }
 
@@ -25,7 +26,7 @@ rustls = { version = "0.23.42", default-features = false, features = [
     "aws-lc-rs",
     "prefer-post-quantum",
 ] }
-sqlx = { version = "0.8.6", features = ["mysql", "runtime-tokio", "migrate"] }
+sqlx = { version = "0.9.0", features = ["mysql", "runtime-tokio", "migrate"] }
 strum = "0.28.0"
 strum_macros = "0.28.0"
 tokio = { version = "*", features = ["full"] }

src/db/user_repo.rs

@@ -380,27 +380,44 @@ pub async fn delete_user_with_pending_erasure(id: UserId) -> Result<Option<IotaI
         .ok_or(OmegaError::NotFound)?;
     let iota_id: Option<i64> = row.get("iota_id");
     if let Some(iota_id) = iota_id {
-        sqlx::query("INSERT IGNORE INTO pending_iota_user_erasure (user_id, iota_id) VALUES (?, ?)")
+        sqlx::query(
-            .bind(id.0)
+            "INSERT IGNORE INTO pending_iota_user_erasure (user_id, iota_id) VALUES (?, ?)",
-            .bind(iota_id)
+        )
-            .execute(&mut *tx)
+        .bind(id.0)
-            .await?;
+        .bind(iota_id)
+        .execute(&mut *tx)
+        .await?;
     }
-    sqlx::query("DELETE FROM registration_leases WHERE user_id = ?").bind(id.0).execute(&mut *tx).await?;
+    sqlx::query("DELETE FROM registration_leases WHERE user_id = ?")
-    sqlx::query("DELETE FROM users WHERE id = ?").bind(id.0).execute(&mut *tx).await?;
+        .bind(id.0)
+        .execute(&mut *tx)
+        .await?;
+    sqlx::query("DELETE FROM users WHERE id = ?")
+        .bind(id.0)
+        .execute(&mut *tx)
+        .await?;
     tx.commit().await?;
     Ok(iota_id.map(IotaId::from))
 }
 
 pub async fn pending_erasures_for_iota(iota_id: IotaId) -> Result<Vec<UserId>> {
     let rows = sqlx::query("SELECT user_id FROM pending_iota_user_erasure WHERE iota_id = ?")
-        .bind(iota_id.0).fetch_all(&pool().await?).await?;
+        .bind(iota_id.0)
-    Ok(rows.into_iter().map(|row| UserId::from(row.get::<i64, _>("user_id"))).collect())
+        .fetch_all(&pool().await?)
+        .await?;
+    Ok(rows
+        .into_iter()
+        .map(|row| UserId::from(row.get::<i64, _>("user_id")))
+        .collect())
 }
 
 pub async fn acknowledge_pending_erasure(user_id: UserId, iota_id: IotaId) -> Result<bool> {
-    let result = sqlx::query("DELETE FROM pending_iota_user_erasure WHERE user_id = ? AND iota_id = ?")
+    let result =
-        .bind(user_id.0).bind(iota_id.0).execute(&pool().await?).await?;
+        sqlx::query("DELETE FROM pending_iota_user_erasure WHERE user_id = ? AND iota_id = ?")
+            .bind(user_id.0)
+            .bind(iota_id.0)
+            .execute(&pool().await?)
+            .await?;
     Ok(result.rows_affected() == 1)
 }
 

src/server/api.rs

@@ -85,11 +85,13 @@ async fn route(path_parts: &[&str]) -> Result<(StatusCode, String)> {
                 let user = get_by_user_id(UserId::from(id)).await?;
                 match user.iota_id {
                     Some(iota_id) => get_iota_primary_omikron_connection(iota_id.0),
-                    None => get_random_omikron()
+                    None => {
-                        .await
+                        get_random_omikron()
-                        .map_err(|_| OmegaError::NotFound)?
+                            .await
-                        .get_omikron_id()
+                            .map_err(|_| OmegaError::NotFound)?
-                        .await,
+                            .get_omikron_id()
+                            .await
+                    }
                 }
                 .ok_or(OmegaError::NotFound)?
             };

src/sql/user_online_tracker.rs

@@ -1044,20 +1044,24 @@ mod tests {
         assert_eq!(tracker.subscribers(22).len(), 1);
         tracker.remove_session(7, 3, 42);
         assert!(tracker.check_index_consistency().is_ok());
-        assert!(tracker
+        assert!(
-            .routes
+            tracker
-            .read()
+                .routes
-            .unwrap()
+                .read()
-            .indices
+                .unwrap()
-            .sessions_by_user
+                .indices
-            .is_empty());
+                .sessions_by_user
-        assert!(tracker
+                .is_empty()
-            .routes
+        );
-            .read()
+        assert!(
-            .unwrap()
+            tracker
-            .indices
+                .routes
-            .targets_by_subscriber
+                .read()
-            .is_empty());
+                .unwrap()
+                .indices
+                .targets_by_subscriber
+                .is_empty()
+        );
     }
 
     #[test]

src/state.rs

@@ -1,10 +1,13 @@
 use crate::sql::user_online_tracker::PresenceTracker;
-use std::sync::Arc;
 use dashmap::DashMap;
+use std::sync::Arc;
 use std::time::{Duration, Instant};
 
 #[derive(Clone, Copy, Debug, PartialEq, Eq, Hash)]
-pub enum AccountChallengeOperation { Attach, Delete }
+pub enum AccountChallengeOperation {
+    Attach,
+    Delete,
+}
 
 #[derive(Clone, Debug)]
 pub struct AccountChallenge {
@@ -30,15 +33,38 @@ impl Default for OmegaState {
 }
 
 impl OmegaState {
-    pub fn issue_challenge(&self, operation: AccountChallengeOperation, user_id: i64, requester_iota_id: i64) -> u64 {
+    pub fn issue_challenge(
+        &self,
+        operation: AccountChallengeOperation,
+        user_id: i64,
+        requester_iota_id: i64,
+    ) -> u64 {
         let nonce = rand::random::<u64>();
-        self.challenges.insert((operation, user_id, requester_iota_id), AccountChallenge { operation, user_id, requester_iota_id, nonce, created_at: Instant::now() });
+        self.challenges.insert(
+            (operation, user_id, requester_iota_id),
+            AccountChallenge {
+                operation,
+                user_id,
+                requester_iota_id,
+                nonce,
+                created_at: Instant::now(),
+            },
+        );
         nonce
     }
 
-    pub fn consume_challenge(&self, operation: AccountChallengeOperation, user_id: i64, requester_iota_id: i64, nonce: u64) -> bool {
+    pub fn consume_challenge(
-        self.challenges.remove(&(operation, user_id, requester_iota_id)).is_some_and(|(_, value)|
+        &self,
-            value.nonce == nonce && value.created_at.elapsed() <= Duration::from_secs(120))
+        operation: AccountChallengeOperation,
+        user_id: i64,
+        requester_iota_id: i64,
+        nonce: u64,
+    ) -> bool {
+        self.challenges
+            .remove(&(operation, user_id, requester_iota_id))
+            .is_some_and(|(_, value)| {
+                value.nonce == nonce && value.created_at.elapsed() <= Duration::from_secs(120)
+            })
     }
 }
 

src/transport/connection.rs

@@ -1 +1,48 @@
-pub(crate) use super::omikron_connection::OmikronConnection;
+pub(crate) use super::omikron_connection::{OmikronConnection, OmikronResult};
+
+use mtp::codec::{CommunicationValue, DataValue};
+
+pub(crate) trait MtpValueCompat {
+    fn get_id(&self) -> u32;
+    fn get_sender(&self) -> u64;
+    fn get_receiver(&self) -> u64;
+}
+
+impl MtpValueCompat for CommunicationValue {
+    fn get_id(&self) -> u32 {
+        self.id().unwrap_or_default()
+    }
+
+    fn get_sender(&self) -> u64 {
+        self.sender().unwrap_or_default()
+    }
+
+    fn get_receiver(&self) -> u64 {
+        self.receiver().unwrap_or_default()
+    }
+}
+
+pub(crate) trait OptionalDataValueCompat {
+    fn as_number(&self) -> Option<i128>;
+    fn as_signed_number(&self) -> Option<i128>;
+    fn as_str(&self) -> Option<&str>;
+    fn as_bytes(&self) -> Option<Vec<u8>>;
+}
+
+impl OptionalDataValueCompat for Option<&DataValue> {
+    fn as_number(&self) -> Option<i128> {
+        self.and_then(|value| value.as_number())
+    }
+
+    fn as_signed_number(&self) -> Option<i128> {
+        self.and_then(|value| value.as_signed_number())
+    }
+
+    fn as_str(&self) -> Option<&str> {
+        self.and_then(|value| value.as_str())
+    }
+
+    fn as_bytes(&self) -> Option<Vec<u8>> {
+        self.and_then(|value| value.as_bytes())
+    }
+}

src/transport/handlers/account.rs

@@ -1,10 +1,15 @@
-use super::super::omikron_connection::{OmikronConnection, OmikronResult};
+use super::super::connection::{
+    MtpValueCompat, OmikronConnection, OmikronResult, OptionalDataValueCompat,
+};
 use crate::{
     db::{iota_repo, user_repo},
     models::{IotaId, UserId},
     state::AccountChallengeOperation,
 };
-use mtp::{codec::{CommunicationType, CommunicationValue, DataType, DataValue}, crypto::{verify_ed25519, verify_ml_dsa}};
+use mtp::{
+    codec::{CommunicationType, CommunicationValue, DataType, DataValue},
+    crypto::{verify_ed25519, verify_ml_dsa},
+};
 use std::sync::Arc;
 
 async fn delete(
@@ -67,14 +72,22 @@ pub async fn release_from_iota(
     let previous_iota = user.iota_id;
     match user_repo::change_iota_id(user.id, None).await {
         Ok(()) => {
-            if let Some(iota) = previous_iota { crate::transport::omikron_manager::publish_iota_user_snapshot(iota.0).await; }
+            if let Some(iota) = previous_iota {
-            connection.send(&CommunicationValue::new(CommunicationType::Success).with_id(value.get_id())).await
+                crate::transport::omikron_manager::publish_iota_user_snapshot(iota.0).await;
-        },
+            }
-        Err(error) => connection
+            connection
-            .send(&CommunicationValue::new(CommunicationType::ErrorInternal)
+                .send(&CommunicationValue::new(CommunicationType::Success).with_id(value.get_id()))
-                .with_id(value.get_id())
+                .await
-                .add_typed_default(DataType::ErrorType, DataValue::Str(error.to_string())))
+        }
-            .await,
+        Err(error) => {
+            connection
+                .send(
+                    &CommunicationValue::new(CommunicationType::ErrorInternal)
+                        .with_id(value.get_id())
+                        .add_typed_default(DataType::ErrorType, DataValue::Str(error.to_string())),
+                )
+                .await
+        }
     }
 }
 
@@ -87,43 +100,121 @@ fn lifecycle_payload(domain: &[u8], user_id: i64, iota_id: i64, nonce: u64) -> V
     payload
 }
 
-pub async fn attach_begin(connection: Arc<OmikronConnection>, value: CommunicationValue) -> OmikronResult<()> {
+pub async fn attach_begin(
-    let Some(user_id) = value.get_data(DataType::UserId).as_signed_number().and_then(|v| i64::try_from(v).ok()).filter(|v| *v > 0) else {
+    connection: Arc<OmikronConnection>,
-        return connection.send_error_response(value.get_id(), CommunicationType::ErrorInvalidUserId).await;
+    value: CommunicationValue,
+) -> OmikronResult<()> {
+    let Some(user_id) = value
+        .get_data(DataType::UserId)
+        .as_signed_number()
+        .and_then(|v| i64::try_from(v).ok())
+        .filter(|v| *v > 0)
+    else {
+        return connection
+            .send_error_response(value.get_id(), CommunicationType::ErrorInvalidUserId)
+            .await;
     };
-    if user_repo::get_by_user_id(UserId::from(user_id)).await.is_err() {
+    if user_repo::get_by_user_id(UserId::from(user_id))
-        return connection.send_error_response(value.get_id(), CommunicationType::ErrorNotFound).await;
+        .await
+        .is_err()
+    {
+        return connection
+            .send_error_response(value.get_id(), CommunicationType::ErrorNotFound)
+            .await;
     }
     let requester = value.get_sender() as i64;
-    let nonce = connection.state().issue_challenge(AccountChallengeOperation::Attach, user_id, requester);
+    let nonce =
-    connection.send(&CommunicationValue::new(CommunicationType::AttachUserChallenge).with_id(value.get_id())
+        connection
-        .add_typed_default(DataType::UserId, DataValue::SignedNumber(user_id.into()))
+            .state()
-        .add_typed_default(DataType::ServerNonce, DataValue::SignedNumber(nonce.into()))).await
+            .issue_challenge(AccountChallengeOperation::Attach, user_id, requester);
+    connection
+        .send(
+            &CommunicationValue::new(CommunicationType::AttachUserChallenge)
+                .with_id(value.get_id())
+                .add_typed_default(DataType::UserId, DataValue::SignedNumber(user_id.into()))
+                .add_typed_default(DataType::ServerNonce, DataValue::SignedNumber(nonce.into())),
+        )
+        .await
 }
 
-pub async fn attach_complete(connection: Arc<OmikronConnection>, value: CommunicationValue) -> OmikronResult<()> {
+pub async fn attach_complete(
-    let Some(user_id) = value.get_data(DataType::UserId).as_signed_number().and_then(|v| i64::try_from(v).ok()).filter(|v| *v > 0) else { return connection.send_error_response(value.get_id(), CommunicationType::ErrorInvalidUserId).await; };
+    connection: Arc<OmikronConnection>,
+    value: CommunicationValue,
+) -> OmikronResult<()> {
+    let Some(user_id) = value
+        .get_data(DataType::UserId)
+        .as_signed_number()
+        .and_then(|v| i64::try_from(v).ok())
+        .filter(|v| *v > 0)
+    else {
+        return connection
+            .send_error_response(value.get_id(), CommunicationType::ErrorInvalidUserId)
+            .await;
+    };
     let requester = value.get_sender() as i64;
-    let Some(nonce) = value.get_data(DataType::ServerNonce).as_signed_number().and_then(|v| u64::try_from(v).ok()) else { return connection.send_error_response(value.get_id(), CommunicationType::ErrorInvalidChallenge).await; };
+    let Some(nonce) = value
+        .get_data(DataType::ServerNonce)
+        .as_signed_number()
+        .and_then(|v| u64::try_from(v).ok())
+    else {
+        return connection
+            .send_error_response(value.get_id(), CommunicationType::ErrorInvalidChallenge)
+            .await;
+    };
     let signature = value.get_data(DataType::Signature).as_bytes();
     let pq_signature = value.get_data(DataType::PqSignature).as_bytes();
-    let (Some(signature), Some(pq_signature)) = (signature, pq_signature) else { return connection.send_error_response(value.get_id(), CommunicationType::ErrorInvalidChallenge).await; };
+    let (Some(signature), Some(pq_signature)) = (signature, pq_signature) else {
-    if !connection.state().consume_challenge(AccountChallengeOperation::Attach, user_id, requester, nonce) { return connection.send_error_response(value.get_id(), CommunicationType::ErrorInvalidChallenge).await; }
+        return connection
-    let Ok(user) = user_repo::get_by_user_id(UserId::from(user_id)).await else { return connection.send_error_response(value.get_id(), CommunicationType::ErrorNotFound).await; };
+            .send_error_response(value.get_id(), CommunicationType::ErrorInvalidChallenge)
+            .await;
+    };
+    if !connection.state().consume_challenge(
+        AccountChallengeOperation::Attach,
+        user_id,
+        requester,
+        nonce,
+    ) {
+        return connection
+            .send_error_response(value.get_id(), CommunicationType::ErrorInvalidChallenge)
+            .await;
+    }
+    let Ok(user) = user_repo::get_by_user_id(UserId::from(user_id)).await else {
+        return connection
+            .send_error_response(value.get_id(), CommunicationType::ErrorNotFound)
+            .await;
+    };
     let payload = lifecycle_payload(b"tensamin:user-attach:v1\0", user_id, requester, nonce);
-    if verify_ed25519(&user.public_key.sig_cl_public_key, &payload, &signature).is_err() || verify_ml_dsa(&user.public_key.sig_pq_public_key, &payload, &pq_signature).is_err() { return connection.send_error_response(value.get_id(), CommunicationType::ErrorNotAuthenticated).await; }
+    if verify_ed25519(&user.public_key.sig_cl_public_key, &payload, &signature).is_err()
+        || verify_ml_dsa(&user.public_key.sig_pq_public_key, &payload, &pq_signature).is_err()
+    {
+        return connection
+            .send_error_response(value.get_id(), CommunicationType::ErrorNotAuthenticated)
+            .await;
+    }
     let previous_iota = user.iota_id;
     match user_repo::change_iota_id(user.id, Some(IotaId::from(requester))).await {
         Ok(()) => {
-            if let Some(iota) = previous_iota.filter(|id| id.0 != requester) { crate::transport::omikron_manager::publish_iota_user_snapshot(iota.0).await; }
+            if let Some(iota) = previous_iota.filter(|id| id.0 != requester) {
+                crate::transport::omikron_manager::publish_iota_user_snapshot(iota.0).await;
+            }
             crate::transport::omikron_manager::publish_iota_user_snapshot(requester).await;
-            connection.send(&CommunicationValue::new(CommunicationType::Success).with_id(value.get_id())).await
+            connection
-        },
+                .send(&CommunicationValue::new(CommunicationType::Success).with_id(value.get_id()))
-        Err(_) => connection.send_error_response(value.get_id(), CommunicationType::ErrorInternal).await,
+                .await
+        }
+        Err(_) => {
+            connection
+                .send_error_response(value.get_id(), CommunicationType::ErrorInternal)
+                .await
+        }
     }
 }
 
-async fn complete_delete(connection: Arc<OmikronConnection>, value: CommunicationValue, user_id: UserId) -> OmikronResult<()> {
+async fn complete_delete(
+    connection: Arc<OmikronConnection>,
+    value: CommunicationValue,
+    user_id: UserId,
+) -> OmikronResult<()> {
     match user_repo::delete_user_with_pending_erasure(user_id).await {
         Ok(iota_id) => {
             let cleanup_pending = iota_id.is_some();
@@ -131,45 +222,160 @@ async fn complete_delete(connection: Arc<OmikronConnection>, value: Communicatio
                 crate::transport::omikron_manager::publish_iota_user_snapshot(iota_id.0).await;
                 crate::transport::omikron_manager::deliver_pending_erasures(iota_id.0).await;
             }
-            connection.send(&CommunicationValue::new(CommunicationType::Success).with_id(value.get_id())
+            connection
-                .add_typed_default(DataType::CleanupPending, DataValue::Bool(cleanup_pending))).await
+                .send(
+                    &CommunicationValue::new(CommunicationType::Success)
+                        .with_id(value.get_id())
+                        .add_typed_default(
+                            DataType::CleanupPending,
+                            DataValue::Bool(cleanup_pending),
+                        ),
+                )
+                .await
+        }
+        Err(crate::error::OmegaError::NotFound) => {
+            connection
+                .send_error_response(value.get_id(), CommunicationType::ErrorNotFound)
+                .await
+        }
+        Err(error) => {
+            connection
+                .send(
+                    &CommunicationValue::new(CommunicationType::ErrorInternal)
+                        .with_id(value.get_id())
+                        .add_typed_default(DataType::ErrorType, DataValue::Str(error.to_string())),
+                )
+                .await
         }
-        Err(crate::error::OmegaError::NotFound) => connection.send_error_response(value.get_id(), CommunicationType::ErrorNotFound).await,
-        Err(error) => connection.send(&CommunicationValue::new(CommunicationType::ErrorInternal).with_id(value.get_id()).add_typed_default(DataType::ErrorType, DataValue::Str(error.to_string()))).await,
     }
 }
 
-pub async fn delete_credential_begin(connection: Arc<OmikronConnection>, value: CommunicationValue) -> OmikronResult<()> {
+pub async fn delete_credential_begin(
-    let Some(user_id) = value.get_data(DataType::UserId).as_signed_number().and_then(|v| i64::try_from(v).ok()).filter(|v| *v > 0) else {
+    connection: Arc<OmikronConnection>,
-        return connection.send_error_response(value.get_id(), CommunicationType::ErrorInvalidUserId).await;
+    value: CommunicationValue,
+) -> OmikronResult<()> {
+    let Some(user_id) = value
+        .get_data(DataType::UserId)
+        .as_signed_number()
+        .and_then(|v| i64::try_from(v).ok())
+        .filter(|v| *v > 0)
+    else {
+        return connection
+            .send_error_response(value.get_id(), CommunicationType::ErrorInvalidUserId)
+            .await;
     };
-    if user_repo::get_by_user_id(UserId::from(user_id)).await.is_err() { return connection.send_error_response(value.get_id(), CommunicationType::ErrorNotFound).await; }
+    if user_repo::get_by_user_id(UserId::from(user_id))
+        .await
+        .is_err()
+    {
+        return connection
+            .send_error_response(value.get_id(), CommunicationType::ErrorNotFound)
+            .await;
+    }
     let requester = value.get_sender() as i64;
-    let nonce = connection.state().issue_challenge(AccountChallengeOperation::Delete, user_id, requester);
+    let nonce =
-    connection.send(&CommunicationValue::new(CommunicationType::DeleteUserCredentialChallenge).with_id(value.get_id())
+        connection
-        .add_typed_default(DataType::UserId, DataValue::SignedNumber(user_id.into()))
+            .state()
-        .add_typed_default(DataType::ServerNonce, DataValue::SignedNumber(nonce.into()))).await
+            .issue_challenge(AccountChallengeOperation::Delete, user_id, requester);
+    connection
+        .send(
+            &CommunicationValue::new(CommunicationType::DeleteUserCredentialChallenge)
+                .with_id(value.get_id())
+                .add_typed_default(DataType::UserId, DataValue::SignedNumber(user_id.into()))
+                .add_typed_default(DataType::ServerNonce, DataValue::SignedNumber(nonce.into())),
+        )
+        .await
 }
 
-pub async fn delete_credential_complete(connection: Arc<OmikronConnection>, value: CommunicationValue) -> OmikronResult<()> {
+pub async fn delete_credential_complete(
-    let Some(user_id) = value.get_data(DataType::UserId).as_signed_number().and_then(|v| i64::try_from(v).ok()).filter(|v| *v > 0) else { return connection.send_error_response(value.get_id(), CommunicationType::ErrorInvalidUserId).await; };
+    connection: Arc<OmikronConnection>,
+    value: CommunicationValue,
+) -> OmikronResult<()> {
+    let Some(user_id) = value
+        .get_data(DataType::UserId)
+        .as_signed_number()
+        .and_then(|v| i64::try_from(v).ok())
+        .filter(|v| *v > 0)
+    else {
+        return connection
+            .send_error_response(value.get_id(), CommunicationType::ErrorInvalidUserId)
+            .await;
+    };
     let requester = value.get_sender() as i64;
-    let Some(nonce) = value.get_data(DataType::ServerNonce).as_signed_number().and_then(|v| u64::try_from(v).ok()) else { return connection.send_error_response(value.get_id(), CommunicationType::ErrorInvalidChallenge).await; };
+    let Some(nonce) = value
-    let (Some(signature), Some(pq_signature)) = (value.get_data(DataType::Signature).as_bytes(), value.get_data(DataType::PqSignature).as_bytes()) else { return connection.send_error_response(value.get_id(), CommunicationType::ErrorInvalidChallenge).await; };
+        .get_data(DataType::ServerNonce)
-    if !connection.state().consume_challenge(AccountChallengeOperation::Delete, user_id, requester, nonce) { return connection.send_error_response(value.get_id(), CommunicationType::ErrorInvalidChallenge).await; }
+        .as_signed_number()
-    let Ok(user) = user_repo::get_by_user_id(UserId::from(user_id)).await else { return connection.send_error_response(value.get_id(), CommunicationType::ErrorNotFound).await; };
+        .and_then(|v| u64::try_from(v).ok())
+    else {
+        return connection
+            .send_error_response(value.get_id(), CommunicationType::ErrorInvalidChallenge)
+            .await;
+    };
+    let (Some(signature), Some(pq_signature)) = (
+        value.get_data(DataType::Signature).as_bytes(),
+        value.get_data(DataType::PqSignature).as_bytes(),
+    ) else {
+        return connection
+            .send_error_response(value.get_id(), CommunicationType::ErrorInvalidChallenge)
+            .await;
+    };
+    if !connection.state().consume_challenge(
+        AccountChallengeOperation::Delete,
+        user_id,
+        requester,
+        nonce,
+    ) {
+        return connection
+            .send_error_response(value.get_id(), CommunicationType::ErrorInvalidChallenge)
+            .await;
+    }
+    let Ok(user) = user_repo::get_by_user_id(UserId::from(user_id)).await else {
+        return connection
+            .send_error_response(value.get_id(), CommunicationType::ErrorNotFound)
+            .await;
+    };
     let payload = lifecycle_payload(b"tensamin:user-delete:v1\0", user_id, requester, nonce);
-    if verify_ed25519(&user.public_key.sig_cl_public_key, &payload, &signature).is_err() || verify_ml_dsa(&user.public_key.sig_pq_public_key, &payload, &pq_signature).is_err() { return connection.send_error_response(value.get_id(), CommunicationType::ErrorNotAuthenticated).await; }
+    if verify_ed25519(&user.public_key.sig_cl_public_key, &payload, &signature).is_err()
+        || verify_ml_dsa(&user.public_key.sig_pq_public_key, &payload, &pq_signature).is_err()
+    {
+        return connection
+            .send_error_response(value.get_id(), CommunicationType::ErrorNotAuthenticated)
+            .await;
+    }
     complete_delete(connection, value, user.id).await
 }
 
-pub async fn erase_hosted_user_data_ack(connection: Arc<OmikronConnection>, value: CommunicationValue) -> OmikronResult<()> {
+pub async fn erase_hosted_user_data_ack(
-    let Some(user_id) = value.get_data(DataType::UserId).as_signed_number().and_then(|v| i64::try_from(v).ok()).filter(|v| *v > 0) else { return connection.send_error_response(value.get_id(), CommunicationType::ErrorInvalidUserId).await; };
+    connection: Arc<OmikronConnection>,
+    value: CommunicationValue,
+) -> OmikronResult<()> {
+    let Some(user_id) = value
+        .get_data(DataType::UserId)
+        .as_signed_number()
+        .and_then(|v| i64::try_from(v).ok())
+        .filter(|v| *v > 0)
+    else {
+        return connection
+            .send_error_response(value.get_id(), CommunicationType::ErrorInvalidUserId)
+            .await;
+    };
     let iota_id = IotaId::from(value.get_sender() as i64);
     match user_repo::acknowledge_pending_erasure(UserId::from(user_id), iota_id).await {
-        Ok(true) => connection.send(&CommunicationValue::new(CommunicationType::Success).with_id(value.get_id())).await,
+        Ok(true) => {
-        Ok(false) => connection.send_error_response(value.get_id(), CommunicationType::ErrorNotAuthenticated).await,
+            connection
-        Err(_) => connection.send_error_response(value.get_id(), CommunicationType::ErrorInternal).await,
+                .send(&CommunicationValue::new(CommunicationType::Success).with_id(value.get_id()))
+                .await
+        }
+        Ok(false) => {
+            connection
+                .send_error_response(value.get_id(), CommunicationType::ErrorNotAuthenticated)
+                .await
+        }
+        Err(_) => {
+            connection
+                .send_error_response(value.get_id(), CommunicationType::ErrorInternal)
+                .await
+        }
     }
 }
 

src/transport/handlers/links.rs

@@ -1,4 +1,6 @@
-use super::super::omikron_connection::{OmikronConnection, OmikronResult};
+use super::super::connection::{
+    MtpValueCompat, OmikronConnection, OmikronResult, OptionalDataValueCompat,
+};
 use crate::server::short_link::add_short_link;
 use mtp::codec::{CommunicationType, CommunicationValue, DataType, DataValue};
 use std::sync::Arc;
@@ -7,8 +9,8 @@ pub async fn shorten(
     connection: Arc<OmikronConnection>,
     value: CommunicationValue,
 ) -> OmikronResult<()> {
-    let link = value
+    let link_data = value.get_data(DataType::Link);
-        .get_data(DataType::Link)
+    let link = link_data
         .as_str()
         .ok_or(crate::error::OmegaError::InvalidResponse)?;
     let short = add_short_link(link)

src/transport/handlers/notifications.rs

@@ -1,4 +1,6 @@
-use super::super::omikron_connection::{OmikronConnection, OmikronResult};
+use super::super::connection::{
+    MtpValueCompat, OmikronConnection, OmikronResult, OptionalDataValueCompat,
+};
 use crate::{db::notification_repo, log, models::UserId};
 use mtp::codec::{CommunicationType, CommunicationValue, DataType, DataValue};
 use mtp::type_map::TypeMap;

src/transport/handlers/presence.rs

@@ -1,4 +1,6 @@
-use super::super::omikron_connection::{OmikronConnection, OmikronResult};
+use super::super::connection::{
+    MtpValueCompat, OmikronConnection, OmikronResult, OptionalDataValueCompat,
+};
 use crate::{
     db::user_repo, log_in, models::IotaId, sql::connection_status::UserStatus, state::OmegaState,
 };
@@ -19,7 +21,7 @@ fn parse_subscription(value: &CommunicationValue) -> Result<(i64, i64, Vec<i64>)
         .and_then(|id| i64::try_from(id).ok())
         .filter(|id| *id > 0)
         .ok_or("session_id")?;
-    let DataValue::Array(values) = value.get_data(DataType::UserIds) else {
+    let Some(DataValue::Array(values)) = value.get_data(DataType::UserIds) else {
         return Err("user_ids");
     };
 
@@ -51,9 +53,10 @@ fn states_for_users(state: &OmegaState, users: &[crate::models::User]) -> HashMa
         .map(|user| {
             (
                 user.id.0,
-                state
+                state.presence.resolve_public_state(
-                    .presence
+                    user.id.0,
-                    .resolve_public_state(user.id.0, user.iota_id.map(|id| id.0).unwrap_or_default()),
+                    user.iota_id.map(|id| id.0).unwrap_or_default(),
+                ),
             )
         })
         .collect()
@@ -349,7 +352,7 @@ pub async fn set_user_state(
             .send_error_response(value.get_id(), CommunicationType::ErrorNoUserId)
             .await;
     };
-    if let Some(requested_user) = value.get_data_opt(DataType::UserId) {
+    if let Some(requested_user) = value.get_data(DataType::UserId) {
         let Some(requested_user_id) = requested_user
             .as_number()
             .and_then(|id| i64::try_from(id).ok())
@@ -538,12 +541,12 @@ pub async fn sync_status(
     omikron_id: i64,
 ) -> OmikronResult<()> {
     let request_id = value.get_id();
-    let DataValue::Array(iota_values) = value.get_data(DataType::IotaIds) else {
+    let Some(DataValue::Array(iota_values)) = value.get_data(DataType::IotaIds) else {
         return connection
             .send_error_response(request_id, CommunicationType::ErrorInvalidData)
             .await;
     };
-    let DataValue::Array(session_values) = value.get_data(DataType::UserStates) else {
+    let Some(DataValue::Array(session_values)) = value.get_data(DataType::UserStates) else {
         return connection
             .send_error_response(request_id, CommunicationType::ErrorInvalidData)
             .await;
@@ -572,7 +575,7 @@ pub async fn sync_status(
     }
 
     if !connection.peer_capabilities().session_snapshot_v1 {
-        let DataValue::Array(user_values) = value.get_data(DataType::UserIds) else {
+        let Some(DataValue::Array(user_values)) = value.get_data(DataType::UserIds) else {
             return connection
                 .send_error_response(request_id, CommunicationType::ErrorInvalidData)
                 .await;

src/transport/handlers/register.rs

@@ -1,4 +1,6 @@
-use super::super::omikron_connection::{OmikronConnection, OmikronResult};
+use super::super::connection::{
+    MtpValueCompat, OmikronConnection, OmikronResult, OptionalDataValueCompat,
+};
 use crate::{
     db::{iota_repo, user_repo},
     models::{IotaId, UserId},

src/transport/handlers/states.rs

@@ -1,4 +1,6 @@
-use super::super::omikron_connection::{OmikronConnection, OmikronResult};
+use super::super::connection::{
+    MtpValueCompat, OmikronConnection, OmikronResult, OptionalDataValueCompat,
+};
 use crate::db::user_repo;
 use mtp::{
     codec::{CommunicationType, CommunicationValue, DataType, DataValue},
@@ -29,7 +31,7 @@ pub async fn get(
 ) -> OmikronResult<()> {
     let state = connection.state();
     let legacy_peer = !connection.peer_capabilities().client_state_push_v1;
-    let DataValue::Array(ids) = value.get_data(DataType::UserIds) else {
+    let Some(DataValue::Array(ids)) = value.get_data(DataType::UserIds) else {
         return send_error(
             connection,
             value.get_id(),

src/transport/handlers/user_data.rs

@@ -1,4 +1,6 @@
-use super::super::omikron_connection::{OmikronConnection, OmikronResult};
+use super::super::connection::{
+    MtpValueCompat, OmikronConnection, OmikronResult, OptionalDataValueCompat,
+};
 use crate::{
     db::{iota_repo, user_repo},
     models::{IotaId, UserId},
@@ -91,7 +93,9 @@ pub async fn get_user(
         }
         state.presence.resolve_private_state(id)
     } else {
-        iota_id.map(|iota_id| state.presence.resolve_public_state(id, iota_id)).unwrap_or(crate::sql::connection_status::UserStatus::user_offline)
+        iota_id
+            .map(|iota_id| state.presence.resolve_public_state(id, iota_id))
+            .unwrap_or(crate::sql::connection_status::UserStatus::user_offline)
     };
     response = response
         .add_typed_default(
@@ -100,10 +104,13 @@ pub async fn get_user(
         )
         .add_typed_default(
             DataType::OmikronConnections,
-            iota_id.map(|iota_id| connections(&connection, iota_id)).unwrap_or_else(|| DataValue::Array(Vec::new())),
+            iota_id
+                .map(|iota_id| connections(&connection, iota_id))
+                .unwrap_or_else(|| DataValue::Array(Vec::new())),
         );
     if let Some(iota_id) = iota_id {
-        response = response.add_typed_default(DataType::IotaId, DataValue::SignedNumber(iota_id.into()));
+        response =
+            response.add_typed_default(DataType::IotaId, DataValue::SignedNumber(iota_id.into()));
     }
     if let Some(route) = route {
         response = response.add_typed_default(
@@ -138,10 +145,14 @@ pub async fn get_iota(
     } else if let Some(name) = value.get_data(DataType::Username).as_str() {
         if let Ok(user) = user_repo::get_by_username(name).await {
             match user.iota_id {
-                Some(iota_id) => iota_repo::get_iota_by_id(iota_id)
+                Some(iota_id) => iota_repo::get_iota_by_id(iota_id).await.ok().map(|iota| {
-                    .await
+                    (
-                    .ok()
+                        iota.id.0,
-                    .map(|iota| (iota.id.0, iota.public_key, Some(user.id.0), Some(name.to_owned()))),
+                        iota.public_key,
+                        Some(user.id.0),
+                        Some(name.to_owned()),
+                    )
+                }),
                 None => None,
             }
         } else {
@@ -245,12 +256,14 @@ pub async fn change_iota(
     connection: Arc<OmikronConnection>,
     value: CommunicationValue,
 ) -> OmikronResult<()> {
-    let Some(reset) = value.get_data(DataType::ResetToken).as_str() else {
+    let reset_data = value.get_data(DataType::ResetToken);
+    let Some(reset) = reset_data.as_str() else {
         return connection
             .send_error_response(value.get_id(), CommunicationType::ErrorInvalidData)
             .await;
     };
-    let Some(new_token) = value.get_data(DataType::NewToken).as_str() else {
+    let new_token_data = value.get_data(DataType::NewToken);
+    let Some(new_token) = new_token_data.as_str() else {
         return connection
             .send_error_response(value.get_id(), CommunicationType::ErrorInvalidData)
             .await;
@@ -270,7 +283,9 @@ pub async fn change_iota(
             .await;
     }
     let result =
-        match user_repo::change_iota_id(user_id, Some(IotaId::from(value.get_sender() as i64))).await {
+        match user_repo::change_iota_id(user_id, Some(IotaId::from(value.get_sender() as i64)))
+            .await
+        {
             Ok(()) => user_repo::change_token(user_id, new_token.to_owned()).await,
             Err(error) => Err(error),
         };

src/transport/mod.rs

@@ -3,3 +3,4 @@ pub mod connection;
 pub mod handlers;
 pub mod omikron_connection;
 pub mod omikron_manager;
+pub mod relay_router;

src/transport/omikron_connection.rs

@@ -10,7 +10,7 @@ use dashmap::DashMap;
 use mtp::{
     codec::{CommunicationType, CommunicationValue},
     crypto::PublicKeyBundle,
-    host::{AuthenticationPolicy, HostConfig, Policy, SendMode},
+    host::{AuthState, AuthenticationPolicy, HostConfig, Policy, SendMode},
     webserver::{MTPWebServer, WebMtpReceiver, WebMtpSender},
 };
 use std::{
@@ -21,7 +21,10 @@ use std::{
     },
     time::{Duration, Instant},
 };
-use tokio::{sync::Mutex, time::interval};
+use tokio::{
+    sync::{Mutex, mpsc},
+    time::interval,
+};
 
 const CLEANUP_INTERVAL: Duration = Duration::from_secs(30);
 const MAX_WAITING_AGE: Duration = Duration::from_secs(60);
@@ -70,7 +73,11 @@ impl OmikronConnection {
         id: u64,
         description: Option<&str>,
         state: Arc<OmegaState>,
+        authenticated: bool,
     ) -> Option<Arc<Self>> {
+        if !authenticated {
+            return None;
+        }
         let peer_capabilities =
             PeerCapabilities::from_identification_description(description).ok()?;
         Some(Arc::new(Self {
@@ -87,7 +94,6 @@ impl OmikronConnection {
         &self.peer_capabilities
     }
 
-
     pub async fn handle(self: Arc<Self>, receiver: &mut WebMtpReceiver) {
         log_in!(
             self.id as i64,
@@ -158,7 +164,12 @@ impl OmikronConnection {
 
     async fn process_message(self: Arc<Self>, value: CommunicationValue) -> OmikronResult<()> {
         log_cv_in!(PrintType::Omikron, &value);
-        if let Some((_, task)) = self.waiting_tasks.remove(&value.get_id()) {
+        if value.is_type(CommunicationType::Relay) {
+            return self.dispatch(value).await;
+        }
+        if let Some(message_id) = value.id()
+            && let Some((_, task)) = self.waiting_tasks.remove(&message_id)
+        {
             let _ = (task.task)(self.clone(), value);
             return Ok(());
         }
@@ -168,6 +179,36 @@ impl OmikronConnection {
     async fn dispatch(self: Arc<Self>, value: CommunicationValue) -> OmikronResult<()> {
         let id = self.id as i64;
         let state = self.state.clone();
+        if value.is_type(CommunicationType::Relay) {
+            let value = crate::transport::relay_router::ensure_relay_frame_id(value);
+            let request_id = value.id();
+            let result = crate::transport::relay_router::route_from_omikron(id, value).await;
+            let response = match &result {
+                Ok(()) => request_id.map(|request_id| {
+                    CommunicationValue::new(CommunicationType::Success).with_id(request_id)
+                }),
+                Err(error) => {
+                    log_err!(id, PrintType::Omega, "Relay routing failed: {}", error);
+                    request_id.map(|request_id| {
+                        CommunicationValue::new(
+                            crate::transport::relay_router::error_response_type(error),
+                        )
+                        .with_id(request_id)
+                    })
+                }
+            };
+            if let Some(response) = response
+                && let Err(send_error) = self.clone().send(&response).await
+            {
+                log_err!(
+                    id,
+                    PrintType::Omega,
+                    "Relay routing response failed: {}",
+                    send_error
+                );
+            }
+            return result.map_err(|error| crate::error::OmegaError::Transport(error.to_string()));
+        }
         match value.get_comm_type_enum() {
             Some(CommunicationType::ShortenLink) => {
                 crate::transport::handlers::links::shorten(self, value).await
@@ -179,9 +220,6 @@ impl OmikronConnection {
                 crate::transport::handlers::presence::user_disconnected(state, self, value, id)
                     .await
             }
-            Some(CommunicationType::SetUserState) => {
-                crate::transport::handlers::presence::set_user_state(state, self, value, id).await
-            }
             Some(CommunicationType::IotaConnected) => {
                 crate::transport::handlers::presence::iota_connected(state, self, value, id).await
             }
@@ -297,6 +335,47 @@ impl OmikronConnection {
         }
         Ok(())
     }
+
+    pub(crate) async fn await_response(
+        self: Arc<Self>,
+        value: &CommunicationValue,
+        timeout: Duration,
+    ) -> OmikronResult<CommunicationValue> {
+        let (tx, mut rx) = mpsc::channel(1);
+        let message_id = value.id().unwrap_or_default();
+        let task_tx = tx.clone();
+        self.waiting_tasks.insert(
+            message_id,
+            WaitingTask {
+                task: Box::new(move |_, response| {
+                    let task_tx = task_tx.clone();
+                    tokio::spawn(async move {
+                        let _ = task_tx.send(response).await;
+                    });
+                    true
+                }),
+                inserted_at: Instant::now(),
+            },
+        );
+
+        if let Err(error) = self.clone().send(value).await {
+            self.waiting_tasks.remove(&message_id);
+            return Err(error);
+        }
+
+        match tokio::time::timeout(timeout, rx.recv()).await {
+            Ok(Some(response)) => Ok(response),
+            Ok(None) => Err(crate::error::OmegaError::Transport(
+                "Relay response channel closed".into(),
+            )),
+            Err(_) => {
+                self.waiting_tasks.remove(&message_id);
+                Err(crate::error::OmegaError::Transport(
+                    "Relay response timed out".into(),
+                ))
+            }
+        }
+    }
     pub(crate) async fn send_error_response(
         self: Arc<Self>,
         message_id: u32,
@@ -439,16 +518,18 @@ pub async fn start(port: u16, state: Arc<OmegaState>) -> Result<(), Box<dyn std:
             );
             continue;
         }
+        let authenticated = matches!(&conn.auth_state, AuthState::Authenticated);
         let Some(connection) = OmikronConnection::new(
             conn.sender,
             conn.client_id,
             conn.description.as_deref(),
             state.clone(),
+            authenticated,
         ) else {
             log_err!(
                 0,
                 PrintType::Omega,
-                "Rejected Omikron connection with invalid capabilities"
+                "Rejected Omikron connection without authenticated transport state or valid capabilities"
             );
             continue;
         };

src/transport/omikron_manager.rs

@@ -66,7 +66,8 @@ pub async fn get_all_connections()
                 for route in state.presence.routes_for_user(user.id.0) {
                     if let Some(iotas) = result.get_mut(&route.omikron_id) {
                         if let Some(iota_id) = user.iota_id
-                            && let Some(users) = iotas.get_mut(&iota_id.0) {
+                            && let Some(users) = iotas.get_mut(&iota_id.0)
+                        {
                             users.push(user.id.0);
                         }
                     }
@@ -118,10 +119,20 @@ pub async fn send_to_user(user_id: i64, cv: &CommunicationValue) {
 /// Publish the authoritative membership list after an attach, migration, or
 /// release. Omikron replaces its full local index from this snapshot.
 pub async fn publish_iota_user_snapshot(iota_id: i64) {
-    let Some(omikron_id) = get_iota_primary_omikron_connection(iota_id) else { return; };
+    let Some(omikron_id) = get_iota_primary_omikron_connection(iota_id) else {
-    let Some(connection) = get_connected_omikron(omikron_id) else { return; };
+        return;
-    let Ok(users) = user_repo::get_users_by_iota_id(crate::models::IotaId::from(iota_id)).await else { return; };
+    };
-    let user_ids = users.into_iter().map(|user| DataValue::SignedNumber(user.id.0.into())).collect();
+    let Some(connection) = get_connected_omikron(omikron_id) else {
+        return;
+    };
+    let Ok(users) = user_repo::get_users_by_iota_id(crate::models::IotaId::from(iota_id)).await
+    else {
+        return;
+    };
+    let user_ids = users
+        .into_iter()
+        .map(|user| DataValue::SignedNumber(user.id.0.into()))
+        .collect();
     let snapshot = CommunicationValue::new(CommunicationType::IotaUserData)
         .add_typed_default(DataType::IotaId, DataValue::SignedNumber(iota_id.into()))
         .add_typed_default(DataType::UserIds, DataValue::Array(user_ids));
@@ -129,9 +140,17 @@ pub async fn publish_iota_user_snapshot(iota_id: i64) {
 }
 
 pub async fn deliver_pending_erasures(iota_id: i64) {
-    let Ok(users) = user_repo::pending_erasures_for_iota(crate::models::IotaId::from(iota_id)).await else { return; };
+    let Ok(users) =
-    let Some(omikron_id) = get_iota_primary_omikron_connection(iota_id) else { return; };
+        user_repo::pending_erasures_for_iota(crate::models::IotaId::from(iota_id)).await
-    let Some(connection) = get_connected_omikron(omikron_id) else { return; };
+    else {
+        return;
+    };
+    let Some(omikron_id) = get_iota_primary_omikron_connection(iota_id) else {
+        return;
+    };
+    let Some(connection) = get_connected_omikron(omikron_id) else {
+        return;
+    };
     for user_id in users {
         let request = CommunicationValue::new(CommunicationType::EraseHostedUserData)
             .add_typed_default(DataType::UserId, DataValue::SignedNumber(user_id.0.into()))

src/transport/relay_router.rs

@@ -0,0 +1,243 @@
+use super::omikron_manager;
+use crate::{log_err, util::logger::PrintType};
+use mtp::codec::{CommunicationType, CommunicationValue, RelayError, forward_relay_frame};
+use std::{
+    convert::TryFrom,
+    sync::atomic::{AtomicU32, Ordering},
+    time::Duration,
+};
+use thiserror::Error;
+
+const TARGET_KIND_MASK: u64 = 0xC000_0000_0000_0000;
+const TARGET_ID_MASK: u64 = (1_u64 << 48) - 1;
+const USER_TARGET_KIND: u64 = 0x4000_0000_0000_0000;
+const IOTA_TARGET_KIND: u64 = 0x8000_0000_0000_0000;
+static NEXT_RELAY_FRAME_ID: AtomicU32 = AtomicU32::new(1);
+
+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+pub enum RouteTarget {
+    User(u64),
+    Iota(u64),
+}
+
+impl RouteTarget {
+    pub fn wire_id(self) -> Option<u64> {
+        let (kind, id) = match self {
+            Self::User(id) => (USER_TARGET_KIND, id),
+            Self::Iota(id) => (IOTA_TARGET_KIND, id),
+        };
+        (id > 0 && id <= TARGET_ID_MASK).then_some(kind | id)
+    }
+
+    pub fn from_wire_id(value: u64) -> Option<Self> {
+        let id = value & TARGET_ID_MASK;
+        if id == 0 || value & !(TARGET_KIND_MASK | TARGET_ID_MASK) != 0 {
+            return None;
+        }
+        match value & TARGET_KIND_MASK {
+            USER_TARGET_KIND => Some(Self::User(id)),
+            IOTA_TARGET_KIND => Some(Self::Iota(id)),
+            _ => None,
+        }
+    }
+
+    pub const fn id(self) -> u64 {
+        match self {
+            Self::User(id) | Self::Iota(id) => id,
+        }
+    }
+}
+
+#[derive(Debug, Error)]
+pub enum RelayRouteError {
+    #[error("relay has no destination Iota")]
+    MissingDestinationIota,
+    #[error("relay has an outer sender")]
+    OuterSenderPresent,
+    #[error("relay has invalid route target {0}")]
+    InvalidDestinationTarget(u64),
+    #[error("relay destination Iota is outside Omega's ID range")]
+    DestinationIotaOutOfRange,
+    #[error("destination Iota is offline")]
+    IotaOffline,
+    #[error("destination Omikron is offline")]
+    OmikronOffline,
+    #[error("relay route resolves back to source Omikron")]
+    RouteLoop,
+    #[error(transparent)]
+    Relay(#[from] RelayError),
+    #[error("sending relay to destination Omikron failed: {0}")]
+    Send(String),
+}
+
+pub fn ensure_relay_frame_id(frame: CommunicationValue) -> CommunicationValue {
+    if frame.id().is_some_and(|id| id != 0) {
+        return frame;
+    }
+    let id = NEXT_RELAY_FRAME_ID.fetch_add(1, Ordering::Relaxed).max(1);
+    frame.with_id(id)
+}
+
+pub fn error_response_type(error: &RelayRouteError) -> CommunicationType {
+    match error {
+        RelayRouteError::IotaOffline | RelayRouteError::OmikronOffline => {
+            CommunicationType::ErrorNoIota
+        }
+        RelayRouteError::Send(_) => CommunicationType::ErrorInternal,
+        RelayRouteError::MissingDestinationIota
+        | RelayRouteError::OuterSenderPresent
+        | RelayRouteError::InvalidDestinationTarget(_)
+        | RelayRouteError::DestinationIotaOutOfRange
+        | RelayRouteError::RouteLoop
+        | RelayRouteError::Relay(_) => CommunicationType::ErrorInvalidData,
+    }
+}
+
+pub async fn route_from_omikron(
+    source_omikron_id: i64,
+    frame: CommunicationValue,
+) -> Result<(), RelayRouteError> {
+    let frame = ensure_relay_frame_id(frame);
+    if !frame.is_type(CommunicationType::Relay) {
+        return Err(RelayRouteError::Relay(RelayError::NotRelay));
+    }
+    if frame.sender().is_some() {
+        return Err(RelayRouteError::OuterSenderPresent);
+    }
+    let Some(destination_wire_id) = frame.receiver() else {
+        log_err!(
+            source_omikron_id,
+            PrintType::Omega,
+            "Relay routing failed: missing destination Iota"
+        );
+        return Err(RelayRouteError::MissingDestinationIota);
+    };
+    let Some(RouteTarget::Iota(destination_iota)) = RouteTarget::from_wire_id(destination_wire_id)
+    else {
+        return Err(RelayRouteError::InvalidDestinationTarget(
+            destination_wire_id,
+        ));
+    };
+    let destination_iota_i64 =
+        i64::try_from(destination_iota).map_err(|_| RelayRouteError::DestinationIotaOutOfRange)?;
+    let frame = forward_relay_frame(&frame, destination_wire_id)?;
+
+    let Some(destination_omikron) =
+        omikron_manager::get_iota_primary_omikron_connection(destination_iota_i64)
+    else {
+        log_err!(
+            source_omikron_id,
+            PrintType::Omega,
+            "Relay destination Iota {} is offline",
+            destination_iota
+        );
+        return Err(RelayRouteError::IotaOffline);
+    };
+    if destination_omikron == source_omikron_id {
+        log_err!(
+            source_omikron_id,
+            PrintType::Omega,
+            "Relay route loop for destination Iota {} and Omikron {}",
+            destination_iota,
+            destination_omikron
+        );
+        return Err(RelayRouteError::RouteLoop);
+    }
+    let Some(connection) = omikron_manager::get_connected_omikron(destination_omikron) else {
+        log_err!(
+            source_omikron_id,
+            PrintType::Omega,
+            "Relay destination Iota {} resolves to disconnected Omikron {}",
+            destination_iota,
+            destination_omikron
+        );
+        return Err(RelayRouteError::OmikronOffline);
+    };
+    let response = connection
+        .await_response(&frame, Duration::from_secs(20))
+        .await
+        .map_err(|error| {
+            log_err!(
+                source_omikron_id,
+                PrintType::Omega,
+                "Relay send to destination Iota {} via Omikron {} failed: {}",
+                destination_iota,
+                destination_omikron,
+                error
+            );
+            RelayRouteError::Send(error.to_string())
+        })?;
+    if response.is_type(CommunicationType::Success) {
+        Ok(())
+    } else {
+        Err(RelayRouteError::Send(format!(
+            "destination Omikron rejected the Relay with {}",
+            response.get_type()
+        )))
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use mtp::codec::DataValue;
+
+    fn wire(target: RouteTarget) -> u64 {
+        let Some(value) = target.wire_id() else {
+            panic!("valid route target was rejected");
+        };
+        value
+    }
+
+    fn relay_frame() -> CommunicationValue {
+        CommunicationValue::new(CommunicationType::Relay)
+            .without_sender()
+            .with_receiver(wire(RouteTarget::Iota(42)))
+            .with_payload(DataValue::Bytes(vec![1, 2, 3, 4]))
+    }
+
+    #[test]
+    fn forwarding_preserves_relay_payload_and_next_hop() {
+        let frame = relay_frame().with_receiver(wire(RouteTarget::Iota(7)));
+        let result = forward_relay_frame(&frame, wire(RouteTarget::Iota(42)));
+        assert!(result.is_ok());
+        let Ok(forwarded) = result else { return };
+        assert_eq!(forwarded.receiver(), Some(wire(RouteTarget::Iota(42))));
+        assert_eq!(forwarded.sender(), None);
+        assert_eq!(forwarded.payload(), frame.payload());
+        assert_eq!(forwarded.id(), frame.id());
+    }
+
+    #[tokio::test]
+    async fn outer_sender_is_rejected_before_route_lookup() {
+        let frame = relay_frame().with_sender(9);
+        let result = route_from_omikron(1, frame).await;
+        assert!(matches!(result, Err(RelayRouteError::OuterSenderPresent)));
+    }
+
+    #[tokio::test]
+    async fn missing_destination_is_rejected_before_route_lookup() {
+        let frame = relay_frame().without_receiver();
+        let result = route_from_omikron(1, frame).await;
+        assert!(matches!(
+            result,
+            Err(RelayRouteError::MissingDestinationIota)
+        ));
+    }
+
+    #[tokio::test]
+    async fn user_route_target_is_rejected_by_opaque_omega_router() {
+        let frame = relay_frame().with_receiver(wire(RouteTarget::User(42)));
+        let result = route_from_omikron(1, frame).await;
+        assert!(matches!(
+            result,
+            Err(RelayRouteError::InvalidDestinationTarget(_))
+        ));
+    }
+
+    #[tokio::test]
+    async fn unavailable_destination_is_reported_as_iota_offline() {
+        let result = route_from_omikron(1, relay_frame()).await;
+        assert!(matches!(result, Err(RelayRouteError::IotaOffline)));
+    }
+}

src/util/logger.rs

@@ -10,6 +10,7 @@ use std::{
 use ansi_term::Color;
 use mtp::codec::{CommunicationValue, DataTypeId, DataValue, Version};
 
+use crate::transport::connection::MtpValueCompat;
 use crate::util::file_util::get_directory;
 
 static LOGGER: OnceLock<mpsc::Sender<LogMessage>> = OnceLock::new();
@@ -250,11 +251,11 @@ pub fn format_cv(cv: &CommunicationValue) -> String {
         .unwrap_or_else(|| cv.get_type().to_string());
     parts.push(format!("{} (id={})", comm_type, cv.get_id()));
 
-    let data = cv.data();
+    let data = cv.data().unwrap_or(&[]);
 
     let formated_data = format_data_container(
         data.iter().map(|(k, v)| (*k, v.clone())).collect(),
-        Version(1, 0),
+        Version(3, 0),
     );
 
     parts.push(format!("{}", formated_data));