fix signal sessions after recovery

rust/src/api/messages/incoming/messages.rs

@@ -782,7 +782,7 @@ pub(crate) async fn queue_decryption_error(
     .encode_to_vec();
     NewReceipt::new(receipt_id, from_user_id, &response)
         .contact_will_send_receipt(false)
-        .insert_or_replace(tr)
+        .insert_if_absent(tr)
         .await?;
     Ok(())
 }
@@ -920,4 +920,56 @@ mod tests {
         let _held = FlushGuard::claim(&first).expect("nothing else holds the claim");
         assert!(FlushGuard::claim(&second).is_some());
     }
+
+    #[tokio::test]
+    async fn a_duplicate_decryption_failure_cannot_downgrade_a_queued_session_reset() {
+        let directory = tempfile::tempdir().unwrap();
+        let path = directory.path().join("app.sqlite");
+        let database = crate::database::app::AppDatabase::new(path.to_str().unwrap(), None, false)
+            .await
+            .unwrap();
+        database.run_migrations().await.unwrap();
+        sqlx::query("INSERT INTO contacts(user_id, username) VALUES (7, 'alice')")
+            .execute(&database.pool)
+            .await
+            .unwrap();
+
+        let mut transaction = database.pool.begin().await.unwrap();
+        queue_decryption_error(
+            &mut transaction,
+            7,
+            "failed-receipt",
+            proto::plaintext_content::decryption_error_message::Type::SessionResetRequired as i32,
+        )
+        .await
+        .unwrap();
+        // A redelivery reaches the reset limiter while the first response is
+        // still queued and falls back to Unknown. It must leave the stronger
+        // response untouched.
+        queue_decryption_error(
+            &mut transaction,
+            7,
+            "failed-receipt",
+            proto::plaintext_content::decryption_error_message::Type::Unknown as i32,
+        )
+        .await
+        .unwrap();
+        transaction.commit().await.unwrap();
+
+        let bytes = sqlx::query_scalar::<_, Vec<u8>>(
+            "SELECT message FROM receipts WHERE receipt_id = 'failed-receipt'",
+        )
+        .fetch_one(&database.pool)
+        .await
+        .unwrap();
+        let message = proto::Message::decode(bytes.as_slice()).unwrap();
+        let error = message
+            .plaintext_content
+            .and_then(|content| content.decryption_error_message)
+            .expect("the queued response is a decryption error");
+        assert_eq!(
+            error.r#type,
+            proto::plaintext_content::decryption_error_message::Type::SessionResetRequired as i32
+        );
+    }
 }

rust/src/backup/backup_archive.rs

@@ -321,6 +321,26 @@ impl BackupArchive {
         )
         .await?;
         rust_database.run_migrations().await?;
+
+        // A backup is a snapshot of a Double Ratchet at an earlier point in
+        // time. Once either peer has advanced past that snapshot, restoring
+        // its session record cannot make it current again and can make both
+        // sides keep encrypting against incompatible states. Start with no
+        // peer sessions instead. The regular V2 send path establishes them
+        // lazily from a current prekey bundle, while an old inbound ciphertext
+        // is answered by the existing SESSION_RESET_REQUIRED recovery flow.
+        //
+        // Reset throttles describe the discarded sessions, so retaining them
+        // could suppress the first legitimate repair after recovery.
+        let mut signal_transaction = rust_database.pool.begin().await?;
+        sqlx::query!("DELETE FROM signal_sessions")
+            .execute(&mut *signal_transaction)
+            .await?;
+        sqlx::query!("DELETE FROM signal_session_resets")
+            .execute(&mut *signal_transaction)
+            .await?;
+        signal_transaction.commit().await?;
+
         ctx.replace_rust_database(rust_database, &key_manager)
             .await?;
 
