[Fix] client state synchronization, shared data

Cargo.lock

@@ -174,7 +174,7 @@ dependencies = [
  "foldhash",
  "futures-core",
  "futures-util",
- "impl-more 0.3.5",
+ "impl-more 0.3.7",
  "itoa",
  "language-tags",
  "log",
@@ -866,6 +866,12 @@ version = "0.8.7"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "773648b94d0e5d620f64f280777445740e61fe701025087ec8b57f45c791888b"
 
+[[package]]
+name = "core_detect"
+version = "1.0.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "7f8f80099a98041a3d1622845c271458a2d73e688351bf3cb999266764b81d48"
+
 [[package]]
 name = "cpufeatures"
 version = "0.2.17"
@@ -1278,11 +1284,17 @@ checksum = "252afb9ae5eaa683babdc6a068b3f5726eb19e05070c731f9b2a23a7c3e8ed34"
 
 [[package]]
 name = "encoding_rs"
-version = "0.8.35"
+version = "0.8.40"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "75030f3c4f45dafd7586dd6780965a8c7e8e285a5ecb86713e63a79c5b2766f3"
+checksum = "2a7a45518d2863d18aa47f4a0cf9faec2aa4304cc09df5e41299f276b3ad135e"
 dependencies = [
  "cfg-if",
+ "core_detect",
+ "multiversion",
+ "multiversion_no_op",
+ "rustversion",
+ "scopeguard",
+ "simdutf8",
 ]
 
 [[package]]
@@ -1696,9 +1708,9 @@ dependencies = [
 
 [[package]]
 name = "hashlink"
-version = "0.12.1"
+version = "0.12.2"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "32069d97bb81e38fa67eab65e3393bf804bb85969f2bc06bf13f64aef5aba248"
+checksum = "a596f1b20ed2cc5ecac41a164aaebc7258057060f06c0cf7a2ba3991ee7990fb"
 dependencies = [
  "hashbrown 0.17.1",
 ]
@@ -2018,9 +2030,9 @@ checksum = "e8a5a9a0ff0086c7a148acb942baaabeadf9504d10400b5a05645853729b9cd2"
 
 [[package]]
 name = "impl-more"
-version = "0.3.5"
+version = "0.3.7"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "277ff51754a3f68f12f58446c5d006aa8baa4914ea273cce24a599cfaff33d4f"
+checksum = "edaff2ce006342d4d0e00fae676f7082dada44a203560629ba18ea50a19277bb"
 
 [[package]]
 name = "indexmap"
@@ -2896,6 +2908,34 @@ dependencies = [
  "tracing",
 ]
 
+[[package]]
+name = "multiversion"
+version = "0.8.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "7edb7f0ff51249dfda9ab96b5823695e15a052dc15074c9dbf3d118afaf2c201"
+dependencies = [
+ "multiversion-macros",
+ "target-features",
+]
+
+[[package]]
+name = "multiversion-macros"
+version = "0.8.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "b093064383341eb3271f42e381cb8f10a01459478446953953c75d24bd339fc0"
+dependencies = [
+ "proc-macro2",
+ "quote",
+ "syn 2.0.119",
+ "target-features",
+]
+
+[[package]]
+name = "multiversion_no_op"
+version = "1.0.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "743fb55ba31b18fb1ecef6bdc9aa2743314978ac084044301a7eee33fb99a20d"
+
 [[package]]
 name = "nix"
 version = "0.29.0"
@@ -3009,9 +3049,9 @@ dependencies = [
 
 [[package]]
 name = "octets"
-version = "0.3.6"
+version = "0.3.7"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "866cb5af6f3aa3c1b44c3c2d79d22165fbb1b102e1b3fb499864bfe34736ec4b"
+checksum = "4d59d30d3ad7f7e1c9a66e51c9cb71ae1c794b28ffe7053fd785defb0d180069"
 
 [[package]]
 name = "oid-registry"
@@ -3899,9 +3939,9 @@ dependencies = [
 
 [[package]]
 name = "rustls"
-version = "0.23.43"
+version = "0.23.44"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "0283386ce02abc0151e1761d08802dfe86c173b0b494af5cbc086574e453da06"
+checksum = "6725596c3f2c3a0aef021139e145d4eafe314a6623e4680ca83852b2c67ab2ba"
 dependencies = [
  "aws-lc-rs",
  "log",
@@ -4467,6 +4507,12 @@ dependencies = [
  "libc",
 ]
 
+[[package]]
+name = "target-features"
+version = "0.1.6"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "c1bbb9f3c5c463a01705937a24fdabc5047929ac764b2d5b9cf681c1f5041ed5"
+
 [[package]]
 name = "tempfile"
 version = "3.27.0"

client/src/client_connection.rs

@@ -210,15 +210,21 @@ impl ClientConnection {
             return;
         }
 
-        if cv.is_type(CommunicationType::ClientConnected) {
+        if cv.is_type(CommunicationType::ClientStateGet) {
-            self.send_message(&message_handlers::handle_client_connected(&cv))
+            // Durable client synchronization is routed through Omikron, which
-                .await;
+            // is the authority that assigns the numeric transport SessionId.
+            self.send_message(
+                &CommunicationValue::new(CommunicationType::ErrorInvalidData).with_request_id(&cv),
+            )
+            .await;
             return;
         }
 
         if cv.is_type(CommunicationType::ClientStateAck) {
-            self.send_message(&message_handlers::handle_client_state_ack(&cv))
+            self.send_message(
-                .await;
+                &CommunicationValue::new(CommunicationType::ErrorInvalidData).with_request_id(&cv),
+            )
+            .await;
             return;
         }
 

iota-connection/src/message_handlers.rs

@@ -17,6 +17,17 @@ use std::sync::atomic::{AtomicU32, Ordering};
 
 static NEXT_NOTIFICATION_ID: AtomicU32 = AtomicU32::new(1);
 
+fn valid_device_id(id: &str) -> bool {
+    id.len() == 36
+        && id
+            .chars()
+            .enumerate()
+            .all(|(index, character)| match index {
+                8 | 13 | 18 | 23 => character == '-',
+                _ => character.is_ascii_hexdigit(),
+            })
+}
+
 fn next_notification_id() -> u32 {
     NEXT_NOTIFICATION_ID.fetch_add(1, Ordering::Relaxed).max(1)
 }
@@ -952,7 +963,7 @@ fn sync_error(cv: &CommunicationValue) -> CommunicationValue {
 }
 
 /// The sender is authenticated by MTP; a UserId embedded by a client is never trusted here.
-pub fn handle_client_connected(cv: &CommunicationValue) -> CommunicationValue {
+pub fn handle_client_state_get(cv: &CommunicationValue) -> CommunicationValue {
     use iota_storage::util::sync::{self, CACHE_SCHEMA_VERSION};
     let user_id = match required_sender_id(cv) {
         Ok(id) if id > 0 => id,
@@ -962,21 +973,31 @@ pub fn handle_client_connected(cv: &CommunicationValue) -> CommunicationValue {
         Some(id) if id > 0 => id,
         _ => return sync_error(cv),
     };
+    let device_id = match cv.get_data(DataType::DeviceId).and_then(DataValue::as_str) {
+        Some(id) if valid_device_id(id) => id.to_owned(),
+        _ => return sync_error(cv),
+    };
     let reported_version = match data_i64(cv, DataType::VersionNumber) {
         Some(version) if version >= 0 => version,
         _ => return sync_error(cv),
     };
-    let cache_valid = cv.get_data(DataType::CacheValid).as_bool().unwrap_or(false);
+    let cache_valid = match cv.get_data(DataType::CacheValid).as_bool() {
-    let schema = data_i64(cv, DataType::CacheSchemaVersion).unwrap_or(0);
+        Some(value) => value,
+        None => return sync_error(cv),
+    };
+    let schema = match data_i64(cv, DataType::CacheSchemaVersion) {
+        Some(value) if value >= 0 => value,
+        _ => return sync_error(cv),
+    };
     let head = match sync::head(user_id) {
         Ok(version) => version,
         Err(_) => return sync_error(cv),
     };
-    let known_session = sync::has_session(user_id, session_id).unwrap_or(false);
+    let known_device = sync::has_device(user_id, &device_id).unwrap_or(false);
-    let acknowledged_version = sync::acknowledged_version(user_id, session_id).unwrap_or(None);
+    let acknowledged_version = sync::acknowledged_version(user_id, &device_id).unwrap_or(None);
     let full = !cache_valid
         || reported_version == 0
-        || !known_session
+        || !known_device
         || acknowledged_version.is_some_and(|version| reported_version < version)
         || reported_version > head
         || schema != CACHE_SCHEMA_VERSION;
@@ -1181,12 +1202,17 @@ pub fn handle_client_state_ack(cv: &CommunicationValue) -> CommunicationValue {
         Some(id) if id > 0 => id,
         _ => return sync_error(cv),
     };
+    let device_id = match cv.get_data(DataType::DeviceId).and_then(DataValue::as_str) {
+        Some(id) if valid_device_id(id) => id.to_owned(),
+        _ => return sync_error(cv),
+    };
     let version = match data_i64(cv, DataType::VersionNumber) {
         Some(version) if version >= 0 => version,
         _ => return sync_error(cv),
     };
     if iota_storage::util::client_message_delivery::acknowledge_client_state(
         user_id,
+        &device_id,
         session_id,
         version,
         CACHE_SCHEMA_VERSION,
@@ -1874,7 +1900,24 @@ fn blob_mutation_response(
         )
 }
 
-fn blob_changed(user_id: i64, blob_id: String, revision: i64, deleted: bool) -> CommunicationValue {
+fn blob_changed(blob: &UserBlob) -> CommunicationValue {
+    CommunicationValue::new(CommunicationType::UserBlobChanged)
+        .with_id(next_notification_id())
+        .with_receiver(sender_wire_id(blob.user_id))
+        .add_typed_default(DataType::BlobId, DataValue::Str(blob.blob_id.clone()))
+        .add_typed_default(DataType::Blob, DataValue::Bytes(blob.blob.clone()))
+        .add_typed_default(
+            DataType::VersionNumber,
+            DataValue::SignedNumber(blob.revision.into()),
+        )
+        .add_typed_default(
+            DataType::UpdatedAt,
+            DataValue::SignedNumber(blob.updated_at.into()),
+        )
+        .add_typed_default(DataType::Deleted, DataValue::Bool(false))
+}
+
+fn blob_deleted(user_id: i64, blob_id: String, revision: i64) -> CommunicationValue {
     CommunicationValue::new(CommunicationType::UserBlobChanged)
         .with_id(next_notification_id())
         .with_receiver(sender_wire_id(user_id))
@@ -1883,7 +1926,7 @@ fn blob_changed(user_id: i64, blob_id: String, revision: i64, deleted: bool) ->
             DataType::VersionNumber,
             DataValue::SignedNumber(revision.into()),
         )
-        .add_typed_default(DataType::Deleted, DataValue::Bool(deleted))
+        .add_typed_default(DataType::Deleted, DataValue::Bool(true))
 }
 
 fn blob_request(cv: &CommunicationValue) -> Result<(i64, String), CommunicationValue> {
@@ -2136,12 +2179,7 @@ pub fn handle_user_blob_put(cv: &CommunicationValue) -> BlobMutation {
     match user_blobs::put(user_id, &blob_id, &blob, expected_revision) {
         Ok(stored) => BlobMutation {
             response: blob_mutation_response(cv, CommunicationType::UserBlobPut, &stored),
-            changed: Some(blob_changed(
+            changed: Some(blob_changed(&stored)),
-                user_id,
-                stored.blob_id.clone(),
-                stored.revision,
-                false,
-            )),
         },
         Err(iota_storage::storage_error::StorageError::RevisionConflict) => BlobMutation {
             response: error_response(cv, CommunicationType::ErrorInvalidData),
@@ -2184,7 +2222,7 @@ pub fn handle_user_blob_delete(cv: &CommunicationValue) -> BlobMutation {
                 ),
             changed: deleted
                 .changed
-                .then(|| blob_changed(user_id, blob_id, deleted.revision, true)),
+                .then(|| blob_deleted(user_id, blob_id, deleted.revision)),
         },
         Ok(None) => BlobMutation {
             response: CommunicationValue::new(CommunicationType::UserBlobDelete)

iota-storage/src/users/user_manager.rs

@@ -450,6 +450,10 @@ fn purge_database_rows(user_id: i64) -> Result<(), crate::storage_error::Storage
             "DELETE FROM client_sync_state WHERE user_id = ?1",
             params![user_id],
         )?;
+        tx.execute(
+            "DELETE FROM client_device_sync_state WHERE user_id = ?1",
+            params![user_id],
+        )?;
         tx.execute(
             "DELETE FROM client_message_deliveries WHERE user_id = ?1",
             params![user_id],

iota-storage/src/util/client_message_delivery.rs

@@ -41,11 +41,12 @@ pub fn acknowledge_sync_delivery(
 /* Acknowledge a state-sync response and delete only its recorded visible messages atomically. */
 pub fn acknowledge_client_state(
     user_id: i64,
+    device_id: &str,
     session_id: i64,
     version: i64,
     cache_schema_version: i64,
 ) -> Result<(), StorageError> {
-    if user_id <= 0 || session_id <= 0 || version < 0 {
+    if user_id <= 0 || device_id.is_empty() || session_id <= 0 || version < 0 {
         return Err(StorageError::Other("invalid sync acknowledgement".into()));
     }
     db::with_immediate_transaction(|tx| {
@@ -70,7 +71,7 @@ pub fn acknowledge_client_state(
                 })?
                 .collect::<Result<Vec<_>, _>>()?
         };
-        tx.execute("INSERT INTO client_sync_state (user_id, session_id, acknowledged_version, cache_schema_version, updated_at) VALUES (?1, ?2, ?3, ?4, ?5) ON CONFLICT(user_id, session_id) DO UPDATE SET acknowledged_version = MAX(acknowledged_version, excluded.acknowledged_version), cache_schema_version = excluded.cache_schema_version, updated_at = excluded.updated_at", params![user_id, session_id, version, cache_schema_version, sync::now_millis()])?;
+        tx.execute("INSERT INTO client_device_sync_state (user_id, device_id, acknowledged_version, cache_schema_version, updated_at) VALUES (?1, ?2, ?3, ?4, ?5) ON CONFLICT(user_id, device_id) DO UPDATE SET acknowledged_version = MAX(acknowledged_version, excluded.acknowledged_version), cache_schema_version = excluded.cache_schema_version, updated_at = excluded.updated_at", params![user_id, device_id, version, cache_schema_version, sync::now_millis()])?;
         if message_storage_policy::get_in_tx(tx, user_id)?.history_mode
             == message_storage_policy::MessageHistoryMode::DeleteAfterClientDelivery
         {

iota-storage/src/util/db.rs

@@ -887,6 +887,22 @@ fn run_migrations_on_connection(conn: &Connection) -> Result<(), StorageError> {
         conn.pragma_update(None, "user_version", 25)?;
     }
 
+    if current_version < 26 {
+        conn.execute_batch(
+            r#"
+            CREATE TABLE IF NOT EXISTS client_device_sync_state (
+                user_id INTEGER NOT NULL,
+                device_id TEXT NOT NULL,
+                acknowledged_version INTEGER NOT NULL,
+                cache_schema_version INTEGER NOT NULL,
+                updated_at INTEGER NOT NULL,
+                PRIMARY KEY (user_id, device_id)
+            );
+            "#,
+        )?;
+        conn.pragma_update(None, "user_version", 26)?;
+    }
+
     Ok(())
 }
 
@@ -956,7 +972,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, 25);
+        assert_eq!(version, 26);
         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")?;
@@ -975,11 +991,12 @@ 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, 25);
+        assert_eq!(version, 26);
         for table in [
             "sync_heads",
             "sync_events",
             "client_sync_state",
+            "client_device_sync_state",
             "user_residency",
             "relay_replay",
             "pending_relays",
@@ -1021,7 +1038,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, 25);
+        assert_eq!(version, 26);
         for column in [
             "id",
             "user_id",
@@ -1116,7 +1133,7 @@ mod tests {
         )?;
         let version: i64 = conn.pragma_query_value(None, "user_version", |row| row.get(0))?;
         assert_eq!(preserved, "remote_committed");
-        assert_eq!(version, 25);
+        assert_eq!(version, 26);
         Ok(())
     }
 }

iota-storage/src/util/sync.rs

@@ -114,23 +114,23 @@ pub fn head(user_id: i64) -> Result<i64, StorageError> {
     })
 }
 
-pub fn has_session(user_id: i64, session_id: i64) -> Result<bool, StorageError> {
+pub fn has_device(user_id: i64, device_id: &str) -> Result<bool, StorageError> {
     db::with_db(|conn| {
         Ok(conn
             .query_row(
-                "SELECT 1 FROM client_sync_state WHERE user_id = ?1 AND session_id = ?2",
+                "SELECT 1 FROM client_device_sync_state WHERE user_id = ?1 AND device_id = ?2",
-                params![user_id, session_id],
+                params![user_id, device_id],
                 |_| Ok(()),
             )
             .is_ok())
     })
 }
 
-pub fn acknowledged_version(user_id: i64, session_id: i64) -> Result<Option<i64>, StorageError> {
+pub fn acknowledged_version(user_id: i64, device_id: &str) -> Result<Option<i64>, StorageError> {
     db::with_db(|conn| {
         conn.query_row(
-            "SELECT acknowledged_version FROM client_sync_state WHERE user_id = ?1 AND session_id = ?2",
+            "SELECT acknowledged_version FROM client_device_sync_state WHERE user_id = ?1 AND device_id = ?2",
-            params![user_id, session_id],
+            params![user_id, device_id],
             |row| row.get(0),
         )
         .optional()
@@ -138,33 +138,6 @@ pub fn acknowledged_version(user_id: i64, session_id: i64) -> Result<Option<i64>
     })
 }
 
-pub fn acknowledge(
-    user_id: i64,
-    session_id: i64,
-    version: i64,
-    cache_schema_version: i64,
-) -> Result<(), StorageError> {
-    if user_id <= 0 || session_id <= 0 || version < 0 {
-        return Err(StorageError::Other("invalid sync acknowledgement".into()));
-    }
-    db::with_db(|conn| {
-        let head = conn
-            .query_row(
-                "SELECT version FROM sync_heads WHERE user_id = ?1",
-                [user_id],
-                |r| r.get(0),
-            )
-            .unwrap_or(0);
-        if version > head {
-            return Err(StorageError::Other(
-                "acknowledgement is ahead of head".into(),
-            ));
-        }
-        conn.execute("INSERT INTO client_sync_state (user_id, session_id, acknowledged_version, cache_schema_version, updated_at) VALUES (?1, ?2, ?3, ?4, ?5) ON CONFLICT(user_id, session_id) DO UPDATE SET acknowledged_version = MAX(acknowledged_version, excluded.acknowledged_version), cache_schema_version = excluded.cache_schema_version, updated_at = excluded.updated_at", params![user_id, session_id, version, cache_schema_version, now_millis()])?;
-        Ok(())
-    })
-}
-
 /* Retain deltas until every non-stale client has acknowledged them. */
 pub fn compact_user_sync_state(user_id: i64) -> Result<SyncCompactionResult, StorageError> {
     if user_id <= 0 {
@@ -173,15 +146,15 @@ pub fn compact_user_sync_state(user_id: i64) -> Result<SyncCompactionResult, Sto
     let stale_before = now_millis().saturating_sub(STALE_CLIENT_SYNC_STATE_MS);
     db::with_immediate_transaction(|tx| {
         tx.execute(
-            "DELETE FROM client_message_deliveries WHERE user_id = ?1 AND session_id IN (SELECT session_id FROM client_sync_state WHERE user_id = ?1 AND updated_at < ?2)",
+            "DELETE FROM client_message_deliveries WHERE user_id = ?1 AND created_at < ?2",
             params![user_id, stale_before],
         )?;
         let removed_stale_clients = tx.execute(
-            "DELETE FROM client_sync_state WHERE user_id = ?1 AND updated_at < ?2",
+            "DELETE FROM client_device_sync_state WHERE user_id = ?1 AND updated_at < ?2",
             params![user_id, stale_before],
         )?;
         let minimum_acknowledged: Option<i64> = tx.query_row(
-            "SELECT MIN(acknowledged_version) FROM client_sync_state WHERE user_id = ?1",
+            "SELECT MIN(acknowledged_version) FROM client_device_sync_state WHERE user_id = ?1",
             [user_id],
             |row| row.get(0),
         )?;

mtp-type-maps

@@ -1 +1 @@
-Subproject commit 154096cd198d6ccd0663e5e701643ebc859fafab
+Subproject commit 2de3be6016410304d6505d12ef37c35e63a9c731

omikron-connector/src/omikron_connection.rs

@@ -2321,7 +2321,7 @@ impl OmikronConnection {
         dispatch!(LoadAppData, handle_load_app_data);
         dispatch!(CreateApp, handle_create_app);
         dispatch!(DeleteApp, handle_delete_app);
-        dispatch!(ClientConnected, handle_client_connected);
+        dispatch!(ClientStateGet, handle_client_state_get);
         dispatch!(ClientStateAck, handle_client_state_ack);
         dispatch!(ReadNotification, handle_read_notification);
         dispatch!(MessageEdit, handle_message_edit);
@@ -2622,8 +2622,8 @@ impl OmikronConnection {
             .await;
     }
 
-    async fn handle_client_connected(self: Arc<Self>, cv: &CommunicationValue) {
+    async fn handle_client_state_get(self: Arc<Self>, cv: &CommunicationValue) {
-        let response = message_handlers::handle_client_connected(cv);
+        let response = message_handlers::handle_client_state_get(cv);
         let user_id = cv
             .require_sender()
             .ok()
@@ -2663,7 +2663,7 @@ impl OmikronConnection {
         }
 
         if let Err(error) = self.send_message(&response).await {
-            log!("ClientConnected response delivery failed: {}", error);
+            log!("ClientStateGet response delivery failed: {}", error);
         }
     }