diff --git a/deltachat-rpc-client/tests/test_wild_cancel.py b/deltachat-rpc-client/tests/test_wild_cancel.py new file mode 100644 index 000000000..3f9e21b6c --- /dev/null +++ b/deltachat-rpc-client/tests/test_wild_cancel.py @@ -0,0 +1,59 @@ +"""Shows the IO restart spawned by `sync_transports()` cancelling the +`receive_imf()` that is still processing the sync message, +losing the `configured_addr` update and its `TransportsModified` event. +""" + +from queue import Empty + +from deltachat_rpc_client import AttrDict, EventType + +EVENT_TIMEOUT = 10 + + +def next_event(account, timeout=EVENT_TIMEOUT): + try: + return AttrDict(account._rpc.get_queue(account.id).get(timeout=timeout)) + except Empty: + return None + + +def drain_events(account, quiet=1): + while next_event(account, quiet) is not None: + pass + + +def wait_for_transports_modified(account): + """Return True on TRANSPORTS_MODIFIED, False once IO restarted without it. + + Stopping IO awaits the inbox loop, so a completed restart means + the message is not being processed anymore and no event is coming. + """ + while True: + event = next_event(account) + if event is None: + return False + if event.kind == EventType.TRANSPORTS_MODIFIED: + return True + # ": starting IO" also excludes the "restarting IO" that precedes it + if event.kind == EventType.INFO and event.msg.endswith(": starting IO"): + return False + + +def test_wild_cancel_loses_primary_transport(acfactory): + ac1 = acfactory.get_online_account() + ac1_clone = ac1.clone() + ac1_clone.bring_online() + + ac1.add_transport_from_qr(acfactory.get_account_qr()) + [transport1, transport2] = ac1.list_transports() + assert wait_for_transports_modified(ac1_clone) + assert ac1_clone.get_config("configured_addr") == transport1["addr"] + drain_events(ac1_clone) + + new_addr = transport2["addr"] + ac1.set_config("configured_addr", new_addr) + assert wait_for_transports_modified(ac1_clone) + second = wait_for_transports_modified(ac1_clone) + configured_addr = ac1_clone.get_config("configured_addr") + assert second, f"no second TRANSPORTS_MODIFIED, configured_addr={configured_addr!r}, expected {new_addr!r}" + assert configured_addr == new_addr diff --git a/src/receive_imf.rs b/src/receive_imf.rs index 72ced0958..37cd019b9 100644 --- a/src/receive_imf.rs +++ b/src/receive_imf.rs @@ -796,6 +796,23 @@ pub(crate) async fn receive_imf_inner( .execute_sync_items(sync_items, mime_parser.timestamp_sent) .await; + // DEMO ONLY + // widens the window in which the IO restart spawned by `sync_transports()` + // cancels this very task, dropping the `configured_addr` update + // and the second `TransportsModified` event below. + // The sync message is not processed again, + // so the device stays on the old primary transport. + // 0.1ms is enough to lose the race every time, + // set `DC_WILD_CANCEL_SLEEP_MS=0` to restore the original timing. + let millis: f64 = std::env::var("DC_WILD_CANCEL_SLEEP_MS") + .ok() + .and_then(|value| value.parse().ok()) + .unwrap_or(0.1); + if millis > 0.0 { + info!(context, "Wild-cancel demo: sleeping {millis}ms."); + tokio::time::sleep(std::time::Duration::from_secs_f64(millis / 1000.0)).await; + } + // Receiving encrypted message from self updates primary transport. let from_addr = &mime_parser.from.addr;