@@ -331,10 +351,10 @@ impl BackupArchive {
         // marks makes the next `on_connected` publish a signed prekey and a
         // fresh batch of PQC prekeys that this database actually holds.
         //
-        // Sessions restored alongside them are left as they are: they are only
+        // Peer sessions were discarded above. They are rebuilt lazily from
-        // broken for peers who ratcheted past this archive, and the first
+        // current bundles, avoiding use of ratchet state rewound by the
-        // message that fails to decrypt resets that peer's session on its own.
+        // archive. See `signal::reset` for recovery of messages already in
-        // See `signal::reset`.
+        // flight when the backup was restored.
         if let Err(error) = crate::user_config::UserConfig::update(ctx, |config| {
             config.signal_last_signed_pre_key_updated = None;
             config.signal_last_pqc_pre_keys_uploaded = None;
@@ -457,6 +477,37 @@ mod tests {
             .await
             .unwrap();
         }
+        {
+            let rust_db = ctx.rust_db.read().await.clone();
+            sqlx::query!(
+                r#"
+                INSERT INTO signal_identities(name, identity_key, timestamp)
+                VALUES('restored-peer', x'040506', 1)
+                "#
+            )
+            .execute(&rust_db.pool)
+            .await
+            .unwrap();
+            sqlx::query!(
+                r#"
+                INSERT INTO signal_sessions(name, device_id, record_bytes)
+                VALUES('restored-peer', 1, x'010203')
+                "#
+            )
+            .execute(&rust_db.pool)
+            .await
+            .unwrap();
+            sqlx::query!(
+                r#"
+                INSERT INTO signal_session_resets(
+                    name, device_id, last_reset_at, window_started_at, resets_in_window
+                ) VALUES('restored-peer', 1, 1, 1, 1)
+                "#
+            )
+            .execute(&rust_db.pool)
+            .await
+            .unwrap();
+        }
 
         // 2. Create backup
         let backup_path = BackupArchive::create_backup(&ctx).await.unwrap();
@@ -510,6 +561,37 @@ mod tests {
             .await
             .unwrap();
             assert_eq!(username, "original contact");
+
+            let rust_db = ctx.rust_db.read().await.clone();
+            let session_count =
+                sqlx::query_scalar!(r#"SELECT COUNT(*) AS "count!: i64" FROM signal_sessions"#)
+                    .fetch_one(&rust_db.pool)
+                    .await
+                    .unwrap();
+            assert_eq!(session_count, 0, "restored peer sessions must be discarded");
+
+            let reset_count = sqlx::query_scalar!(
+                r#"SELECT COUNT(*) AS "count!: i64" FROM signal_session_resets"#
+            )
+            .fetch_one(&rust_db.pool)
+            .await
+            .unwrap();
+            assert_eq!(
+                reset_count, 0,
+                "reset throttles for discarded sessions must be cleared"
+            );
+
+            let identity_key = sqlx::query_scalar!(
+                "SELECT identity_key FROM signal_identities WHERE name = 'restored-peer'"
+            )
+            .fetch_one(&rust_db.pool)
+            .await
+            .unwrap();
+            assert_eq!(
+                identity_key,
+                vec![4, 5, 6],
+                "a contact's trusted identity key must survive archive recovery"
+            );
         }
     }
 

rust/src/database/app/tables/receipt.rs

@@ -248,11 +248,15 @@ impl<'a> NewReceipt<'a> {
         Ok(())
     }
 
-    pub async fn insert_or_replace(&self, transaction: &mut Transaction<'_, Sqlite>) -> Result<()> {
+    /// Queues a response only while no response for this inbound receipt is in
+    /// flight. A redelivery can race the sender task, but it must not replace
+    /// the response that task is about to send.
+    pub async fn insert_if_absent(&self, transaction: &mut Transaction<'_, Sqlite>) -> Result<()> {
         sqlx::query!(
             r#"
-            INSERT OR REPLACE INTO receipts(receipt_id, contact_id, message, contact_will_sends_receipt, wake_receiver)
+            INSERT INTO receipts(receipt_id, contact_id, message, contact_will_sends_receipt, wake_receiver)
             VALUES (?, ?, ?, ?, ?)
+            ON CONFLICT(receipt_id) DO NOTHING
             "#,
             self.receipt_id,
             self.contact_id,

rust/tests/api/media.rs

@@ -78,7 +78,11 @@ async fn test_media_lifecycle_actions_and_reupload() -> anyhow::Result<()> {
         .join("mediafiles/tmp")
         .join(format!("{local_media_id}.webp"));
     std::fs::create_dir_all(plaintext_path.parent().expect("tmp directory"))?;
-    std::fs::write(&plaintext_path, b"plaintext media bytes")?;
+    let source = image::RgbImage::from_pixel(8, 6, image::Rgb([12, 34, 56]));
+    let plaintext = webp::Encoder::from_rgb(source.as_raw(), source.width(), source.height())
+        .encode_simple(false, 80.0)
+        .map_err(|error| anyhow::anyhow!("could not encode media fixture: {error:?}"))?;
+    std::fs::write(&plaintext_path, plaintext.as_ref())?;
 
     let media_content = proto::EncryptedContent {
         group_id: Some(group_id.clone()),

rust/tests/api/session_recovery.rs

@@ -71,6 +71,70 @@ async fn test_signal_session_auto_recovery_on_missing_session() -> anyhow::Resul
     Ok(())
 }
 
+/// Archive recovery deliberately discards the receiver's restored sessions.
+/// A sender may still have its pre-recovery half and send one stale ciphertext;
+/// the receiver must request a reset and receive the re-encrypted message over
+/// the fresh prekey session without user intervention.
+#[tokio::test]
+async fn test_signal_session_recovers_after_receiver_sessions_are_purged() -> anyhow::Result<()> {
+    init_tracing();
+    let tester_a = create_authenticated_tester().await?;
+    let tester_b = create_authenticated_tester().await?;
+
+    ContactService::new(&tester_a.context)
+        .request_by_username(tester_b.username.clone(), true)
+        .await?;
+    tester_b
+        .wait_for_contact_state(tester_a.user_id, false, true)
+        .await?;
+    ContactService::new(&tester_b.context)
+        .accept_request(tester_a.user_id, true)
+        .await?;
+    tester_a
+        .wait_for_contact_state(tester_b.user_id, true, false)
+        .await?;
+
+    let group_id = Group::direct_chat_id(tester_a.user_id, tester_b.user_id);
+    let initial_id = MessageService::new(&tester_a.context)
+        .insert_and_send_text(group_id.clone(), "Before session purge".into(), None, None)
+        .await?;
+    tester_b
+        .wait_for_text_message(&initial_id, tester_a.user_id, "Before session purge")
+        .await?;
+
+    // This is the Signal-state result of restoring an archive: B retains its
+    // identity and prekeys, but no peer session survives.
+    let rust_db_b = tester_b.context.rust_db.read().await.clone();
+    sqlx::query!("DELETE FROM signal_sessions")
+        .execute(&rust_db_b.pool)
+        .await?;
+    sqlx::query!("DELETE FROM signal_session_resets")
+        .execute(&rust_db_b.pool)
+        .await?;
+
+    // A still encrypts with the old half. B reports that it cannot use that
+    // session, A rebuilds from B's current bundle, and the queued plaintext is
+    // re-encrypted and delivered over a PreKey message.
+    let recovered_id = MessageService::new(&tester_a.context)
+        .insert_and_send_text(group_id.clone(), "After session purge".into(), None, None)
+        .await?;
+    tester_b
+        .wait_for_text_message(&recovered_id, tester_a.user_id, "After session purge")
+        .await?;
+
+    // The newly established session works in the restored user's direction as
+    // well, proving that the repair converged rather than only delivering the
+    // retried message.
+    let reply_id = MessageService::new(&tester_b.context)
+        .insert_and_send_text(group_id, "Reply after session purge".into(), None, None)
+        .await?;
+    tester_a
+        .wait_for_text_message(&reply_id, tester_b.user_id, "Reply after session purge")
+        .await?;
+
+    Ok(())
+}
+
 /// A restored backup rewinds one side's ratchet behind the peer's, so nothing
 /// the peer sends afterwards decrypts and resending it never helps. The
 /// receiver has to retire its session, the sender has to build a new one from