feat: compress backups with gzip

join() method of Gossip [1]
waits for at least one connection
and this is not what we want
because it may block receive_imf()
forever if no connection arrives.

[1] https://docs.rs/iroh-gossip/0.25.0/iroh_gossip/net/struct.Gossip.html#method.join

CHANGELOG.md

@@ -1,5 +1,42 @@
 # Changelog
 
+## [1.148.4] - 2024-10-24
+
+### Features / Changes
+
+- Jsonrpc: add `private_tag` to `Account::Configured` Object ([#6107](https://github.com/deltachat/deltachat-core-rust/pull/6107)).
+
+### Fixes
+
+- Normalize proxy URLs before saving into proxy_url.
+- Do not wait for connections in maybe_add_gossip_peers().
+
+## [1.148.3] - 2024-10-24
+
+### Fixes
+
+- Fix reception of realtime advertisements.
+
+### Features / Changes
+
+- Allow sending realtime messages up to 128 KB in size.
+
+### API-Changes
+
+- deltachat-rpc-client: Add EventType.WEBXDC_REALTIME_ADVERTISEMENT_RECEIVED.
+
+### Documentation
+
+- Fix DC_QR_PROXY docs ([#6099](https://github.com/deltachat/deltachat-core-rust/pull/6099)).
+
+### Refactor
+
+- Generate topic inside create_iroh_header().
+
+### Tests
+
+- Test that realtime advertisements work after chatting.
+
 ## [1.148.2] - 2024-10-23
 
 ### Fixes
@@ -5100,3 +5137,5 @@ https://github.com/deltachat/deltachat-core-rust/pulls?q=is%3Apr+is%3Aclosed
 [1.148.0]: https://github.com/deltachat/deltachat-core-rust/compare/v1.147.1..v1.148.0
 [1.148.1]: https://github.com/deltachat/deltachat-core-rust/compare/v1.148.0..v1.148.1
 [1.148.2]: https://github.com/deltachat/deltachat-core-rust/compare/v1.148.1..v1.148.2
+[1.148.3]: https://github.com/deltachat/deltachat-core-rust/compare/v1.148.2..v1.148.3
+[1.148.4]: https://github.com/deltachat/deltachat-core-rust/compare/v1.148.3..v1.148.4

Cargo.lock

@@ -1293,11 +1293,12 @@ dependencies = [
 
 [[package]]
 name = "deltachat"
-version = "1.148.2"
+version = "1.148.4"
 dependencies = [
  "anyhow",
  "async-broadcast",
  "async-channel 2.3.1",
+ "async-compression",
  "async-imap",
  "async-native-tls",
  "async-smtp",
@@ -1393,7 +1394,7 @@ dependencies = [
 
 [[package]]
 name = "deltachat-jsonrpc"
-version = "1.148.2"
+version = "1.148.4"
 dependencies = [
  "anyhow",
  "async-channel 2.3.1",
@@ -1418,7 +1419,7 @@ dependencies = [
 
 [[package]]
 name = "deltachat-repl"
-version = "1.148.2"
+version = "1.148.4"
 dependencies = [
  "anyhow",
  "deltachat",
@@ -1434,7 +1435,7 @@ dependencies = [
 
 [[package]]
 name = "deltachat-rpc-server"
-version = "1.148.2"
+version = "1.148.4"
 dependencies = [
  "anyhow",
  "deltachat",
@@ -1463,7 +1464,7 @@ dependencies = [
 
 [[package]]
 name = "deltachat_ffi"
-version = "1.148.2"
+version = "1.148.4"
 dependencies = [
  "anyhow",
  "deltachat",

Cargo.toml

@@ -1,6 +1,6 @@
 [package]
 name = "deltachat"
-version = "1.148.2"
+version = "1.148.4"
 edition = "2021"
 license = "MPL-2.0"
 rust-version = "1.77"
@@ -42,6 +42,7 @@ anyhow = { workspace = true }
 async-broadcast = "0.7.1"
 async-channel = { workspace = true }
 async-imap = { version = "0.10.2", default-features = false, features = ["runtime-tokio", "compress"] }
+async-compression = { version = "0.4.15", default-features = false, features = ["tokio", "gzip"] }
 async-native-tls = { version = "0.5", default-features = false, features = ["runtime-tokio"] }
 async-smtp = { version = "0.9", default-features = false, features = ["runtime-tokio"] }
 async_zip = { version = "0.0.17", default-features = false, features = ["deflate", "tokio-fs"] }

deltachat-ffi/Cargo.toml

@@ -1,6 +1,6 @@
 [package]
 name = "deltachat_ffi"
-version = "1.148.2"
+version = "1.148.4"
 description = "Deltachat FFI"
 edition = "2018"
 readme = "README.md"

deltachat-ffi/deltachat.h

@@ -2533,8 +2533,8 @@ void            dc_stop_ongoing_process      (dc_context_t* context);
  *   ask the user if they want to use the given service for video chats;
  *   if so, call dc_set_config_from_qr().
  *
- * - DC_QR_SOCKS5_PROXY with dc_lot_t::text1=host, dc_lot_t::text2=port:
- *   ask the user if they want to use the given proxy and overwrite the previous one, if any.
+ * - DC_QR_PROXY with dc_lot_t::text1=address:
+ *   ask the user if they want to use the given proxy.
  *   if so, call dc_set_config_from_qr() and restart I/O.
  *
  * - DC_QR_ADDR with dc_lot_t::id=Contact ID:

deltachat-jsonrpc/Cargo.toml

@@ -1,6 +1,6 @@
 [package]
 name = "deltachat-jsonrpc"
-version = "1.148.2"
+version = "1.148.4"
 description = "DeltaChat JSON-RPC API"
 edition = "2021"
 default-run = "deltachat-jsonrpc-server"

deltachat-jsonrpc/src/api/types/account.rs

@@ -17,6 +17,9 @@ pub enum Account {
         // size: u32,
         profile_image: Option<String>, // TODO: This needs to be converted to work with blob http server.
         color: String,
+        /// Optional tag as "Work", "Family".
+        /// Meant to help profile owner to differ between profiles with similar names.
+        private_tag: Option<String>,
     },
     #[serde(rename_all = "camelCase")]
     Unconfigured { id: u32 },
@@ -31,12 +34,14 @@ impl Account {
             let color = color_int_to_hex_string(
                 Contact::get_by_id(ctx, ContactId::SELF).await?.get_color(),
             );
+            let private_tag = ctx.get_config(Config::PrivateTag).await?;
             Ok(Account::Configured {
                 id,
                 display_name,
                 addr,
                 profile_image,
                 color,
+                private_tag,
             })
         } else {
             Ok(Account::Unconfigured { id })

deltachat-jsonrpc/typescript/package.json

@@ -58,5 +58,5 @@
   },
   "type": "module",
   "types": "dist/deltachat.d.ts",
-  "version": "1.148.2"
+  "version": "1.148.4"
 }

deltachat-repl/Cargo.toml

@@ -1,6 +1,6 @@
 [package]
 name = "deltachat-repl"
-version = "1.148.2"
+version = "1.148.4"
 license = "MPL-2.0"
 edition = "2021"
 repository = "https://github.com/deltachat/deltachat-core-rust"

deltachat-rpc-client/pyproject.toml

@@ -4,7 +4,7 @@ build-backend = "setuptools.build_meta"
 
 [project]
 name = "deltachat-rpc-client"
-version = "1.148.2"
+version = "1.148.4"
 description = "Python client for Delta Chat core JSON-RPC interface"
 classifiers = [
     "Development Status :: 5 - Production/Stable",

deltachat-rpc-client/src/deltachat_rpc_client/const.py

@@ -63,6 +63,7 @@ class EventType(str, Enum):
     CHATLIST_ITEM_CHANGED = "ChatlistItemChanged"
     CONFIG_SYNCED = "ConfigSynced"
     WEBXDC_REALTIME_DATA = "WebxdcRealtimeData"
+    WEBXDC_REALTIME_ADVERTISEMENT_RECEIVED = "WebxdcRealtimeAdvertisementReceived"
 
 
 class ChatId(IntEnum):

deltachat-rpc-client/tests/test_iroh_webxdc.py

@@ -7,6 +7,7 @@ If you want to debug iroh at rust-trace/log level set
     RUST_LOG=iroh_net=trace,iroh_gossip=trace
 """
 
+import os
 import sys
 import threading
 import time
@@ -107,13 +108,15 @@ def test_realtime_sequentially(acfactory, path_to_webxdc):
     assert snapshot.text == "ping2"
 
     log("sending realtime data ac1 -> ac2")
-    ac1_webxdc_msg.send_webxdc_realtime_data(b"foo")
+    # Test that 128 KB of data can be sent in a single message.
+    data = os.urandom(128000)
+    ac1_webxdc_msg.send_webxdc_realtime_data(data)
 
     log("ac2: waiting for realtime data")
     while 1:
         event = ac2.wait_for_event()
         if event.kind == EventType.WEBXDC_REALTIME_DATA:
-            assert event.data == list(b"foo")
+            assert event.data == list(data)
             break
 
 
@@ -208,3 +211,28 @@ def test_no_reordering(acfactory, path_to_webxdc):
                 if event.data[0] == i:
                     break
                 pytest.fail("Reordering detected")
+
+
+def test_advertisement_after_chatting(acfactory, path_to_webxdc):
+    """Test that realtime advertisement is assigned to the correct message after chatting."""
+    ac1, ac2 = acfactory.get_online_accounts(2)
+    ac1.set_config("webxdc_realtime_enabled", "1")
+    ac2.set_config("webxdc_realtime_enabled", "1")
+
+    ac1_ac2_chat = ac1.create_chat(ac2)
+    ac1_webxdc_msg = ac1_ac2_chat.send_message(text="WebXDC", file=path_to_webxdc)
+    ac2_webxdc_msg = ac2.wait_for_incoming_msg()
+    assert ac2_webxdc_msg.get_snapshot().text == "WebXDC"
+
+    ac1_ac2_chat.send_text("Hello!")
+    ac2_hello_msg = ac2.wait_for_incoming_msg()
+    ac2_hello_msg_snapshot = ac2_hello_msg.get_snapshot()
+    assert ac2_hello_msg_snapshot.text == "Hello!"
+    ac2_hello_msg_snapshot.chat.accept()
+
+    ac2_webxdc_msg.send_webxdc_realtime_advertisement()
+    while 1:
+        event = ac1.wait_for_event()
+        if event.kind == EventType.WEBXDC_REALTIME_ADVERTISEMENT_RECEIVED:
+            assert event.msg_id == ac1_webxdc_msg.id
+            break

deltachat-rpc-client/tests/test_securejoin.py

@@ -61,7 +61,7 @@ def test_qr_securejoin(acfactory, protect, tmp_path):
     # Setup second device for Alice
     # to test observing securejoin protocol.
     alice.export_backup(tmp_path)
-    files = list(tmp_path.glob("*.tar"))
+    files = list(tmp_path.glob("*.tar.gz"))
     alice2 = acfactory.get_unconfigured_account()
     alice2.import_backup(files[0])
 

deltachat-rpc-client/tests/test_something.py

@@ -379,7 +379,7 @@ def test_import_export_backup(acfactory, tmp_path) -> None:
     alice = acfactory.new_configured_account()
     alice.export_backup(tmp_path)
 
-    files = list(tmp_path.glob("*.tar"))
+    files = list(tmp_path.glob("*.tar.gz"))
     alice2 = acfactory.get_unconfigured_account()
     alice2.import_backup(files[0])
 
@@ -630,7 +630,7 @@ def test_markseen_contact_request(acfactory, tmp_path):
 
     # Bob sets up a second device.
     bob.export_backup(tmp_path)
-    files = list(tmp_path.glob("*.tar"))
+    files = list(tmp_path.glob("*.tar.gz"))
     bob2 = acfactory.get_unconfigured_account()
     bob2.import_backup(files[0])
     bob2.start_io()

deltachat-rpc-server/Cargo.toml

@@ -1,6 +1,6 @@
 [package]
 name = "deltachat-rpc-server"
-version = "1.148.2"
+version = "1.148.4"
 description = "DeltaChat JSON-RPC server"
 edition = "2021"
 readme = "README.md"

deltachat-rpc-server/npm-package/package.json

@@ -15,5 +15,5 @@
   },
   "type": "module",
   "types": "index.d.ts",
-  "version": "1.148.2"
+  "version": "1.148.4"
 }

package.json

@@ -55,5 +55,5 @@
     "test:mocha": "mocha node/test/test.mjs --growl --reporter=spec --bail --exit"
   },
   "types": "node/dist/index.d.ts",
-  "version": "1.148.2"
+  "version": "1.148.4"
 }

python/pyproject.toml

@@ -4,7 +4,7 @@ build-backend = "setuptools.build_meta"
 
 [project]
 name = "deltachat"
-version = "1.148.2"
+version = "1.148.4"
 description = "Python bindings for the Delta Chat Core library using CFFI against the Rust-implemented libdeltachat"
 readme = "README.rst"
 requires-python = ">=3.7"

release-date.in

@@ -1 +1 @@
-2024-10-23
+2024-10-24

src/constants.rs

@@ -4,12 +4,16 @@
 
 use deltachat_derive::{FromSql, ToSql};
 use once_cell::sync::Lazy;
+use percent_encoding::{AsciiSet, NON_ALPHANUMERIC};
 use serde::{Deserialize, Serialize};
 
 use crate::chat::ChatId;
 
 pub static DC_VERSION_STR: Lazy<String> = Lazy::new(|| env!("CARGO_PKG_VERSION").to_string());
 
+/// Set of characters to percent-encode in email addresses and names.
+pub(crate) const NON_ALPHANUMERIC_WITHOUT_DOT: &AsciiSet = &NON_ALPHANUMERIC.remove(b'.');
+
 #[derive(
     Debug,
     Default,

src/imex.rs

@@ -11,7 +11,7 @@ use futures_lite::FutureExt;
 use pin_project::pin_project;
 
 use tokio::fs::{self, File};
-use tokio::io::{AsyncRead, AsyncWrite, ReadBuf};
+use tokio::io::{AsyncRead, AsyncWrite, AsyncWriteExt, ReadBuf};
 use tokio_tar::Archive;
 
 use crate::blob::BlobDirContents;
@@ -123,7 +123,7 @@ pub async fn has_backup(_context: &Context, dir_name: &Path) -> Result<String> {
         let name = dirent.file_name();
         let name: String = name.to_string_lossy().into();
         if name.starts_with("delta-chat")
-            && name.ends_with(".tar")
+            && (name.ends_with(".tar") || name.ends_with(".tar.gz"))
             && (newest_backup_name.is_empty() || name > newest_backup_name)
         {
             // We just use string comparison to determine which backup is newer.
@@ -269,30 +269,24 @@ async fn import_backup(
         context.get_dbfile().display()
     );
 
-    import_backup_stream(context, backup_file, file_size, passphrase).await?;
+    let backup_file = ProgressReader::new(backup_file, context.clone(), file_size);
+    if backup_to_import.extension() == Some(OsStr::new("gz")) {
+        let backup_file = tokio::io::BufReader::new(backup_file);
+        let backup_file = async_compression::tokio::bufread::GzipDecoder::new(backup_file);
+        import_backup_stream(context, backup_file, passphrase).await?;
+    } else {
+        import_backup_stream(context, backup_file, passphrase).await?;
+    }
     Ok(())
 }
 
 /// Imports backup by reading a tar file from a stream.
-///
-/// `file_size` is used to calculate the progress
-/// and emit progress events.
-/// Ideally it is the sum of the entry
-/// sizes without the header overhead,
-/// but can be estimated as tar file size
-/// in which case the progress is underestimated
-/// and may not reach 99.9% by the end of import.
-/// Underestimating is better than
-/// overestimating because the progress
-/// jumps to 100% instead of getting stuck at 99.9%
-/// for some time.
 pub(crate) async fn import_backup_stream<R: tokio::io::AsyncRead + Unpin>(
     context: &Context,
     backup_file: R,
-    file_size: u64,
     passphrase: String,
 ) -> Result<()> {
-    import_backup_stream_inner(context, backup_file, file_size, passphrase)
+    import_backup_stream_inner(context, backup_file, passphrase)
         .await
         .0
 }
@@ -319,6 +313,19 @@ struct ProgressReader<R> {
 }
 
 impl<R> ProgressReader<R> {
+    /// Creates a new `ProgressReader`.
+    ///
+    /// `file_size` is used to calculate the progress
+    /// and emit progress events.
+    /// Ideally it is the sum of the entry
+    /// sizes without the header overhead,
+    /// but can be estimated as tar file size
+    /// in which case the progress is underestimated
+    /// and may not reach 99.9% by the end of import.
+    /// Underestimating is better than
+    /// overestimating because the progress
+    /// jumps to 100% instead of getting stuck at 99.9%
+    /// for some time.
     fn new(r: R, context: Context, file_size: u64) -> Self {
         Self {
             inner: r,
@@ -358,10 +365,8 @@ where
 async fn import_backup_stream_inner<R: tokio::io::AsyncRead + Unpin>(
     context: &Context,
     backup_file: R,
-    file_size: u64,
     passphrase: String,
 ) -> (Result<()>,) {
-    let backup_file = ProgressReader::new(backup_file, context.clone(), file_size);
     let mut archive = Archive::new(backup_file);
 
     let mut entries = match archive.entries() {
@@ -461,10 +466,10 @@ fn get_next_backup_path(
         tempdbfile.push(format!("{stem}-{i:02}-{addr}.db"));
 
         let mut tempfile = folder.clone();
-        tempfile.push(format!("{stem}-{i:02}-{addr}.tar.part"));
+        tempfile.push(format!("{stem}-{i:02}-{addr}.tar.gz.part"));
 
         let mut destfile = folder.clone();
-        destfile.push(format!("{stem}-{i:02}-{addr}.tar"));
+        destfile.push(format!("{stem}-{i:02}-{addr}.tar.gz"));
 
         if !tempdbfile.exists() && !tempfile.exists() && !destfile.exists() {
             return Ok((tempdbfile, tempfile, destfile));
@@ -504,9 +509,13 @@ async fn export_backup(context: &Context, dir: &Path, passphrase: String) -> Res
         file_size += blob.to_abs_path().metadata()?.len()
     }
 
-    export_backup_stream(context, &temp_db_path, blobdir, file, file_size)
-        .await
-        .context("Exporting backup to file failed")?;
+    let gzip_encoder = async_compression::tokio::write::GzipEncoder::new(file);
+    let mut gzip_encoder =
+        export_backup_stream(context, &temp_db_path, blobdir, gzip_encoder, file_size)
+            .await
+            .context("Exporting backup to file failed")?;
+    gzip_encoder.shutdown().await?;
+
     fs::rename(temp_path, &dest_path).await?;
     context.emit_event(EventType::ImexFileWritten(dest_path));
     Ok(())
@@ -543,6 +552,10 @@ impl<W> ProgressWriter<W> {
             context,
         }
     }
+
+    fn into_inner(self) -> W {
+        self.inner
+    }
 }
 
 impl<W> AsyncWrite for ProgressWriter<W>
@@ -590,12 +603,12 @@ pub(crate) async fn export_backup_stream<'a, W>(
     blobdir: BlobDirContents<'a>,
     writer: W,
     file_size: u64,
-) -> Result<()>
+) -> Result<W>
 where
     W: tokio::io::AsyncWrite + tokio::io::AsyncWriteExt + Unpin + Send + 'static,
 {
-    let writer = ProgressWriter::new(writer, context.clone(), file_size);
-    let mut builder = tokio_tar::Builder::new(writer);
+    let progress_writer = ProgressWriter::new(writer, context.clone(), file_size);
+    let mut builder = tokio_tar::Builder::new(progress_writer);
 
     builder
         .append_path_with_name(temp_db_path, DBFILE_BACKUP_NAME)
@@ -607,8 +620,9 @@ where
         builder.append_file(path_in_archive, &mut file).await?;
     }
 
-    builder.finish().await?;
-    Ok(())
+    // Convert tar builder back into the underlying stream.
+    let progress_writer = builder.into_inner().await?;
+    Ok(progress_writer.into_inner())
 }
 
 /// Imports secret key from a file.

src/imex/transfer.rs

@@ -36,12 +36,13 @@ use futures_lite::FutureExt;
 use iroh_net::relay::RelayMode;
 use iroh_net::Endpoint;
 use tokio::fs;
+use tokio::io::AsyncWriteExt;
 use tokio::task::JoinHandle;
 use tokio_util::sync::CancellationToken;
 
 use crate::chat::add_device_msg;
 use crate::context::Context;
-use crate::imex::BlobDirContents;
+use crate::imex::{BlobDirContents, ProgressReader};
 use crate::message::{Message, Viewtype};
 use crate::qr::Qr;
 use crate::stock_str::backup_transfer_msg_body;
@@ -190,9 +191,11 @@ impl BackupProvider {
 
         send_stream.write_all(&file_size.to_be_bytes()).await?;
 
-        export_backup_stream(&context, &dbfile, blobdir, send_stream, file_size)
-            .await
-            .context("Failed to write backup into QUIC stream")?;
+        let mut send_stream =
+            export_backup_stream(&context, &dbfile, blobdir, send_stream, file_size)
+                .await
+                .context("Failed to write backup into QUIC stream")?;
+        send_stream.shutdown().await?;
         info!(context, "Finished writing backup into QUIC stream.");
         let mut buf = [0u8; 1];
         info!(context, "Waiting for acknowledgment.");
@@ -310,7 +313,8 @@ pub async fn get_backup2(
     let mut file_size_buf = [0u8; 8];
     recv_stream.read_exact(&mut file_size_buf).await?;
     let file_size = u64::from_be_bytes(file_size_buf);
-    import_backup_stream(context, recv_stream, file_size, passphrase)
+    let recv_stream = ProgressReader::new(recv_stream, context.clone(), file_size);
+    import_backup_stream(context, recv_stream, passphrase)
         .await
         .context("Failed to import backup from QUIC stream")?;
     info!(context, "Finished importing backup from the stream.");

src/mimefactory.rs

@@ -20,6 +20,7 @@ use crate::e2ee::EncryptHelper;
 use crate::ephemeral::Timer as EphemeralTimer;
 use crate::headerdef::HeaderDef;
 use crate::html::new_html_mimepart;
+use crate::location;
 use crate::message::{self, Message, MsgId, Viewtype};
 use crate::mimeparser::SystemMessage;
 use crate::param::Param;
@@ -32,7 +33,6 @@ use crate::tools::{
     create_outgoing_rfc724_mid, create_smeared_timestamp, remove_subject_prefix, time,
 };
 use crate::webxdc::StatusUpdateSerial;
-use crate::{location, peer_channels};
 
 // attachments of 25 mb brutto should work on the majority of providers
 // (brutto examples: web.de=50, 1&1=40, t-online.de=32, gmail=25, posteo=50, yahoo=25, all-inkl=100).
@@ -1387,8 +1387,7 @@ impl MimeFactory {
             let json = msg.param.get(Param::Arg).unwrap_or_default();
             parts.push(context.build_status_update_part(json));
         } else if msg.viewtype == Viewtype::Webxdc {
-            let topic = peer_channels::create_random_topic();
-            headers.push(create_iroh_header(context, topic, msg.id).await?);
+            headers.push(create_iroh_header(context, msg.id).await?);
             if let (Some(json), _) = context
                 .render_webxdc_status_update_object(
                     msg.id,

src/net/proxy.rs

@@ -12,13 +12,14 @@ use fast_socks5::client::Socks5Stream;
 use fast_socks5::util::target_addr::ToTargetAddr;
 use fast_socks5::AuthenticationMethod;
 use fast_socks5::Socks5Command;
-use percent_encoding::{percent_encode, NON_ALPHANUMERIC};
+use percent_encoding::{percent_encode, utf8_percent_encode, NON_ALPHANUMERIC};
 use tokio::io::{AsyncReadExt, AsyncWriteExt};
 use tokio::net::TcpStream;
 use tokio_io_timeout::TimeoutStream;
 use url::Url;
 
 use crate::config::Config;
+use crate::constants::NON_ALPHANUMERIC_WITHOUT_DOT;
 use crate::context::Context;
 use crate::net::connect_tcp;
 use crate::net::session::SessionStream;
@@ -41,6 +42,12 @@ impl PartialEq for ShadowsocksConfig {
 
 impl Eq for ShadowsocksConfig {}
 
+impl ShadowsocksConfig {
+    fn to_url(&self) -> String {
+        self.server_config.to_url()
+    }
+}
+
 #[derive(Debug, Clone, PartialEq, Eq)]
 pub struct HttpConfig {
     /// HTTP proxy host.
@@ -84,6 +91,17 @@ impl HttpConfig {
         };
         Ok(http_config)
     }
+
+    fn to_url(&self, scheme: &str) -> String {
+        let host = utf8_percent_encode(&self.host, NON_ALPHANUMERIC_WITHOUT_DOT);
+        if let Some((user, password)) = &self.user_password {
+            let user = utf8_percent_encode(user, NON_ALPHANUMERIC);
+            let password = utf8_percent_encode(password, NON_ALPHANUMERIC);
+            format!("{scheme}://{user}:{password}@{host}:{}", self.port)
+        } else {
+            format!("{scheme}://{host}:{}", self.port)
+        }
+    }
 }
 
 #[derive(Debug, Clone, PartialEq, Eq)]
@@ -123,6 +141,17 @@ impl Socks5Config {
 
         Ok(socks_stream)
     }
+
+    fn to_url(&self) -> String {
+        let host = utf8_percent_encode(&self.host, NON_ALPHANUMERIC_WITHOUT_DOT);
+        if let Some((user, password)) = &self.user_password {
+            let user = utf8_percent_encode(user, NON_ALPHANUMERIC);
+            let password = utf8_percent_encode(password, NON_ALPHANUMERIC);
+            format!("socks5://{user}:{password}@{host}:{}", self.port)
+        } else {
+            format!("socks5://{host}:{}", self.port)
+        }
+    }
 }
 
 #[derive(Debug, Clone, PartialEq, Eq)]
@@ -217,7 +246,7 @@ where
 
 impl ProxyConfig {
     /// Creates a new proxy configuration by parsing given proxy URL.
-    fn from_url(url: &str) -> Result<Self> {
+    pub(crate) fn from_url(url: &str) -> Result<Self> {
         let url = Url::parse(url).context("Cannot parse proxy URL")?;
         match url.scheme() {
             "http" => {
@@ -272,6 +301,19 @@ impl ProxyConfig {
         }
     }
 
+    /// Serializes proxy config into an URL.
+    ///
+    /// This function can be used to normalize proxy URL
+    /// by parsing it and serializing back.
+    pub(crate) fn to_url(&self) -> String {
+        match self {
+            Self::Http(http_config) => http_config.to_url("http"),
+            Self::Https(http_config) => http_config.to_url("https"),
+            Self::Socks5(socks5_config) => socks5_config.to_url(),
+            Self::Shadowsocks(shadowsocks_config) => shadowsocks_config.to_url(),
+        }
+    }
+
     /// Migrates legacy `socks5_host`, `socks5_port`, `socks5_user` and `socks5_password`
     /// config into `proxy_url` if `proxy_url` is unset or empty.
     ///

src/peer_channels.rs

@@ -143,9 +143,10 @@ impl Iroh {
                 self.endpoint.add_node_addr(peer.clone())?;
             }
 
-            self.gossip
-                .join(topic, peers.into_iter().map(|peer| peer.node_id).collect())
-                .await?;
+            self.gossip.join_with_opts(
+                topic,
+                JoinOptions::with_bootstrap(peers.into_iter().map(|peer| peer.node_id)),
+            );
         }
         Ok(())
     }
@@ -259,7 +260,14 @@ impl Context {
 
         // create gossip
         let my_addr = endpoint.node_addr().await?;
-        let gossip = Gossip::from_endpoint(endpoint.clone(), Default::default(), &my_addr.info);
+        let gossip_config = iroh_gossip::proto::topic::Config {
+            // Allow messages up to 128 KB in size.
+            // We set the limit to 128 KiB to account for internal overhead,
+            // but only guarantee 128 KB of payload to WebXDC developers.
+            max_message_size: 128 * 1024,
+            ..Default::default()
+        };
+        let gossip = Gossip::from_endpoint(endpoint.clone(), gossip_config, &my_addr.info);
 
         // spawn endpoint loop that forwards incoming connections to the gossiper
         let context = self.clone();
@@ -419,15 +427,15 @@ pub async fn leave_webxdc_realtime(ctx: &Context, msg_id: MsgId) -> Result<()> {
     Ok(())
 }
 
-pub(crate) fn create_random_topic() -> TopicId {
+/// Creates a new random gossip topic.
+fn create_random_topic() -> TopicId {
     TopicId::from_bytes(rand::random())
 }
 
-pub(crate) async fn create_iroh_header(
-    ctx: &Context,
-    topic: TopicId,
-    msg_id: MsgId,
-) -> Result<Header> {
+/// Creates `Iroh-Gossip-Header` with a new random topic
+/// and stores the topic for the message.
+pub(crate) async fn create_iroh_header(ctx: &Context, msg_id: MsgId) -> Result<Header> {
+    let topic = create_random_topic();
     insert_topic_stub(ctx, msg_id, topic).await?;
     Ok(Header::new(
         HeaderDef::IrohGossipTopic.get_headername().to_string(),

src/qr.rs

@@ -20,7 +20,7 @@ use crate::events::EventType;
 use crate::key::Fingerprint;
 use crate::message::Message;
 use crate::net::http::post_empty;
-use crate::net::proxy::DEFAULT_SOCKS_PORT;
+use crate::net::proxy::{ProxyConfig, DEFAULT_SOCKS_PORT};
 use crate::peerstate::Peerstate;
 use crate::token;
 use crate::tools::validate_id;
@@ -723,6 +723,10 @@ pub async fn set_config_from_qr(context: &Context, qr: &str) -> Result<()> {
                 .get_config(Config::ProxyUrl)
                 .await?
                 .unwrap_or_default();
+
+            // Normalize the URL.
+            let url = ProxyConfig::from_url(&url)?.to_url();
+
             let proxy_urls: Vec<&str> = std::iter::once(url.as_str())
                 .chain(
                     old_proxy_url_value
@@ -1787,6 +1791,17 @@ mod tests {
             )
         );
 
+        // SOCKS5 config does not have port 1080 explicitly specified,
+        // but should bring `socks5://1.2.3.4:1080` to the top instead of creating another entry.
+        set_config_from_qr(&t, "socks5://1.2.3.4").await?;
+        assert_eq!(
+            t.get_config(Config::ProxyUrl).await?,
+            Some(
+                "socks5://1.2.3.4:1080\nss://YWVzLTEyOC1nY206dGVzdA@192.168.100.1:8888#Example1\nsocks5://foo:666\nsocks5://Da:x%26%25%24X@jau:1080"
+                    .to_string()
+            )
+        );
+
         Ok(())
     }
 

src/receive_imf.rs

@@ -1442,33 +1442,59 @@ async fn add_parts(
     }
 
     if let Some(node_addr) = mime_parser.get_header(HeaderDef::IrohNodeAddr) {
+        chat_id = DC_CHAT_ID_TRASH;
         match serde_json::from_str::<NodeAddr>(node_addr).context("Failed to parse node address") {
             Ok(node_addr) => {
                 info!(context, "Adding iroh peer with address {node_addr:?}.");
-                let instance_id = parent.context("Failed to get parent message")?.id;
-                context.emit_event(EventType::WebxdcRealtimeAdvertisementReceived {
-                    msg_id: instance_id,
-                });
-                if let Some(topic) = get_iroh_topic_for_msg(context, instance_id).await? {
-                    let node_id = node_addr.node_id;
-                    let relay_server = node_addr.relay_url().map(|relay| relay.as_str());
-                    iroh_add_peer_for_topic(context, instance_id, topic, node_id, relay_server)
-                        .await?;
-                    if context
-                        .get_config_bool(Config::WebxdcRealtimeEnabled)
-                        .await?
-                    {
-                        let iroh = context.get_or_try_init_peer_channel().await?;
-                        iroh.maybe_add_gossip_peers(topic, vec![node_addr]).await?;
+                match mime_parser.get_header(HeaderDef::InReplyTo) {
+                    Some(in_reply_to) => match rfc724_mid_exists(context, in_reply_to).await? {
+                        Some((instance_id, _ts_sent)) => {
+                            context.emit_event(EventType::WebxdcRealtimeAdvertisementReceived {
+                                msg_id: instance_id,
+                            });
+                            if let Some(topic) =
+                                get_iroh_topic_for_msg(context, instance_id).await?
+                            {
+                                let node_id = node_addr.node_id;
+                                let relay_server =
+                                    node_addr.relay_url().map(|relay| relay.as_str());
+                                iroh_add_peer_for_topic(
+                                    context,
+                                    instance_id,
+                                    topic,
+                                    node_id,
+                                    relay_server,
+                                )
+                                .await?;
+                                if context
+                                    .get_config_bool(Config::WebxdcRealtimeEnabled)
+                                    .await?
+                                {
+                                    let iroh = context.get_or_try_init_peer_channel().await?;
+                                    iroh.maybe_add_gossip_peers(topic, vec![node_addr]).await?;
+                                }
+                                info!(context, "Added iroh peer to the topic of {instance_id}.");
+                            } else {
+                                warn!(
+                                    context,
+                                    "Could not add iroh peer because {instance_id} has no topic."
+                                );
+                            }
+                        }
+                        None => {
+                            warn!(
+                                context,
+                                "Cannot add iroh peer because WebXDC instance does not exist."
+                            );
+                        }
+                    },
+                    None => {
+                        warn!(
+                            context,
+                            "Cannot add iroh peer because the message has no In-Reply-To."
+                        );
                     }
-                    info!(context, "Added iroh peer to the topic of {instance_id}.");
-                } else {
-                    warn!(
-                        context,
-                        "Could not add iroh peer because {instance_id} has no topic."
-                    );
                 }
-                chat_id = DC_CHAT_ID_TRASH;
             }
             Err(err) => {
                 warn!(context, "Couldn't parse NodeAddr: {err:#}.");

src/securejoin.rs

@@ -1,13 +1,13 @@
 //! Implementation of [SecureJoin protocols](https://securejoin.delta.chat/).
 
 use anyhow::{ensure, Context as _, Error, Result};
-use percent_encoding::{utf8_percent_encode, AsciiSet, NON_ALPHANUMERIC};
+use percent_encoding::{utf8_percent_encode, NON_ALPHANUMERIC};
 
 use crate::aheader::EncryptPreference;
 use crate::chat::{self, get_chat_id_by_grpid, Chat, ChatId, ChatIdBlocked, ProtectionStatus};
 use crate::chatlist_events;
 use crate::config::Config;
-use crate::constants::{Blocked, Chattype};
+use crate::constants::{Blocked, Chattype, NON_ALPHANUMERIC_WITHOUT_DOT};
 use crate::contact::{Contact, ContactId, Origin};
 use crate::context::Context;
 use crate::e2ee::ensure_secret_key_exists;
@@ -34,9 +34,6 @@ use qrinvite::QrInvite;
 
 use crate::token::Namespace;
 
-/// Set of characters to percent-encode in email addresses and names.
-pub const NON_ALPHANUMERIC_WITHOUT_DOT: &AsciiSet = &NON_ALPHANUMERIC.remove(b'.');
-
 fn inviter_progress(context: &Context, contact_id: ContactId, progress: usize) {
     debug_assert!(
         progress <= 1000,