Update Rust crate base64 to 0.23.0

Cargo.lock

@@ -5017,7 +5017,10 @@ name = "web-ui"
 version = "0.1.0"
 dependencies = [
  "actix-web",
+ "iota-cli",
+ "iota-ipc",
  "iota-logger",
+ "iota-paths",
  "iota-state",
  "iota-storage",
  "iota-util",

client/src/client_connection.rs

@@ -280,12 +280,6 @@ impl ClientConnection {
             return;
         }
 
-        if cv.is_type(CommunicationType::AddConversation) {
-            self.send_message(&message_handlers::handle_add_conversation(&cv))
-                .await;
-            return;
-        }
-
         if cv.is_type(CommunicationType::AddCommunity) {
             self.send_message(&message_handlers::handle_add_community(&cv))
                 .await;
@@ -416,7 +410,9 @@ impl ClientConnection {
             return;
         };
 
-        let Some(encrypted_challenge) = cv.get_data(DataType::Challenge).as_str() else { return };
+        let Some(encrypted_challenge) = cv.get_data(DataType::Challenge).as_str() else {
+            return;
+        };
 
         let solved = crypto_util::decrypt_challenge(encrypted_challenge, &keyring).ok();
 

iota-connection/src/message_handlers.rs

@@ -10,6 +10,7 @@ use mtp::codec::{
 };
 
 use crate::relay::VerifiedRelayContext;
+use iota_storage::storage_error::StorageError;
 
 #[derive(Debug)]
 pub struct MessageMutation {
@@ -61,6 +62,20 @@ pub fn success_response(cv: &CommunicationValue) -> CommunicationValue {
     error_response(cv, CommunicationType::Success)
 }
 
+fn add_conversation_for_user(
+    user_id: i64,
+    other_id: i64,
+    name: Option<&str>,
+) -> Result<(), StorageError> {
+    let mut contact = get_user(user_id, other_id)?
+        .unwrap_or_else(|| iota_storage::users::contact::Contact::new(other_id));
+    if let Some(name) = name {
+        contact.user_name = Some(name.to_string());
+    }
+    contact.set_last_message_at(now_millis_i64());
+    mod_user(user_id, &contact)
+}
+
 fn relay_field<'a>(
     payload: &'a DataValue,
     data_type: DataType,
@@ -142,14 +157,11 @@ pub fn apply_verified_relay_content(
 
     match content.message_type.as_str() {
         "MessageState" => {
-            let partner_id = relay_number(
-                &content.content,
-                DataType::ChatPartnerId,
-                &context.type_map,
-            )
-            .and_then(|value| i64::try_from(value).ok())
-            .filter(|id| *id == recipient_id)
-            .ok_or_else(|| "Relay MessageState has an invalid ChatPartnerId".to_string())?;
+            let partner_id =
+                relay_number(&content.content, DataType::ChatPartnerId, &context.type_map)
+                    .and_then(|value| i64::try_from(value).ok())
+                    .filter(|id| *id == recipient_id)
+                    .ok_or_else(|| "Relay MessageState has an invalid ChatPartnerId".to_string())?;
             let relay_message_id = relay_string(
                 &content.content,
                 DataType::RelayMessageId,
@@ -159,14 +171,10 @@ pub fn apply_verified_relay_content(
             let event_at = relay_number(&content.content, DataType::EventAt, &context.type_map)
                 .and_then(|value| i64::try_from(value).ok())
                 .ok_or_else(|| "Relay MessageState is missing EventAt".to_string())?;
-            let state = relay_string(
-                &content.content,
-                DataType::MessageState,
-                &context.type_map,
-            )
-            .map(MessageState::from_str)
-            .filter(|state| matches!(state, MessageState::Received | MessageState::Read))
-            .ok_or_else(|| "Relay MessageState has an invalid state".to_string())?;
+            let state = relay_string(&content.content, DataType::MessageState, &context.type_map)
+                .map(MessageState::from_str)
+                .filter(|state| matches!(state, MessageState::Received | MessageState::Read))
+                .ok_or_else(|| "Relay MessageState has an invalid state".to_string())?;
             chat_files::record_message_receipt(
                 storage_owner,
                 recipient_id,
@@ -192,22 +200,28 @@ pub fn apply_verified_relay_content(
                 .unwrap_or_default();
             let reply_to = relay_number(&content.content, DataType::ReplyId, &context.type_map)
                 .and_then(|value| i64::try_from(value).ok());
-            let relay_message_id = relay_string(
+            if relay_string(
                 &content.content,
                 DataType::RelayMessageId,
                 &context.type_map,
             )
-            .ok_or_else(|| "Relay MessageSend is missing RelayMessageId".to_string())?;
-            if relay_message_id != context.message_id {
-                return Err("Relay MessageSend identity does not match its protected message ID".into());
+            .is_some_and(|relay_message_id| relay_message_id != context.message_id)
+            {
+                return Err(
+                    "Relay MessageSend identity does not match its protected message ID".into(),
+                );
             }
             chat_files::add_message(chat_files::NewMessage {
                 relay_signer_id: sender_id,
-                relay_message_id,
+                relay_message_id: &context.message_id,
                 authored_at: created_at,
                 send_time,
                 storage_owner,
-                external_user: if sent_by_self { recipient_id } else { sender_id },
+                external_user: if sent_by_self {
+                    recipient_id
+                } else {
+                    sender_id
+                },
                 sent_by_self,
                 content: message,
                 height,
@@ -256,7 +270,7 @@ pub fn apply_verified_relay_content(
                 .ok_or_else(|| "Relay SetChatSecret has no recipients".to_string())?;
             let recipient = recipients
                 .into_iter()
-                .find(|value| value.user_id == context.final_recipient_id.to_string())
+                .find(|value| value.user_id == storage_owner.to_string())
                 .ok_or_else(|| "Relay SetChatSecret recipient mismatch".to_string())?;
             let chat_id = data_string(&frame, DataType::ChatId)
                 .ok_or_else(|| "Relay SetChatSecret is missing ChatId".to_string())?;
@@ -267,7 +281,7 @@ pub fn apply_verified_relay_content(
             let wrapping_scheme = data_string(&frame, DataType::WrappingScheme)
                 .ok_or_else(|| "Relay SetChatSecret is missing WrappingScheme".to_string())?;
             e2ee_storage::put_chat_secret(e2ee_storage::StoredChatSecret {
-                user_id: context.final_recipient_id.to_string(),
+                user_id: storage_owner.to_string(),
                 chat_id,
                 secret_id,
                 version,
@@ -279,6 +293,29 @@ pub fn apply_verified_relay_content(
             })
             .map_err(|error| error.to_string())
         }
+        "AddConversation" => {
+            let other_id =
+                relay_number(&content.content, DataType::ChatPartnerId, &context.type_map)
+                    .and_then(|value| i64::try_from(value).ok())
+                    .filter(|id| *id > 0)
+                    .ok_or_else(|| {
+                        "Relay AddConversation has an invalid ChatPartnerId".to_string()
+                    })?;
+            let user_id = storage_owner;
+            if user_id <= 0 {
+                return Err("Relay AddConversation has an invalid storage owner".into());
+            }
+            add_conversation_for_user(
+                user_id,
+                other_id,
+                relay_string(
+                    &content.content,
+                    DataType::ChatPartnerName,
+                    &context.type_map,
+                ),
+            )
+            .map_err(|error| format!("AddConversation persistence failed: {error}"))
+        }
         _ => Ok(()),
     }
 }
@@ -369,10 +406,7 @@ fn stored_message_fields(
             DataType::SendTime,
             DataValue::SignedNumber(message.message_time as i128),
         ),
-        (
-            DataType::Content,
-            DataValue::Str(message.content.clone()),
-        ),
+        (DataType::Content, DataValue::Str(message.content.clone())),
         (
             DataType::MessageState,
             DataValue::Str(message.message_state.clone()),
@@ -398,10 +432,19 @@ fn stored_message_fields(
     }
     for (data_type, timestamp) in [
         (DataType::AuthoredAt, message.authored_at),
-        (DataType::OriginIotaReceivedAt, message.origin_iota_received_at),
-        (DataType::DestinationIotaReceivedAt, message.destination_iota_received_at),
+        (
+            DataType::OriginIotaReceivedAt,
+            message.origin_iota_received_at,
+        ),
+        (
+            DataType::DestinationIotaReceivedAt,
+            message.destination_iota_received_at,
+        ),
         (DataType::ClientReceivedAt, message.client_received_at),
-        (DataType::ClientReceivedRecordedAt, message.client_received_recorded_at),
+        (
+            DataType::ClientReceivedRecordedAt,
+            message.client_received_recorded_at,
+        ),
         (DataType::ReadAt, message.read_at),
         (DataType::ReadRecordedAt, message.read_recorded_at),
     ] {
@@ -543,7 +586,11 @@ pub fn handle_create_app(cv: &CommunicationValue) -> CommunicationValue {
         .to_string();
 
     if !app_identifier.is_empty() && !app_public_key.is_empty() {
-        if let Some(mut user) = iota_storage::users::user_manager::get_user(sender_id) {
+        let user = match iota_storage::users::user_manager::get_user(sender_id) {
+            Ok(user) => user,
+            Err(_) => return error_response(cv, CommunicationType::ErrorInternal),
+        };
+        if let Some(mut user) = user {
             if !user.trusted_apps.contains_key(&app_identifier) {
                 user.trusted_apps.insert(app_identifier, app_public_key);
                 iota_storage::users::user_manager::update_user(user);
@@ -568,7 +615,11 @@ pub fn handle_delete_app(cv: &CommunicationValue) -> CommunicationValue {
         .to_string();
 
     if !app_identifier.is_empty() {
-        if let Some(mut user) = iota_storage::users::user_manager::get_user(sender_id) {
+        let user = match iota_storage::users::user_manager::get_user(sender_id) {
+            Ok(user) => user,
+            Err(_) => return error_response(cv, CommunicationType::ErrorInternal),
+        };
+        if let Some(mut user) = user {
             if user.trusted_apps.contains_key(&app_identifier) {
                 user.trusted_apps.remove(&app_identifier);
                 iota_storage::users::user_manager::update_user(user);
@@ -612,12 +663,12 @@ fn contact_value(
     typed_container(fields)
 }
 
-fn current_contact_ids(user_id: i64) -> DataValue {
-    contact_ids_value(
-        chats_util::get_users(user_id)
+fn current_contact_ids(user_id: i64) -> Result<DataValue, StorageError> {
+    Ok(contact_ids_value(
+        chats_util::get_users(user_id)?
             .into_iter()
             .map(|contact| contact.user_id),
-    )
+    ))
 }
 
 fn contact_ids_value(ids: impl IntoIterator<Item = i64>) -> DataValue {
@@ -714,7 +765,10 @@ pub fn handle_client_connected(cv: &CommunicationValue) -> CommunicationValue {
                 Err(_) => return sync_error(cv),
             };
             (
-                chats_util::get_users(user_id),
+                match chats_util::get_users(user_id) {
+                    Ok(contacts) => contacts,
+                    Err(_) => return sync_error(cv),
+                },
                 chat_files::get_all_messages(user_id),
                 settings,
                 Vec::new(),
@@ -731,7 +785,10 @@ pub fn handle_client_connected(cv: &CommunicationValue) -> CommunicationValue {
                             Err(_) => return sync_error(cv),
                         };
                     (
-                        chats_util::get_users_by_ids(user_id, &delta.contact_upserts),
+                        match chats_util::get_users_by_ids(user_id, &delta.contact_upserts) {
+                            Ok(contacts) => contacts,
+                            Err(_) => return sync_error(cv),
+                        },
                         chat_files::get_messages_by_ids(user_id, &delta.message_upserts),
                         settings,
                         delta.deleted_message_ids,
@@ -746,7 +803,10 @@ pub fn handle_client_connected(cv: &CommunicationValue) -> CommunicationValue {
                         Err(_) => return sync_error(cv),
                     };
                     (
-                        chats_util::get_users(user_id),
+                        match chats_util::get_users(user_id) {
+                            Ok(contacts) => contacts,
+                            Err(_) => return sync_error(cv),
+                        },
                         chat_files::get_all_messages(user_id),
                         settings,
                         Vec::new(),
@@ -761,6 +821,10 @@ pub fn handle_client_connected(cv: &CommunicationValue) -> CommunicationValue {
         .iter()
         .map(|message| stored_message_value(message, user_id, message.external_user))
         .collect();
+    let contact_ids = match current_contact_ids(user_id) {
+        Ok(contact_ids) => contact_ids,
+        Err(_) => return error_response(cv, CommunicationType::ErrorInternal),
+    };
     CommunicationValue::new(CommunicationType::ClientStateSync)
         .with_request_id(cv)
         .with_receiver(sender_wire_id(user_id))
@@ -822,7 +886,7 @@ pub fn handle_client_connected(cv: &CommunicationValue) -> CommunicationValue {
                     .collect(),
             ),
         )
-        .add_typed_default(DataType::UserIds, current_contact_ids(user_id))
+        .add_typed_default(DataType::UserIds, contact_ids)
         .add_typed_default(DataType::Calls, DataValue::Array(Vec::new()))
 }
 
@@ -948,7 +1012,10 @@ pub fn handle_get_chats(cv: &CommunicationValue) -> CommunicationValue {
     let Ok(user_id_i64) = i64::try_from(user_id) else {
         return error_response(cv, CommunicationType::ErrorInvalidData);
     };
-    let users = chats_util::get_users(user_id_i64);
+    let users = match chats_util::get_users(user_id_i64) {
+        Ok(users) => users,
+        Err(_) => return error_response(cv, CommunicationType::ErrorInternal),
+    };
     let mut user_array = Vec::new();
     for user in users {
         let mut container = Vec::new();
@@ -970,41 +1037,6 @@ pub fn handle_get_chats(cv: &CommunicationValue) -> CommunicationValue {
         .add_typed_default(DataType::UserIds, DataValue::Array(user_array))
 }
 
-pub fn handle_add_conversation(cv: &CommunicationValue) -> CommunicationValue {
-    let user_id = match cv.require_sender() {
-        Ok(user_id) => user_id,
-        Err(_) => return error_response(cv, CommunicationType::ErrorInvalidData),
-    };
-    let Ok(user_id_i64) = i64::try_from(user_id) else {
-        return error_response(cv, CommunicationType::ErrorInvalidData);
-    };
-    let session_id = match data_i64(cv, DataType::SessionId) {
-        Some(id) if id > 0 => id,
-        _ => return sync_error(cv),
-    };
-    let other_id = match data_i64(cv, DataType::ChatPartnerId) {
-        Some(id) if id > 0 => id,
-        _ => return error_response(cv, CommunicationType::ErrorInvalidData),
-    };
-    let mut contact = get_user(user_id_i64, other_id)
-        .unwrap_or(iota_storage::users::contact::Contact::new(other_id));
-
-    if let Some(name) = cv.get_data(DataType::ChatPartnerName).as_str() {
-        contact.user_name = Some(name.to_string());
-    }
-
-    contact.set_last_message_at(now_millis_i64());
-    mod_user(user_id_i64, &contact);
-    CommunicationValue::new(CommunicationType::AddConversation)
-        .with_request_id(cv)
-        .with_receiver(user_id)
-        .add_typed_default(
-            DataType::SessionId,
-            DataValue::SignedNumber(session_id as i128),
-        )
-        .add_typed_default(DataType::UserIds, current_contact_ids(user_id_i64))
-}
-
 pub fn handle_add_community(cv: &CommunicationValue) -> CommunicationValue {
     let sender_id = match required_sender_id(cv) {
         Ok(sender_id) => sender_id,

iota-daemon-lib/src/command_router.rs

@@ -7,11 +7,14 @@ use iota_ipc::{
     TaskSummary, UpdateStatusResponse, UserDetailResponse, UserSummary,
 };
 use iota_logger::{log, log_command};
+use iota_storage::users::pending_operations::{
+    self, PendingUserOperation, PendingUserOperationKind, PendingUserOperationPhase,
+};
 use iota_storage::users::user_manager;
 use iota_storage::util::config_util::{self};
 use mtp::codec::{CommunicationType, CommunicationValue, DataType, DataValue};
 use std::sync::{Arc, Mutex};
-use std::time::Duration;
+use std::time::{Duration, SystemTime, UNIX_EPOCH};
 
 use crate::daemon_state::{ShutdownReason, StartupPhase};
 
@@ -26,6 +29,13 @@ pub struct PeerContext {
 
 const MAX_LOG_ENTRIES_PER_RESPONSE: usize = 512;
 
+fn now_millis() -> i64 {
+    SystemTime::now()
+        .duration_since(UNIX_EPOCH)
+        .unwrap_or_default()
+        .as_millis() as i64
+}
+
 fn bounded_log_entries(mut entries: Vec<LogEntry>) -> Vec<LogEntry> {
     entries.truncate(MAX_LOG_ENTRIES_PER_RESPONSE);
     while !entries.is_empty() {
@@ -152,32 +162,38 @@ impl CommandRouter {
                 ResponseResult::Ok(ResponsePayload::Tasks(tasks))
             }
             LocalRequest::ListUsers => {
-                let users: Vec<UserSummary> = user_manager::get_residency()
+                let users = user_manager::get_residency()
                     .into_iter()
-                    .map(|user| UserSummary {
-                        credential_present: user.state == user_manager::LocalUserState::Managed
-                            && user_manager::get_user(user.user_id).is_some_and(|profile| {
-                                iota_util::file_util::read_user_credential_with_legacy(
-                                    user.user_id,
-                                    &profile.username,
-                                )
-                                .ok()
-                                .flatten()
-                                .is_some()
-                            }),
-                        user_id: user.user_id,
-                        username: user.username,
-                        state: match user.state {
-                            user_manager::LocalUserState::Managed => {
-                                iota_ipc::LocalUserState::Managed
-                            }
-                            user_manager::LocalUserState::Released => {
-                                iota_ipc::LocalUserState::Released
-                            }
-                        },
-                        data_present: user.data_present,
+                    .map(|user| {
+                        let profile = user_manager::get_user(user.user_id)?;
+                        Ok(UserSummary {
+                            credential_present: user.state == user_manager::LocalUserState::Managed
+                                && profile.is_some_and(|profile| {
+                                    iota_util::file_util::read_user_credential_with_legacy(
+                                        user.user_id,
+                                        &profile.username,
+                                    )
+                                    .ok()
+                                    .flatten()
+                                    .is_some()
+                                }),
+                            user_id: user.user_id,
+                            username: user.username,
+                            state: match user.state {
+                                user_manager::LocalUserState::Managed => {
+                                    iota_ipc::LocalUserState::Managed
+                                }
+                                user_manager::LocalUserState::Released => {
+                                    iota_ipc::LocalUserState::Released
+                                }
+                            },
+                            data_present: user.data_present,
+                        })
                     })
-                    .collect();
+                    .collect::<Result<Vec<_>, iota_storage::storage_error::StorageError>>();
+                let Ok(users) = users else {
+                    return ResponseResult::Error(IpcErrorCode::StorageFailure);
+                };
                 ResponseResult::Ok(ResponsePayload::Users(users))
             }
             LocalRequest::CreateUser { username } => {
@@ -206,6 +222,9 @@ impl CommandRouter {
                             omikron_connector::user_ops::CreateUserError::RemoteRejected => {
                                 ResponseResult::Error(IpcErrorCode::Conflict)
                             }
+                            omikron_connector::user_ops::CreateUserError::LocalFinalizationPending { .. } => {
+                                ResponseResult::Error(IpcErrorCode::StorageFailure)
+                            }
                             omikron_connector::user_ops::CreateUserError::LocalPersistence(_) => {
                                 ResponseResult::Error(IpcErrorCode::StorageFailure)
                             }
@@ -247,7 +266,8 @@ impl CommandRouter {
                 let contents = match credential {
                     Some(value) => Ok(value.0),
                     None => user_manager::get_user(user_id)
-                        .ok_or(())
+                        .map_err(|_| ())
+                        .and_then(|user| user.ok_or(()))
                         .and_then(|user| {
                             iota_util::file_util::read_user_credential_with_legacy(
                                 user_id,
@@ -278,8 +298,27 @@ impl CommandRouter {
             }
             LocalRequest::RemoveUser { .. } => ResponseResult::Error(IpcErrorCode::InvalidRequest),
             LocalRequest::ReleaseUser { user_id } => {
-                if user_manager::get_user(user_id).is_none() {
+                let user = match user_manager::get_user(user_id) {
+                    Ok(user) => user,
+                    Err(_) => return ResponseResult::Error(IpcErrorCode::StorageFailure),
+                };
+                let Some(user) = user else {
                     return ResponseResult::Error(IpcErrorCode::NotFound);
+                };
+                if pending_operations::upsert(&PendingUserOperation {
+                    user_id,
+                    operation: PendingUserOperationKind::Release,
+                    username: user.username,
+                    public_key: None,
+                    private_key_hash: None,
+                    reset_token: None,
+                    registration_token: None,
+                    phase: PendingUserOperationPhase::Prepared,
+                    created_at: now_millis(),
+                })
+                .is_err()
+                {
+                    return ResponseResult::Error(IpcErrorCode::StorageFailure);
                 }
                 let request = CommunicationValue::new(CommunicationType::ReleaseUserFromIota)
                     .add_typed_default(DataType::UserId, DataValue::SignedNumber(user_id.into()));
@@ -291,11 +330,14 @@ impl CommandRouter {
                 {
                     Ok(response) if response.is_type(CommunicationType::Success) => {
                         match user_manager::release_user(user_id) {
-                            Ok(()) => ResponseResult::Ok(ResponsePayload::Acknowledged {
-                                message: format!(
-                                    "Released user {user_id}; hosted data was retained"
-                                ),
-                            }),
+                            Ok(()) if pending_operations::remove(user_id).is_ok() => {
+                                ResponseResult::Ok(ResponsePayload::Acknowledged {
+                                    message: format!(
+                                        "Released user {user_id}; hosted data was retained"
+                                    ),
+                                })
+                            }
+                            Ok(()) => ResponseResult::Error(IpcErrorCode::StorageFailure),
                             Err(error) => {
                                 log!(
                                     "Remote release succeeded but local cleanup failed for {user_id}: {error}"
@@ -305,9 +347,13 @@ impl CommandRouter {
                         }
                     }
                     Ok(response) if response.is_type(CommunicationType::ErrorNotAuthenticated) => {
+                        let _ = pending_operations::remove(user_id);
                         ResponseResult::Error(IpcErrorCode::Unauthorized)
                     }
-                    Ok(_) => ResponseResult::Error(IpcErrorCode::Conflict),
+                    Ok(_) => {
+                        let _ = pending_operations::remove(user_id);
+                        ResponseResult::Error(IpcErrorCode::Conflict)
+                    }
                     Err(omikron_connector::OmikronError::Timeout(_)) => {
                         ResponseResult::Error(IpcErrorCode::Timeout)
                     }
@@ -406,7 +452,7 @@ impl CommandRouter {
                 ResponseResult::Ok(ResponsePayload::Components(components))
             }
             LocalRequest::GetUser { user_id } => match user_manager::get_user(user_id) {
-                Some(user) => {
+                Ok(Some(user)) => {
                     let credential_present =
                         iota_util::file_util::read_user_credential_with_legacy(
                             user_id,
@@ -429,7 +475,8 @@ impl CommandRouter {
                         credential_present,
                     }))
                 }
-                None => ResponseResult::Error(IpcErrorCode::NotFound),
+                Ok(None) => ResponseResult::Error(IpcErrorCode::NotFound),
+                Err(_) => ResponseResult::Error(IpcErrorCode::StorageFailure),
             },
             LocalRequest::ImportUser { .. } => ResponseResult::Error(IpcErrorCode::InvalidRequest),
             LocalRequest::GetLogs { limit } => {

iota-storage/src/users/mod.rs

@@ -1,3 +1,4 @@
 pub mod contact;
+pub mod pending_operations;
 pub mod user_manager;
 pub mod user_profile;

iota-storage/src/users/pending_operations.rs

@@ -0,0 +1,173 @@
+use crate::storage_error::StorageError;
+use crate::util::db;
+use rusqlite::params;
+
+#[derive(Clone, Copy, Debug, PartialEq, Eq)]
+pub enum PendingUserOperationKind {
+    Create,
+    Attach,
+    Release,
+}
+
+#[derive(Clone, Copy, Debug, PartialEq, Eq)]
+pub enum PendingUserOperationPhase {
+    Prepared,
+    CredentialWritten,
+    RemoteCommitted,
+    LocalCommitted,
+}
+
+impl PendingUserOperationPhase {
+    fn as_str(self) -> &'static str {
+        match self {
+            Self::Prepared => "prepared",
+            Self::CredentialWritten => "credential_written",
+            Self::RemoteCommitted => "remote_committed",
+            Self::LocalCommitted => "local_committed",
+        }
+    }
+
+    fn parse(value: &str) -> Result<Self, StorageError> {
+        match value {
+            "prepared" => Ok(Self::Prepared),
+            "credential_written" => Ok(Self::CredentialWritten),
+            "remote_committed" => Ok(Self::RemoteCommitted),
+            "local_committed" => Ok(Self::LocalCommitted),
+            _ => Err(StorageError::Other(
+                "unknown pending user operation phase".into(),
+            )),
+        }
+    }
+}
+
+impl PendingUserOperationKind {
+    fn as_str(self) -> &'static str {
+        match self {
+            Self::Create => "create",
+            Self::Attach => "attach",
+            Self::Release => "release",
+        }
+    }
+
+    fn parse(value: &str) -> Result<Self, StorageError> {
+        match value {
+            "create" => Ok(Self::Create),
+            "attach" => Ok(Self::Attach),
+            "release" => Ok(Self::Release),
+            _ => Err(StorageError::Other("unknown pending user operation".into())),
+        }
+    }
+}
+
+#[derive(Clone, Debug, PartialEq, Eq)]
+pub struct PendingUserOperation {
+    pub user_id: i64,
+    pub operation: PendingUserOperationKind,
+    pub username: String,
+    pub public_key: Option<String>,
+    pub private_key_hash: Option<String>,
+    pub reset_token: Option<String>,
+    pub registration_token: Option<String>,
+    pub phase: PendingUserOperationPhase,
+    pub created_at: i64,
+}
+
+pub fn upsert(operation: &PendingUserOperation) -> Result<(), StorageError> {
+    db::with_immediate_transaction(|tx| {
+        tx.execute(
+            r#"
+            INSERT INTO pending_user_operations (
+                user_id, operation, username, public_key, private_key_hash,
+                reset_token, registration_token, phase, created_at
+            ) VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8, ?9)
+            ON CONFLICT(user_id) DO UPDATE SET
+                operation = excluded.operation,
+                username = excluded.username,
+                public_key = excluded.public_key,
+                private_key_hash = excluded.private_key_hash,
+                reset_token = excluded.reset_token,
+                registration_token = excluded.registration_token,
+                phase = excluded.phase,
+                created_at = excluded.created_at
+            "#,
+            params![
+                operation.user_id,
+                operation.operation.as_str(),
+                operation.username,
+                operation.public_key,
+                operation.private_key_hash,
+                operation.reset_token,
+                operation.registration_token,
+                operation.phase.as_str(),
+                operation.created_at,
+            ],
+        )?;
+        Ok(())
+    })
+}
+
+pub fn get_all() -> Result<Vec<PendingUserOperation>, StorageError> {
+    db::with_db(|conn| {
+        let mut statement = conn.prepare(
+            "SELECT user_id, operation, username, public_key, private_key_hash, reset_token, registration_token, phase, created_at FROM pending_user_operations ORDER BY created_at",
+        )?;
+        let rows = statement.query_map([], |row| {
+            Ok((
+                row.get::<_, i64>(0)?,
+                row.get::<_, String>(1)?,
+                row.get::<_, String>(2)?,
+                row.get::<_, Option<String>>(3)?,
+                row.get::<_, Option<String>>(4)?,
+                row.get::<_, Option<String>>(5)?,
+                row.get::<_, Option<String>>(6)?,
+                row.get::<_, String>(7)?,
+                row.get::<_, i64>(8)?,
+            ))
+        })?;
+        rows.map(|row| {
+            let (
+                user_id,
+                operation,
+                username,
+                public_key,
+                private_key_hash,
+                reset_token,
+                registration_token,
+                phase,
+                created_at,
+            ) = row?;
+            Ok(PendingUserOperation {
+                user_id,
+                operation: PendingUserOperationKind::parse(&operation)?,
+                username,
+                public_key,
+                private_key_hash,
+                reset_token,
+                registration_token,
+                phase: PendingUserOperationPhase::parse(&phase)?,
+                created_at,
+            })
+        })
+        .collect()
+    })
+}
+
+pub fn update_phase(user_id: i64, phase: PendingUserOperationPhase) -> Result<(), StorageError> {
+    db::with_immediate_transaction(|tx| {
+        tx.execute(
+            "UPDATE pending_user_operations SET phase = ?1 WHERE user_id = ?2",
+            params![phase.as_str(), user_id],
+        )?;
+        Ok(())
+    })
+}
+
+pub fn remove(user_id: i64) -> Result<(), StorageError> {
+    db::with_immediate_transaction(|tx| {
+        tx.execute(
+            "DELETE FROM pending_user_operations WHERE user_id = ?1",
+            [user_id],
+        )?;
+        Ok(())
+    })
+}

iota-storage/src/users/user_manager.rs

@@ -32,8 +32,8 @@ pub fn add_user(user: UserProfile) {
 }
 
 pub fn try_add_user(user: UserProfile) -> Result<(), crate::storage_error::StorageError> {
-    db::with_db(|conn| {
-        conn.execute(
+    db::with_immediate_transaction(|tx| {
+        tx.execute(
             r#"
             INSERT INTO users (user_id, username, public_key, private_key_hash, reset_token, created_at, display_name)
             VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7)
@@ -56,7 +56,7 @@ pub fn try_add_user(user: UserProfile) -> Result<(), crate::storage_error::Stora
         )?;
 
         for (app_id, app_secret) in &user.trusted_apps {
-            conn.execute(
+            tx.execute(
                 r#"
                 INSERT OR REPLACE INTO trusted_apps (user_id, app_id, app_secret)
                 VALUES (?1, ?2, ?3)
@@ -64,7 +64,7 @@ pub fn try_add_user(user: UserProfile) -> Result<(), crate::storage_error::Stora
                 params![user.user_id, app_id, app_secret],
             )?;
         }
-        conn.execute(
+        tx.execute(
             r#"INSERT INTO user_residency (user_id, username, lifecycle_state, data_state, updated_at)
                VALUES (?1, ?2, 'managed', COALESCE((SELECT data_state FROM user_residency WHERE user_id = ?1), 'present'), ?3)
                ON CONFLICT(user_id) DO UPDATE SET username = excluded.username, lifecycle_state = 'managed', updated_at = excluded.updated_at"#,
@@ -93,7 +93,7 @@ pub fn get_user_by_username(username: &str) -> Option<UserProfile> {
                     private_key_hash: r.get(3)?,
                     created_at: r.get(5)?,
                     reset_token: r.get(4)?,
-                    trusted_apps: load_trusted_apps(user_id),
+                    trusted_apps: std::collections::HashMap::new(),
                 })
             },
         ) {
@@ -110,8 +110,8 @@ pub fn get_user_by_username(username: &str) -> Option<UserProfile> {
     }
 }
 
-pub fn get_user(user_id: i64) -> Option<UserProfile> {
-    match db::with_db(|conn| {
+pub fn get_user(user_id: i64) -> Result<Option<UserProfile>, crate::storage_error::StorageError> {
+    let user = db::with_db(|conn| {
         match conn.query_row(
             "SELECT user_id, username, public_key, private_key_hash, reset_token, created_at, display_name FROM users WHERE user_id = ?1 LIMIT 1",
             params![user_id],
@@ -125,7 +125,7 @@ pub fn get_user(user_id: i64) -> Option<UserProfile> {
                     private_key_hash: r.get(3)?,
                     created_at: r.get(5)?,
                     reset_token: r.get(4)?,
-                    trusted_apps: load_trusted_apps(user_id),
+                    trusted_apps: std::collections::HashMap::new(),
                 })
             },
         ) {
@@ -133,13 +133,12 @@ pub fn get_user(user_id: i64) -> Option<UserProfile> {
             Err(rusqlite::Error::QueryReturnedNoRows) => Ok(None),
             Err(e) => Err(e.into()),
         }
-    }) {
-        Ok(opt) => opt,
-        Err(e) => {
-            eprintln!("Error querying user: {}", e);
-            None
-        }
-    }
+    })?;
+    user.map(|mut user| {
+        user.trusted_apps = load_trusted_apps(user_id)?;
+        Ok(user)
+    })
+    .transpose()
 }
 
 pub fn get_users() -> Vec<UserProfile> {
@@ -177,7 +176,7 @@ pub fn get_users() -> Vec<UserProfile> {
         for row in rows {
             match row {
                 Ok(mut user) => {
-                    user.trusted_apps = load_trusted_apps(user.user_id);
+                    user.trusted_apps = load_trusted_apps(user.user_id)?;
                     out.push(user);
                 }
                 Err(e) => eprintln!("Failed to read user row: {}", e),
@@ -193,8 +192,10 @@ pub fn get_users() -> Vec<UserProfile> {
     }
 }
 
-fn load_trusted_apps(user_id: i64) -> std::collections::HashMap<String, String> {
-    match db::with_db(|conn| {
+fn load_trusted_apps(
+    user_id: i64,
+) -> Result<std::collections::HashMap<String, String>, crate::storage_error::StorageError> {
+    db::with_db(|conn| {
         let mut stmt =
             conn.prepare("SELECT app_id, app_secret FROM trusted_apps WHERE user_id = ?1")?;
         let rows = stmt.query_map(params![user_id], |r| {
@@ -203,18 +204,11 @@ fn load_trusted_apps(user_id: i64) -> std::collections::HashMap<String, String>
 
         let mut map = std::collections::HashMap::new();
         for row in rows {
-            if let Ok((k, v)) = row {
-                map.insert(k, v);
-            }
+            let (key, value) = row?;
+            map.insert(key, value);
         }
         Ok(map)
-    }) {
-        Ok(m) => m,
-        Err(e) => {
-            eprintln!("Failed to load trusted apps: {}", e);
-            std::collections::HashMap::new()
-        }
-    }
+    })
 }
 
 pub fn remove_user(user_id: i64) {
@@ -233,9 +227,11 @@ pub fn remove_user(user_id: i64) {
 /// Remove only local management authority. Hosted content is intentionally
 /// retained and is indexed as released for a later purge operation.
 pub fn release_user(user_id: i64) -> Result<(), crate::storage_error::StorageError> {
-    let username = get_user(user_id).map(|user| user.username).ok_or_else(|| {
-        crate::storage_error::StorageError::Other("managed user was not found".into())
-    })?;
+    let username = get_user(user_id)?
+        .map(|user| user.username)
+        .ok_or_else(|| {
+            crate::storage_error::StorageError::Other("managed user was not found".into())
+        })?;
     db::with_db(|conn| {
         let tx = conn.unchecked_transaction()?;
         tx.execute(
@@ -252,7 +248,7 @@ pub fn release_user(user_id: i64) -> Result<(), crate::storage_error::StorageErr
         tx.commit()?;
         Ok(())
     })?;
-    remove_user_credential(user_id)
+    remove_user_credential(user_id, Some(&username))
         .map_err(|error| crate::storage_error::StorageError::Other(error.to_string()))
 }
 
@@ -312,6 +308,12 @@ pub fn purge_user_data(user_id: i64) -> Result<(), crate::storage_error::Storage
 /// Complete local erasure is idempotent and is the target for a durable
 /// Omega-hosted erasure request after account deletion.
 pub fn erase_user_locally(user_id: i64) -> Result<(), crate::storage_error::StorageError> {
+    let username = get_user(user_id)?.map(|user| user.username).or_else(|| {
+        get_residency()
+            .into_iter()
+            .find(|entry| entry.user_id == user_id)
+            .map(|entry| entry.username)
+    });
     purge_user_data(user_id)?;
     db::with_db(|conn| {
         conn.execute(
@@ -325,7 +327,7 @@ pub fn erase_user_locally(user_id: i64) -> Result<(), crate::storage_error::Stor
         )?;
         Ok(())
     })?;
-    remove_user_credential(user_id)
+    remove_user_credential(user_id, username.as_deref())
         .map_err(|error| crate::storage_error::StorageError::Other(error.to_string()))
 }
 

iota-storage/src/util/chat_files.rs

@@ -523,7 +523,7 @@ pub fn add_message(message: NewMessage<'_>) -> Result<i64, StorageError> {
 
     let mut contact = crate::users::contact::Contact::new(external_user);
     contact.set_last_message_at(send_time);
-    crate::util::chats_util::mod_user(storage_owner, &contact);
+    crate::util::chats_util::mod_user(storage_owner, &contact)?;
     Ok(msg_id)
 }
 
@@ -546,8 +546,17 @@ pub fn change_message_state_by_relay_id(
             return Ok(());
         };
         let state = MessageState::from_str(&current).upgrade(new_state).as_str();
-        tx.execute("UPDATE messages SET message_state = ?1 WHERE id = ?2", params![state, msg_id])?;
-        sync::record_event(&tx, storage_owner, EntityType::Message, msg_id, Operation::Upsert)?;
+        tx.execute(
+            "UPDATE messages SET message_state = ?1 WHERE id = ?2",
+            params![state, msg_id],
+        )?;
+        sync::record_event(
+            &tx,
+            storage_owner,
+            EntityType::Message,
+            msg_id,
+            Operation::Upsert,
+        )?;
         tx.commit()?;
         Ok(())
     })
@@ -581,7 +590,9 @@ pub fn record_message_receipt(
             return Err(StorageError::Other("message receipt target was not found".into()));
         };
         if external_user != receipt_signer_id {
-            return Err(StorageError::Other("message receipt signer is not the chat partner".into()));
+            return Err(StorageError::Other(
+                "message receipt signer is not the chat partner".into(),
+            ));
         }
         tx.execute(
             "INSERT OR IGNORE INTO message_receipts (storage_owner, target_signer_id, target_message_id, receipt_signer_id, receipt_message_id, receipt_type, event_at, recorded_at) VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8)",
@@ -592,9 +603,11 @@ pub fn record_message_receipt(
         } else {
             ("client_received_at", "client_received_recorded_at")
         };
-        let state = MessageState::from_str(
-            &tx.query_row("SELECT message_state FROM messages WHERE id = ?1", [message_id], |row| row.get::<_, String>(0))?,
-        )
+        let state = MessageState::from_str(&tx.query_row(
+            "SELECT message_state FROM messages WHERE id = ?1",
+            [message_id],
+            |row| row.get::<_, String>(0),
+        )?)
         .upgrade(MessageState::from_str(receipt_type))
         .as_str()
         .to_string();
@@ -602,7 +615,13 @@ pub fn record_message_receipt(
             &format!("UPDATE messages SET {state_column} = COALESCE({state_column}, ?1), {recorded_column} = COALESCE({recorded_column}, ?2), message_state = ?3 WHERE id = ?4"),
             params![event_at, recorded_at, state, message_id],
         )?;
-        sync::record_event(&tx, storage_owner, EntityType::Message, message_id, Operation::Upsert)?;
+        sync::record_event(
+            &tx,
+            storage_owner,
+            EntityType::Message,
+            message_id,
+            Operation::Upsert,
+        )?;
         tx.commit()?;
         Ok(())
     })
@@ -630,7 +649,13 @@ pub fn record_destination_iota_received(
             "UPDATE messages SET destination_iota_received_at = COALESCE(destination_iota_received_at, ?1), message_state = CASE WHEN message_state = 'sending' THEN 'sent' ELSE message_state END WHERE id = ?2",
             params![accepted_at, message_id],
         )?;
-        sync::record_event(&tx, storage_owner, EntityType::Message, message_id, Operation::Upsert)?;
+        sync::record_event(
+            &tx,
+            storage_owner,
+            EntityType::Message,
+            message_id,
+            Operation::Upsert,
+        )?;
         tx.commit()?;
         Ok(())
     })

iota-storage/src/util/chats_util.rs

@@ -14,9 +14,8 @@ pub fn has_user(storage_owner: i64, user_id: i64) -> Result<bool, StorageError>
     })
 }
 
-pub fn mod_user(storage_owner: i64, contact: &Contact) {
-    if let Err(e) = db::with_db(|conn| {
-        let tx = conn.unchecked_transaction()?;
+pub fn mod_user(storage_owner: i64, contact: &Contact) -> Result<(), StorageError> {
+    db::with_immediate_transaction(|tx| {
         tx.execute(
             r#"
             INSERT INTO contacts (storage_owner, user_id, user_name, last_message_at)
@@ -28,37 +27,34 @@ pub fn mod_user(storage_owner: i64, contact: &Contact) {
             params![
                 storage_owner,
                 contact.user_id,
-                contact.user_name.clone(),
+                contact.user_name,
                 contact.last_message_at,
             ],
         )?;
         sync::record_event(
-            &tx,
+            tx,
             storage_owner,
             EntityType::Contact,
             contact.user_id,
             Operation::Upsert,
         )?;
-        tx.commit()?;
         Ok(())
-    }) {
-        eprintln!("Failed to mod_user: {}", e);
-    }
+    })
 }
 
-pub fn get_users_by_ids(storage_owner: i64, ids: &[i64]) -> Vec<Contact> {
+pub fn get_users_by_ids(storage_owner: i64, ids: &[i64]) -> Result<Vec<Contact>, StorageError> {
     if ids.is_empty() {
-        return Vec::new();
+        return Ok(Vec::new());
     }
     let wanted: std::collections::HashSet<i64> = ids.iter().copied().collect();
-    get_users(storage_owner)
+    Ok(get_users(storage_owner)?
         .into_iter()
         .filter(|contact| wanted.contains(&contact.user_id))
-        .collect()
+        .collect())
 }
 
-pub fn get_user(storage_owner: i64, user_id: i64) -> Option<Contact> {
-    match db::with_db(|conn| {
+pub fn get_user(storage_owner: i64, user_id: i64) -> Result<Option<Contact>, StorageError> {
+    db::with_db(|conn| {
         match conn.query_row(
             r#"
             SELECT user_id, user_name, last_message_at
@@ -79,17 +75,11 @@ pub fn get_user(storage_owner: i64, user_id: i64) -> Option<Contact> {
             Err(rusqlite::Error::QueryReturnedNoRows) => Ok(None),
             Err(e) => Err(e.into()),
         }
-    }) {
-        Ok(opt) => opt,
-        Err(e) => {
-            eprintln!("Error querying user in get_user: {}", e);
-            None
-        }
-    }
+    })
 }
 
-pub fn get_users(storage_owner: i64) -> Vec<Contact> {
-    match db::with_db(|conn| {
+pub fn get_users(storage_owner: i64) -> Result<Vec<Contact>, StorageError> {
+    db::with_db(|conn| {
         let mut stmt = conn.prepare(
             r#"
             SELECT user_id, user_name, last_message_at
@@ -112,17 +102,8 @@ pub fn get_users(storage_owner: i64) -> Vec<Contact> {
 
         let mut out = Vec::new();
         for row in rows {
-            match row {
-                Ok(contact) => out.push(contact),
-                Err(e) => eprintln!("Failed to read contact row: {}", e),
-            }
+            out.push(row?);
         }
         Ok(out)
-    }) {
-        Ok(v) => v,
-        Err(e) => {
-            eprintln!("Failed to query contacts in get_users: {}", e);
-            Vec::new()
-        }
-    }
+    })
 }

iota-storage/src/util/db.rs

@@ -421,9 +421,15 @@ fn run_migrations_on_connection(conn: &Connection) -> Result<(), StorageError> {
             ("relay_message_id", "relay_message_id TEXT"),
             ("authored_at", "authored_at INTEGER"),
             ("origin_iota_received_at", "origin_iota_received_at INTEGER"),
-            ("destination_iota_received_at", "destination_iota_received_at INTEGER"),
+            (
+                "destination_iota_received_at",
+                "destination_iota_received_at INTEGER",
+            ),
             ("client_received_at", "client_received_at INTEGER"),
-            ("client_received_recorded_at", "client_received_recorded_at INTEGER"),
+            (
+                "client_received_recorded_at",
+                "client_received_recorded_at INTEGER",
+            ),
             ("read_at", "read_at INTEGER"),
             ("read_recorded_at", "read_recorded_at INTEGER"),
         ] {
@@ -489,6 +495,38 @@ fn run_migrations_on_connection(conn: &Connection) -> Result<(), StorageError> {
         )?;
     }
 
+    if current_version < 13 {
+        conn.execute_batch(
+            r#"
+            CREATE TABLE IF NOT EXISTS pending_user_operations (
+                user_id INTEGER PRIMARY KEY,
+                operation TEXT NOT NULL
+                    CHECK (operation IN ('create', 'attach', 'release')),
+                username TEXT NOT NULL,
+                public_key TEXT,
+                private_key_hash TEXT,
+                reset_token TEXT,
+                registration_token TEXT,
+                created_at INTEGER NOT NULL
+            );
+            CREATE INDEX IF NOT EXISTS idx_pending_user_operations_operation
+                ON pending_user_operations (operation, created_at);
+            PRAGMA user_version = 13;
+            "#,
+        )?;
+    }
+
+    if current_version < 14 {
+        conn.execute_batch(
+            r#"
+            ALTER TABLE pending_user_operations
+                ADD COLUMN phase TEXT NOT NULL DEFAULT 'prepared'
+                CHECK (phase IN ('prepared', 'credential_written', 'remote_committed', 'local_committed'));
+            PRAGMA user_version = 14;
+            "#,
+        )?;
+    }
+
     Ok(())
 }
 
@@ -558,7 +596,7 @@ mod tests {
         run_migrations_on_connection(&conn)?;
 
         let version: i64 = conn.pragma_query_value(None, "user_version", |row| row.get(0))?;
-        assert_eq!(version, 12);
+        assert_eq!(version, 13);
         for column in ["height", "reply_to", "edited_count", "deleted_by_external"] {
             let mut statement =
                 conn.prepare("SELECT 1 FROM pragma_table_info('messages') WHERE name = ?1")?;
@@ -577,7 +615,7 @@ mod tests {
         run_migrations_on_connection(&conn)?;
         run_migrations_on_connection(&conn)?;
         let version: i64 = conn.pragma_query_value(None, "user_version", |row| row.get(0))?;
-        assert_eq!(version, 12);
+        assert_eq!(version, 13);
         for table in [
             "sync_heads",
             "sync_events",
@@ -587,6 +625,7 @@ mod tests {
             "pending_relays",
             "relay_inbox",
             "synced_settings",
+            "pending_user_operations",
         ] {
             let exists: i64 = conn.query_row(
                 "SELECT COUNT(*) FROM sqlite_master WHERE type = 'table' AND name = ?1",
@@ -611,8 +650,17 @@ mod tests {
         run_migrations_on_connection(&conn)?;
 
         let version: i64 = conn.pragma_query_value(None, "user_version", |row| row.get(0))?;
-        assert_eq!(version, 12);
-        for column in ["id", "user_id", "scope_type", "scope_key", "name", "payload", "revision", "deleted"] {
+        assert_eq!(version, 13);
+        for column in [
+            "id",
+            "user_id",
+            "scope_type",
+            "scope_key",
+            "name",
+            "payload",
+            "revision",
+            "deleted",
+        ] {
             let mut statement =
                 conn.prepare("SELECT 1 FROM pragma_table_info('synced_settings') WHERE name = ?1")?;
             assert!(statement.exists([column])?);

iota-storage/src/util/sync.rs

@@ -291,30 +291,15 @@ mod tests {
             )
             .unwrap();
         let transaction = connection.unchecked_transaction().unwrap();
-        let message = super::record_event(
-            &transaction,
-            1,
-            EntityType::Message,
-            10,
-            Operation::Upsert,
-        )
-        .unwrap();
-        let setting = super::record_event(
-            &transaction,
-            1,
-            EntityType::Setting,
-            11,
-            Operation::Upsert,
-        )
-        .unwrap();
-        let contact = super::record_event(
-            &transaction,
-            1,
-            EntityType::Contact,
-            12,
-            Operation::Upsert,
-        )
-        .unwrap();
+        let message =
+            super::record_event(&transaction, 1, EntityType::Message, 10, Operation::Upsert)
+                .unwrap();
+        let setting =
+            super::record_event(&transaction, 1, EntityType::Setting, 11, Operation::Upsert)
+                .unwrap();
+        let contact =
+            super::record_event(&transaction, 1, EntityType::Contact, 12, Operation::Upsert)
+                .unwrap();
         transaction.commit().unwrap();
 
         assert_eq!((message, setting, contact), (1, 2, 3));

iota-storage/src/util/synced_settings.rs

@@ -1,7 +1,7 @@
 use crate::storage_error::StorageError;
 use crate::util::db;
 use crate::util::sync::{self, EntityType, Operation};
-use rusqlite::{params, Connection, OptionalExtension, Row, Transaction};
+use rusqlite::{Connection, OptionalExtension, Row, Transaction, params};
 
 #[derive(Debug, Clone, Copy, PartialEq, Eq)]
 pub enum SettingScope {
@@ -401,8 +401,8 @@ pub(crate) fn delete_scope_in_tx(
 #[cfg(test)]
 mod tests {
     use super::{
-        delete_in_tx, get_from_connection, is_valid_name, list_by_ids_from_connection,
-        list_from_connection, set_in_tx, SettingScope,
+        SettingScope, delete_in_tx, get_from_connection, is_valid_name,
+        list_by_ids_from_connection, list_from_connection, set_in_tx,
     };
     use rusqlite::Connection;
 
@@ -618,9 +618,11 @@ mod tests {
         assert!(deleted.changed);
         assert_eq!(journal_operation, "delete");
         assert!(list_from_connection(&connection, 1).unwrap().is_empty());
-        assert!(list_by_ids_from_connection(&connection, 1, &[stored.id])
-            .unwrap()
-            .is_empty());
+        assert!(
+            list_by_ids_from_connection(&connection, 1, &[stored.id])
+                .unwrap()
+                .is_empty()
+        );
     }
 
     #[test]

iota-util/src/file_util.rs

@@ -43,14 +43,35 @@ pub fn delete_user_directory(user_id: i64) -> io::Result<()> {
     fs::remove_dir_all(user_dir)
 }
 
-pub fn credential_path(user_id: i64) -> PathBuf {
-    storage_directory()
-        .join("credentials")
-        .join(format!("{user_id}.tu"))
+fn credential_filename(username: &str) -> io::Result<String> {
+    if username.is_empty()
+        || username.chars().any(char::is_control)
+        || username.contains(['/', '\\'])
+    {
+        return Err(io::Error::new(
+            io::ErrorKind::InvalidInput,
+            "unsafe credential owner name",
+        ));
+    }
+    Ok(format!("{username}.tu"))
 }
 
-pub fn read_user_credential(user_id: i64) -> io::Result<Option<String>> {
-    let path = credential_path(user_id);
+pub fn credential_path(username: &str) -> io::Result<PathBuf> {
+    credential_path_in(&storage_directory(), username)
+}
+
+fn credential_path_in(root: &Path, username: &str) -> io::Result<PathBuf> {
+    Ok(root
+        .join("credentials")
+        .join(credential_filename(username)?))
+}
+
+fn legacy_credential_path(user_id: i64) -> io::Result<PathBuf> {
+    storage_file("credentials", format!("{user_id}.tu"))
+}
+
+pub fn read_user_credential(username: &str) -> io::Result<Option<String>> {
+    let path = credential_path(username)?;
     match fs::read_to_string(path) {
         Ok(value) => Ok(Some(value)),
         Err(error) if error.kind() == io::ErrorKind::NotFound => Ok(None),
@@ -58,55 +79,66 @@ pub fn read_user_credential(user_id: i64) -> io::Result<Option<String>> {
     }
 }
 
-/// Resolve a credential by immutable account id. A valid legacy
-/// `<username>.tu` is migrated atomically the first time it is encountered.
+/* Resolve a credential by account id while using the owner's name for the
+ * canonical filename. Older ID-based and root-level files are migrated when
+ * they are encountered. */
 pub fn read_user_credential_with_legacy(
     user_id: i64,
     username: &str,
 ) -> io::Result<Option<String>> {
-    if let Some(credential) = read_user_credential(user_id)? {
+    let canonical_path = credential_path(username)?;
+    if let Some(credential) = read_user_credential(username)? {
         return Ok(Some(credential));
     }
-    let legacy = storage_file("", format!("{username}.tu"))?;
-    let credential = match fs::read_to_string(&legacy) {
-        Ok(value) => value,
-        Err(error) if error.kind() == io::ErrorKind::NotFound => return Ok(None),
-        Err(error) => return Err(error),
-    };
-    let parsed = crate::tu::TuCredential::parse(&credential)
-        .map_err(|error| io::Error::new(io::ErrorKind::InvalidData, error))?;
-    if parsed.user_id != user_id {
-        return Err(io::Error::new(
-            io::ErrorKind::InvalidData,
-            "legacy credential user id mismatch",
-        ));
+    let legacy_paths = [
+        legacy_credential_path(user_id)?,
+        storage_file("", credential_filename(username)?)?,
+    ];
+    for legacy_path in legacy_paths {
+        let credential = match fs::read_to_string(&legacy_path) {
+            Ok(value) => value,
+            Err(error) if error.kind() == io::ErrorKind::NotFound => continue,
+            Err(error) => return Err(error),
+        };
+        let parsed = crate::tu::TuCredential::parse(&credential)
+            .map_err(|error| io::Error::new(io::ErrorKind::InvalidData, error))?;
+        if parsed.user_id != user_id {
+            return Err(io::Error::new(
+                io::ErrorKind::InvalidData,
+                "legacy credential user id mismatch",
+            ));
+        }
+        write_user_credential(username, &parsed.to_canonical_string())?;
+        if legacy_path != canonical_path {
+            fs::remove_file(legacy_path)?;
+        }
+        return Ok(Some(parsed.to_canonical_string()));
     }
-    write_user_credential(user_id, &parsed.to_canonical_string())?;
-    fs::remove_file(legacy)?;
-    Ok(Some(parsed.to_canonical_string()))
+    Ok(None)
 }
 
-pub fn write_user_credential(user_id: i64, credential: &str) -> io::Result<()> {
-    let path = credential_path(user_id);
-    let parent = path.parent().expect("credential path has parent");
-    fs::create_dir_all(parent)?;
-    let temporary = parent.join(format!(".{user_id}.tu.tmp"));
-    fs::write(&temporary, credential)?;
-    if let Err(error) = fs::rename(&temporary, &path) {
-        let _ = fs::remove_file(&temporary);
-        return Err(error);
+pub fn write_user_credential(username: &str, credential: &str) -> io::Result<()> {
+    let path = credential_path(username)?;
+    crate::atomic_file::replace_private(&path, credential.as_bytes(), 0)
+}
+
+pub fn remove_user_credential(user_id: i64, username: Option<&str>) -> io::Result<()> {
+    let mut paths = vec![legacy_credential_path(user_id)?];
+    if let Some(username) = username {
+        paths.push(credential_path(username)?);
+        paths.push(storage_file("", credential_filename(username)?)?);
+    }
+
+    for path in paths {
+        match fs::remove_file(path) {
+            Ok(()) => {}
+            Err(error) if error.kind() == io::ErrorKind::NotFound => {}
+            Err(error) => return Err(error),
+        }
     }
     Ok(())
 }
 
-pub fn remove_user_credential(user_id: i64) -> io::Result<()> {
-    match fs::remove_file(credential_path(user_id)) {
-        Ok(()) => Ok(()),
-        Err(error) if error.kind() == io::ErrorKind::NotFound => Ok(()),
-        Err(error) => Err(error),
-    }
-}
-
 pub fn load_file_buf(path: &str, name: &str) -> io::Result<BufReader<File>> {
     let file_path = storage_file(path, name)?;
 
@@ -455,3 +487,22 @@ pub async fn download_and_extract_zip(url: &str, as_name: &str) {
         println!("Downloaded and extracted ZIP file successfully.");
     }
 }
+
+#[cfg(test)]
+mod tests {
+    use super::credential_path_in;
+    use std::path::Path;
+
+    #[test]
+    fn credential_path_uses_owner_name() {
+        let path = credential_path_in(Path::new("/tmp/iota"), "alice").unwrap();
+        assert!(path.ends_with("credentials/alice.tu"));
+        assert!(!path.ends_with("credentials/42.tu"));
+    }
+
+    #[test]
+    fn credential_path_rejects_unsafe_owner_name() {
+        assert!(credential_path_in(Path::new("/tmp/iota"), "../alice").is_err());
+        assert!(credential_path_in(Path::new("/tmp/iota"), "alice/bob").is_err());
+    }
+}

iota-util/src/tu.rs

@@ -1,8 +1,6 @@
-//! Strict parsing and storage-independent handling of user credentials.
-//!
-//! A `.tu` file is deliberately identified by the account id embedded in its
-//! contents.  Its filename is presentation data owned by the CLI, never an
-//! account authority.
+/* Strict parsing and storage-independent handling of user credentials. A
+ * `.tu` file is identified by the account ID in its contents, while storage
+ * names the file after its owner's username. */
 
 use crate::crypto_helper::{keyring_from_base64, keyring_to_base64};
 use mtp::crypto::{Keyring, PublicKeyBundle};

omikron-connector/src/omikron_connection.rs

@@ -754,7 +754,9 @@ impl OmikronConnection {
         let signer_id_i64 = i64::try_from(signer_id).map_err(|_| {
             RelayValidationError::KeyLookup("signer ID exceeds local storage range".into())
         })?;
-        if let Some(user) = iota_storage::users::user_manager::get_user(signer_id_i64) {
+        let local_user = iota_storage::users::user_manager::get_user(signer_id_i64)
+            .map_err(|error| RelayValidationError::KeyLookup(error.to_string()))?;
+        if let Some(user) = local_user {
             let key = iota_util::crypto_helper::public_key_bundle_from_base64(&user.public_key)
                 .ok_or_else(|| {
                     RelayValidationError::KeyLookup("stored user key is invalid".into())
@@ -788,7 +790,10 @@ impl OmikronConnection {
     pub async fn hosting_iota_for_user(&self, user_id: u64) -> Result<u64, String> {
         let user_id_i64 = i64::try_from(user_id)
             .map_err(|_| "user ID exceeds local storage range".to_string())?;
-        if iota_storage::users::user_manager::get_user(user_id_i64).is_some() {
+        if iota_storage::users::user_manager::get_user(user_id_i64)
+            .map_err(|error| error.to_string())?
+            .is_some()
+        {
             return CONFIG
                 .load()
                 .iota_id
@@ -901,14 +906,54 @@ impl OmikronConnection {
             }
         };
         let accepted_at = now_millis_i64();
-        let signer_is_local = i64::try_from(verified.context.signer_id)
-            .ok()
-            .and_then(iota_storage::users::user_manager::get_user)
-            .is_some();
-        let recipient_is_local = i64::try_from(verified.context.final_recipient_id)
-            .ok()
-            .and_then(iota_storage::users::user_manager::get_user)
-            .is_some();
+        let signer_id = match i64::try_from(verified.context.signer_id) {
+            Ok(id) => id,
+            Err(_) => {
+                self.send_relay_response(
+                    Some(incoming_frame_id),
+                    CommunicationType::ErrorInvalidData,
+                )
+                .await;
+                return;
+            }
+        };
+        let recipient_id = match i64::try_from(verified.context.final_recipient_id) {
+            Ok(id) => id,
+            Err(_) => {
+                self.send_relay_response(
+                    Some(incoming_frame_id),
+                    CommunicationType::ErrorInvalidData,
+                )
+                .await;
+                return;
+            }
+        };
+        let signer_is_local = match iota_storage::users::user_manager::get_user(signer_id) {
+            Ok(user) => user.is_some(),
+            Err(error) => {
+                log!(
+                    "Relay locality lookup failed for signer {}: {}",
+                    signer_id,
+                    error
+                );
+                self.send_relay_response(Some(incoming_frame_id), CommunicationType::ErrorInternal)
+                    .await;
+                return;
+            }
+        };
+        let recipient_is_local = match iota_storage::users::user_manager::get_user(recipient_id) {
+            Ok(user) => user.is_some(),
+            Err(error) => {
+                log!(
+                    "Relay locality lookup failed for recipient {}: {}",
+                    recipient_id,
+                    error
+                );
+                self.send_relay_response(Some(incoming_frame_id), CommunicationType::ErrorInternal)
+                    .await;
+                return;
+            }
+        };
         if !signer_is_local && !recipient_is_local {
             log!(
                 "Rejecting Relay with no local origin or destination: signer {}, recipient {}",
@@ -991,7 +1036,7 @@ impl OmikronConnection {
             let content = match open_verified_relay_content(
                 &verified,
                 &[&keyring],
-                verified.context.signer_id,
+                verified.context.final_recipient_id,
             ) {
                 Ok(value) => value,
                 Err(error) => {
@@ -1486,7 +1531,6 @@ impl OmikronConnection {
         dispatch!(MessageGet, handle_message_get);
         dispatch!(MessagesGet, handle_messages_get);
         dispatch!(GetChats, handle_get_chats);
-        dispatch!(AddConversation, handle_add_conversation);
         dispatch!(AddCommunity, handle_add_community);
         dispatch!(GetCommunities, handle_get_communities);
         dispatch!(RemoveCommunity, handle_remove_community);
@@ -1568,7 +1612,16 @@ impl OmikronConnection {
         };
 
         let mut trusted = false;
-        if let Some(user) = iota_storage::users::user_manager::get_user(user_id) {
+        let user = match iota_storage::users::user_manager::get_user(user_id) {
+            Ok(user) => user,
+            Err(_) => {
+                let _ = self
+                    .send_message(&error_response(cv, CommunicationType::ErrorInternal))
+                    .await;
+                return;
+            }
+        };
+        if let Some(user) = user {
             if let Some(pub_k) = user.trusted_apps.get(&app_identifier) {
                 if pub_k == &app_public_key {
                     trusted = true;
@@ -1861,7 +1914,17 @@ impl OmikronConnection {
             &mutation,
             vec![(DataType::Content, DataValue::Str(content.to_string()))],
         );
-        if iota_storage::users::user_manager::get_user(mutation.partner_id).is_some()
+        let partner_is_local =
+            match iota_storage::users::user_manager::get_user(mutation.partner_id) {
+                Ok(user) => user.is_some(),
+                Err(_) => {
+                    let _ = self
+                        .send_message(&error_response(cv, CommunicationType::ErrorInternal))
+                        .await;
+                    return;
+                }
+            };
+        if partner_is_local
             && chat_files::apply_remote_edit(
                 mutation.partner_id,
                 mutation.sender_id,
@@ -1923,7 +1986,17 @@ impl OmikronConnection {
                 (DataType::Accepted, DataValue::Bool(add)),
             ],
         );
-        if iota_storage::users::user_manager::get_user(mutation.partner_id).is_some() {
+        let partner_is_local =
+            match iota_storage::users::user_manager::get_user(mutation.partner_id) {
+                Ok(user) => user.is_some(),
+                Err(_) => {
+                    let _ = self
+                        .send_message(&error_response(cv, CommunicationType::ErrorInternal))
+                        .await;
+                    return;
+                }
+            };
+        if partner_is_local {
             let result = if add {
                 chat_files::add_reaction(
                     mutation.partner_id,
@@ -1966,7 +2039,16 @@ impl OmikronConnection {
             Some(sender_id) => sender_id,
             None => return,
         };
-        if iota_storage::users::user_manager::get_user(sender_id).is_none() {
+        let sender_is_local = match iota_storage::users::user_manager::get_user(sender_id) {
+            Ok(user) => user.is_some(),
+            Err(_) => {
+                let _ = self
+                    .send_message(&error_response(cv, CommunicationType::ErrorInternal))
+                    .await;
+                return;
+            }
+        };
+        if !sender_is_local {
             self.persist_and_deliver_remote_delete(cv).await;
             return;
         }
@@ -1988,7 +2070,17 @@ impl OmikronConnection {
             &mutation,
             Vec::new(),
         );
-        if iota_storage::users::user_manager::get_user(mutation.partner_id).is_some()
+        let partner_is_local =
+            match iota_storage::users::user_manager::get_user(mutation.partner_id) {
+                Ok(user) => user.is_some(),
+                Err(_) => {
+                    let _ = self
+                        .send_message(&error_response(cv, CommunicationType::ErrorInternal))
+                        .await;
+                    return;
+                }
+            };
+        if partner_is_local
             && chat_files::apply_remote_delete(
                 mutation.partner_id,
                 mutation.sender_id,
@@ -2024,12 +2116,6 @@ impl OmikronConnection {
             .await;
     }
 
-    async fn handle_add_conversation(self: Arc<Self>, cv: &CommunicationValue) {
-        let _ = self
-            .send_message(&message_handlers::handle_add_conversation(cv))
-            .await;
-    }
-
     async fn handle_add_community(self: Arc<Self>, cv: &CommunicationValue) {
         let _ = self
             .send_message(&message_handlers::handle_add_community(cv))

omikron-connector/src/user_ops.rs

@@ -1,16 +1,19 @@
 use base64::{Engine as _, engine::general_purpose::STANDARD};
 use iota_logger::{PrintType, log, log_cv, log_t};
+use iota_storage::users::pending_operations::{
+    self, PendingUserOperation, PendingUserOperationKind, PendingUserOperationPhase,
+};
 use iota_storage::users::user_manager::try_add_user;
 use iota_storage::users::user_profile::UserProfile;
 use iota_storage::util::config_util::CONFIG;
 use iota_util::crypto_helper::{self, hex_hash, public_key_bundle_to_base64};
-use iota_util::file_util::write_user_credential;
+use iota_util::file_util::{remove_user_credential, write_user_credential};
 use iota_util::mtp_compat::OptionalDataValueExt;
 use iota_util::tu::TuCredential;
 use mtp::codec::{CommunicationType, CommunicationValue, DataType, DataValue};
 use mtp::crypto::{Ed25519Signer, MlDsaSigner, SignatureScheme};
 use rand_core::{OsRng, RngCore};
-use std::time::Duration;
+use std::time::{Duration, SystemTime, UNIX_EPOCH};
 
 use crate::OmikronClient;
 use crate::omega_discovery;
@@ -21,6 +24,7 @@ pub enum CreateUserError {
     Transport(crate::OmikronError),
     InvalidResponse,
     RemoteRejected,
+    LocalFinalizationPending { user_id: i64 },
     LocalPersistence(String),
 }
 
@@ -165,15 +169,6 @@ pub async fn attach_user_from_tu(
     let credential = TuCredential::parse(contents)
         .map_err(|error| LifecycleUserError::InvalidCredential(error.to_string()))?;
     let (username, public_key) = inspect_credential_account(connection, &credential).await?;
-    credential_proof(
-        connection,
-        &credential,
-        CommunicationType::AttachUserBegin,
-        CommunicationType::AttachUserChallenge,
-        CommunicationType::AttachUserComplete,
-        b"tensamin:user-attach:v1\0",
-    )
-    .await?;
     let profile = UserProfile::new(
         credential.user_id,
         username,
@@ -182,10 +177,40 @@ pub async fn attach_user_from_tu(
         hex_hash(contents),
         String::new(),
     );
-    write_user_credential(profile.user_id, &credential.to_canonical_string())
+    write_user_credential(&profile.username, &credential.to_canonical_string())
         .map_err(|error| LifecycleUserError::LocalPersistence(error.to_string()))?;
+    pending_operations::upsert(&PendingUserOperation {
+        user_id: profile.user_id,
+        operation: PendingUserOperationKind::Attach,
+        username: profile.username.clone(),
+        public_key: Some(profile.public_key.clone()),
+        private_key_hash: Some(profile.private_key_hash.clone()),
+        reset_token: Some(profile.reset_token.clone()),
+        registration_token: None,
+        phase: PendingUserOperationPhase::Prepared,
+        created_at: now_millis(),
+    })
+    .map_err(|error| LifecycleUserError::LocalPersistence(error.to_string()))?;
+    if let Err(error) = credential_proof(
+        connection,
+        &credential,
+        CommunicationType::AttachUserBegin,
+        CommunicationType::AttachUserChallenge,
+        CommunicationType::AttachUserComplete,
+        b"tensamin:user-attach:v1\0",
+    )
+    .await
+    {
+        if matches!(error, LifecycleUserError::RemoteRejected) {
+            let _ = pending_operations::remove(profile.user_id);
+            let _ = remove_user_credential(profile.user_id, Some(&profile.username));
+        }
+        return Err(error);
+    }
     try_add_user(profile.clone())
         .map_err(|error| LifecycleUserError::LocalPersistence(error.to_string()))?;
+    pending_operations::remove(profile.user_id)
+        .map_err(|error| LifecycleUserError::LocalPersistence(error.to_string()))?;
     Ok(profile)
 }
 
@@ -219,6 +244,83 @@ pub async fn reconcile_managed_users(connection: &dyn OmikronClient) {
     let Ok(local_iota_id) = configured_iota_id() else {
         return;
     };
+    let pending = match pending_operations::get_all() {
+        Ok(pending) => pending,
+        Err(error) => {
+            log!("Pending user operation reconciliation could not read storage: {error}");
+            return;
+        }
+    };
+    for operation in pending {
+        let request = CommunicationValue::new(CommunicationType::GetUserData).add_typed_default(
+            DataType::UserId,
+            DataValue::SignedNumber(operation.user_id.into()),
+        );
+        let response = connection
+            .await_response(&request, Duration::from_secs(10))
+            .await
+            .ok();
+        let remote_iota_id = response.as_ref().and_then(|response| {
+            response
+                .get_data(DataType::IotaId)
+                .as_signed_number()
+                .and_then(|value| i64::try_from(value).ok())
+        });
+        let remote_matches = response.as_ref().is_some_and(|response| {
+            response.is_type(CommunicationType::GetUserData)
+                && remote_iota_id == Some(local_iota_id)
+                && response.get_data(DataType::Username).as_str() == Some(&operation.username)
+                && response.get_data(DataType::PublicKey).as_str()
+                    == operation.public_key.as_deref()
+        });
+        let completion_retried = matches!(operation.operation, PendingUserOperationKind::Create)
+            && matches!(
+                operation.phase,
+                PendingUserOperationPhase::Prepared | PendingUserOperationPhase::CredentialWritten
+            )
+            && !remote_matches
+            && complete_pending_create(connection, &operation).await;
+        match operation.operation {
+            PendingUserOperationKind::Create | PendingUserOperationKind::Attach
+                if remote_matches || completion_retried =>
+            {
+                let credential_present = iota_util::file_util::read_user_credential_with_legacy(
+                    operation.user_id,
+                    &operation.username,
+                )
+                .ok()
+                .flatten()
+                .is_some();
+                if !credential_present {
+                    log!(
+                        "Pending user {} has no credential; leaving it unresolved",
+                        operation.user_id
+                    );
+                    continue;
+                }
+                let Some(public_key) = operation.public_key else {
+                    continue;
+                };
+                let profile = UserProfile::new(
+                    operation.user_id,
+                    operation.username,
+                    None,
+                    public_key,
+                    operation.private_key_hash.unwrap_or_default(),
+                    operation.reset_token.unwrap_or_default(),
+                );
+                if try_add_user(profile).is_ok() {
+                    let _ = pending_operations::remove(operation.user_id);
+                }
+            }
+            PendingUserOperationKind::Release if remote_iota_id != Some(local_iota_id) => {
+                if iota_storage::users::user_manager::release_user(operation.user_id).is_ok() {
+                    let _ = pending_operations::remove(operation.user_id);
+                }
+            }
+            _ => {}
+        }
+    }
     for user in iota_storage::users::user_manager::get_users() {
         let request = CommunicationValue::new(CommunicationType::GetUserData).add_typed_default(
             DataType::UserId,
@@ -240,11 +342,86 @@ pub async fn reconcile_managed_users(connection: &dyn OmikronClient) {
     }
 }
 
+/*
+ * Retry completion only while the locally persisted operation still owns a
+ * valid registration lease. Omega treats an exact repeat as idempotent, which
+ * repairs an interrupted request without allocating another user ID.
+ */
+async fn complete_pending_create(
+    connection: &dyn OmikronClient,
+    operation: &PendingUserOperation,
+) -> bool {
+    let Some(public_key) = operation.public_key.as_ref() else {
+        return false;
+    };
+    let Some(reset_token) = operation.reset_token.as_ref() else {
+        return false;
+    };
+    let Some(registration_token) = operation.registration_token.as_ref() else {
+        return false;
+    };
+    let request = CommunicationValue::new(CommunicationType::CompleteRegisterUser)
+        .add_typed_default(
+            DataType::UserId,
+            DataValue::SignedNumber(operation.user_id.into()),
+        )
+        .add_typed_default(
+            DataType::Username,
+            DataValue::Str(operation.username.clone()),
+        )
+        .add_typed_default(DataType::PublicKey, DataValue::Str(public_key.clone()))
+        .add_typed_default(DataType::ResetToken, DataValue::Str(reset_token.clone()))
+        .add_typed_default(
+            DataType::RegisterId,
+            DataValue::Str(registration_token.clone()),
+        );
+    match connection
+        .await_response(&request, Duration::from_secs(20))
+        .await
+    {
+        Ok(response) if response.is_type(CommunicationType::Success) => {
+            if let Err(error) = pending_operations::update_phase(
+                operation.user_id,
+                PendingUserOperationPhase::RemoteCommitted,
+            ) {
+                log!(
+                    "Pending user {} completed remotely but could not update its phase: {error}",
+                    operation.user_id
+                );
+            }
+            true
+        }
+        Ok(response) => {
+            log!(
+                "Pending user {} registration retry was rejected with {}",
+                operation.user_id,
+                response.get_type()
+            );
+            false
+        }
+        Err(error) => {
+            log!(
+                "Pending user {} registration retry failed: {error}",
+                operation.user_id
+            );
+            false
+        }
+    }
+}
+
 fn valid_username(username: &str) -> bool {
     !username.is_empty()
-        && username.chars().count() <= 15
-        && !username.chars().any(char::is_control)
-        && !username.contains(['/', '\\'])
+        && username.len() <= 15
+        && username
+            .bytes()
+            .all(|byte| byte.is_ascii_lowercase() || byte.is_ascii_digit())
+}
+
+fn now_millis() -> i64 {
+    SystemTime::now()
+        .duration_since(UNIX_EPOCH)
+        .unwrap_or_default()
+        .as_millis() as i64
 }
 
 async fn request_user_id(connection: &dyn OmikronClient) -> Result<(i64, String), CreateUserError> {
@@ -318,6 +495,28 @@ pub async fn create_user(
         private_key_hash,
         reset_token.clone(),
     );
+    let credential = format!(
+        "{}@{}::{}",
+        user_id,
+        omega_discovery::omega_host(),
+        keyring_b64
+    );
+    pending_operations::upsert(&PendingUserOperation {
+        user_id,
+        operation: PendingUserOperationKind::Create,
+        username: user_profile.username.clone(),
+        public_key: Some(user_profile.public_key.clone()),
+        private_key_hash: Some(user_profile.private_key_hash.clone()),
+        reset_token: Some(user_profile.reset_token.clone()),
+        registration_token: Some(registration_token.clone()),
+        phase: PendingUserOperationPhase::Prepared,
+        created_at: now_millis(),
+    })
+    .map_err(|error| CreateUserError::LocalPersistence(error.to_string()))?;
+    write_user_credential(username, &credential)
+        .map_err(|error| CreateUserError::LocalPersistence(error.to_string()))?;
+    pending_operations::update_phase(user_id, PendingUserOperationPhase::CredentialWritten)
+        .map_err(|error| CreateUserError::LocalPersistence(error.to_string()))?;
 
     let communication_value = CommunicationValue::new(CommunicationType::CompleteRegisterUser)
         .add_typed_default(DataType::UserId, DataValue::SignedNumber(user_id.into()))
@@ -337,6 +536,8 @@ pub async fn create_user(
         Ok(response) => {
             log_cv!(PrintType::Omega, response);
             if !response.is_type(CommunicationType::Success) {
+                let _ = pending_operations::remove(user_id);
+                let _ = remove_user_credential(user_id, Some(username));
                 return Err(CreateUserError::RemoteRejected);
             }
         }
@@ -355,20 +556,15 @@ pub async fn create_user(
             }
         }
     }
-    log!("Created User");
-    write_user_credential(
-        user_id,
-        &format!(
-            "{}@{}::{}",
-            user_id,
-            omega_discovery::omega_host(),
-            keyring_b64
-        ),
-    )
-    .map_err(|error| CreateUserError::LocalPersistence(error.to_string()))?;
-
+    pending_operations::update_phase(user_id, PendingUserOperationPhase::RemoteCommitted)
+        .map_err(|_| CreateUserError::LocalFinalizationPending { user_id })?;
     try_add_user(user_profile.clone())
-        .map_err(|error| CreateUserError::LocalPersistence(error.to_string()))?;
+        .map_err(|_| CreateUserError::LocalFinalizationPending { user_id })?;
+    pending_operations::update_phase(user_id, PendingUserOperationPhase::LocalCommitted)
+        .map_err(|_| CreateUserError::LocalFinalizationPending { user_id })?;
+    pending_operations::remove(user_id)
+        .map_err(|_| CreateUserError::LocalFinalizationPending { user_id })?;
+    log!("Created User");
     Ok(user_profile)
 }
 
@@ -416,10 +612,12 @@ mod tests {
     #[test]
     fn validates_usernames_before_remote_registration() {
         assert!(valid_username("alice"));
-        assert!(valid_username("fifteen_char_ok"));
+        assert!(valid_username("abc123def456ghi"));
         assert!(!valid_username(""));
         assert!(!valid_username("sixteen_chars_bad"));
         assert!(!valid_username("path/name"));
+        assert!(!valid_username("upperCase"));
+        assert!(!valid_username("underscore_name"));
         assert!(!valid_username("line\nbreak"));
     }
 

web-ui/Cargo.toml

@@ -8,6 +8,9 @@ iota-storage = { path = "../iota-storage" }
 iota-state = { path = "../iota-state" }
 iota-util = { path = "../iota-util" }
 iota-logger = { path = "../iota-logger" }
+iota-cli = { path = "../iota-cli" }
+iota-ipc = { path = "../iota-ipc" }
+iota-paths = { path = "../iota-paths" }
 
 actix-web = { version = "4", features = ["rustls-0_23"] }
 rustls = { version = "0.23.37", features = ["aws-lc-rs"] }

web-ui/src/api.rs

@@ -1,5 +1,6 @@
-use crate::server::is_local_network;
 use actix_web::{HttpRequest, HttpResponse, Responder, web};
+use iota_ipc::{IpcErrorCode, LocalRequest, ResponsePayload, ResponseResult};
+use iota_paths::{Scope, socket_path};
 use iota_state::DaemonState;
 use iota_storage::util::config_util::{CONFIG, modify_config};
 use serde_json::{Value, json};
@@ -153,11 +154,27 @@ async fn users_add(
         _ => return error(),
     };
 
-    // The legacy web API is intentionally quarantined until it can use the
-    // daemon's authenticated command/service boundary. It must not create a
-    // second connector or mutate daemon storage directly.
-    let _ = username;
-    error()
+    let client = match iota_cli::ipc_client::IpcClient::connect(socket_path(Scope::User)).await {
+        Ok(client) => client,
+        Err(_) => return HttpResponse::ServiceUnavailable().json(json!({ "status": "not_ready" })),
+    };
+    match client
+        .send_request(LocalRequest::CreateUser {
+            username: username.to_string(),
+        })
+        .await
+    {
+        Ok(ResponseResult::Ok(ResponsePayload::UserCreated { user_id, username })) => {
+            HttpResponse::Created().json(json!({
+                "uuid": user_id,
+                "username": username,
+                "has_tu": true,
+            }))
+        }
+        Ok(ResponseResult::Ok(_)) => HttpResponse::Created().json(json!({ "status": "created" })),
+        Ok(ResponseResult::Error(code)) => ipc_error_response(code),
+        Err(_) => HttpResponse::GatewayTimeout().json(json!({ "status": "timeout" })),
+    }
 }
 
 async fn shutdown(
@@ -200,6 +217,26 @@ fn error() -> HttpResponse {
     HttpResponse::BadRequest().json(json!({ "type": "error" }))
 }
 
+fn ipc_error_response(code: IpcErrorCode) -> HttpResponse {
+    let status = match code {
+        IpcErrorCode::InvalidRequest => actix_web::http::StatusCode::BAD_REQUEST,
+        IpcErrorCode::Conflict => actix_web::http::StatusCode::CONFLICT,
+        IpcErrorCode::NotReady | IpcErrorCode::OmikronUnavailable => {
+            actix_web::http::StatusCode::SERVICE_UNAVAILABLE
+        }
+        IpcErrorCode::Timeout => actix_web::http::StatusCode::GATEWAY_TIMEOUT,
+        IpcErrorCode::Unauthorized => actix_web::http::StatusCode::FORBIDDEN,
+        IpcErrorCode::StorageFailure | IpcErrorCode::InternalFailure => {
+            actix_web::http::StatusCode::INTERNAL_SERVER_ERROR
+        }
+        IpcErrorCode::NotFound => actix_web::http::StatusCode::NOT_FOUND,
+        IpcErrorCode::UnsupportedVersion | IpcErrorCode::Disconnected | IpcErrorCode::Cancelled => {
+            actix_web::http::StatusCode::SERVICE_UNAVAILABLE
+        }
+    };
+    HttpResponse::build(status).json(json!({ "status": code.to_string() }))
+}
+
 fn user_id(payload: &Value) -> Result<i64, HttpResponse> {
     payload
         .get("uuid")
@@ -208,7 +245,8 @@ fn user_id(payload: &Value) -> Result<i64, HttpResponse> {
 }
 
 fn is_allowed(addr: SocketAddr, ssl: bool) -> bool {
-    is_local_network(addr.ip()) || ssl
+    let _ = ssl;
+    addr.ip().is_loopback()
 }
 
 fn is_allowed_req(req: &HttpRequest, ssl: bool) -> bool {

web-ui/src/server.rs

@@ -19,7 +19,7 @@ use tokio::sync::oneshot;
 pub async fn start(port: u16, state: Arc<DaemonState>) -> bool {
     let (tx, rx) = oneshot::channel::<ServerHandle>();
 
-    let bind_addr = std::env::var("BIND_ADDRESS").unwrap_or_else(|_| "0.0.0.0".to_string());
+    let bind_addr = std::env::var("BIND_ADDRESS").unwrap_or_else(|_| "127.0.0.1".to_string());
     let bind_ip: std::net::IpAddr = bind_addr.parse().expect("Invalid BIND_ADDRESS");
 
     let server_state = state.clone();