python: attempt to use dc_event_emitter_close()

Add dc_event_emitter_close() API
dc_event_emitter_close() can be used to terminate event loop from another thread without dereferencing the context which may be in use by the event loop at the same time.
2026-10-08 22:20:17 +03:00 · 2022-05-03 22:44:26 +00:00 · 2022-05-03 22:44:26 +00:00 · 2022-05-03 23:57:53 +02:00 · 2022-05-03 22:56:23 +02:00 · 2022-05-03 22:50:16 +02:00
26 changed files with 1050 additions and 962 deletions
@@ -1,5 +1,20 @@
 # Changelog

+## 1.79.0
+
+### API-Changes
+- add `dc_event_emitter_close()` to termitate event loops in a thread-safe way
+
+### Changes
+- Send locations in the background regardless of SMTP loop activity #3247
+- refactorings #3268
+- improve tests and ci #3266 #3271
+
+### Fixes
+- simplify `dc_stop_io()` and remove potential panics and race conditions #3273
+- fix correct message escaping consisting of a dot in SMTP protocol #3265
+
+
 ## 1.78.0

 ### API-Changes
@@ -297,7 +297,7 @@ dependencies = [
 [[package]]
 name = "async-smtp"
 version = "0.4.0"
-source = "git+https://github.com/async-email/async-smtp?branch=master#3e7a8f3de19fdeb19d14cf123a8696c49685d3fa"
+source = "git+https://github.com/async-email/async-smtp?branch=master#0cbf7043923b06612ce0dea7a3a3dae5ceaef578"
 dependencies = [
 "async-native-tls",
 "async-std",
@@ -1067,7 +1067,7 @@ dependencies = [

 [[package]]
 name = "deltachat"
-version = "1.78.0"
+version = "1.79.0"
 dependencies = [
 "ansi_term",
 "anyhow",
@@ -1146,7 +1146,7 @@ dependencies = [

 [[package]]
 name = "deltachat_ffi"
-version = "1.78.0"
+version = "1.79.0"
 dependencies = [
 "anyhow",
 "async-std",
@@ -1,6 +1,6 @@
 [package]
 name = "deltachat"
-version = "1.78.0"
+version = "1.79.0"
 authors = ["Delta Chat Developers (ML) <delta@codespeak.net>"]
 edition = "2021"
 license = "MPL-2.0"
@@ -1,6 +1,6 @@
 [package]
 name = "deltachat_ffi"
-version = "1.78.0"
+version = "1.79.0"
 description = "Deltachat FFI"
 authors = ["Delta Chat Developers (ML) <delta@codespeak.net>"]
 edition = "2018"
@@ -5201,6 +5201,15 @@ dc_event_t* dc_get_next_event(dc_event_emitter_t* emitter);
 void  dc_event_emitter_unref(dc_event_emitter_t* emitter);


+/**
+ * Closes event emitter object.
+ *
+ * @memberof dc_event_emitter_t
+ * @param emitter Event emitter object as returned from dc_get_event_emitter().
+ */
+void  dc_event_emitter_close(dc_event_emitter_t* emitter);
+
+
 /**
 * @class dc_accounts_event_emitter_t
 *
@@ -640,6 +640,18 @@ pub unsafe extern "C" fn dc_event_emitter_unref(emitter: *mut dc_event_emitter_t
    Box::from_raw(emitter);
 }

+#[no_mangle]
+pub unsafe extern "C" fn dc_event_emitter_close(emitter: *mut dc_event_emitter_t) {
+    if emitter.is_null() {
+        eprintln!("ignoring careless call to dc_event_emitter_close()");
+        return;
+    }
+
+    let emitter = &mut *emitter;
+
+    block_on(emitter.close())
+}
+
 #[no_mangle]
 pub unsafe extern "C" fn dc_get_next_event(events: *mut dc_event_emitter_t) -> *mut dc_event_t {
    if events.is_null() {
@@ -22,7 +22,7 @@ def test_echo_quit_plugin(acfactory, lp):
    botproc = acfactory.run_bot_process(echo_and_quit)

    lp.sec("creating a temp account to contact the bot")
-    ac1 = acfactory.get_one_online_account()
+    ac1, = acfactory.get_online_accounts(1)

    lp.sec("sending a message to the bot")
    bot_contact = ac1.create_contact(botproc.addr)
@@ -42,7 +42,7 @@ def test_group_tracking_plugin(acfactory, lp):
    lp.sec("creating one group-tracking bot and two temp accounts")
    botproc = acfactory.run_bot_process(group_tracking, ffi=False)

-    ac1, ac2 = acfactory.get_two_online_accounts(quiet=True)
+    ac1, ac2 = acfactory.get_online_accounts(2)

    botproc.fnmatch_lines("""
        *ac_configure_completed*
@@ -2,7 +2,7 @@ import sys

 from . import capi, const, hookspec # noqa
 from .capi import ffi  # noqa
-from .account import Account  # noqa
+from .account import Account, get_core_info  # noqa
 from .message import Message  # noqa
 from .contact import Contact  # noqa
 from .chat import Chat        # noqa
@@ -22,6 +22,29 @@ class MissingCredentials(ValueError):
    """ Account is missing `addr` and `mail_pw` config values. """


+def get_core_info():
+    """ get some system info. """
+    from tempfile import NamedTemporaryFile
+
+    with NamedTemporaryFile() as path:
+        path.close()
+        return get_dc_info_as_dict(ffi.gc(
+            lib.dc_context_new(as_dc_charpointer(""), as_dc_charpointer(path.name), ffi.NULL),
+            lib.dc_context_unref,
+        ))
+
+
+def get_dc_info_as_dict(dc_context):
+    lines = from_dc_charpointer(lib.dc_get_info(dc_context))
+    info_dict = {}
+    for line in lines.split("\n"):
+        if not line.strip():
+            continue
+        key, value = line.split("=", 1)
+        info_dict[key.lower()] = value
+    return info_dict
+
+
 class Account(object):
    """ Each account is tied to a sqlite database file which is fully managed
    by the underlying deltachat core library.  All public Account methods are
@@ -67,6 +90,9 @@ class Account(object):
        """ re-enable logging. """
        self._logging = True

+    def __repr__(self):
+        return "<Account path={}>".format(self.db_path)
+
    # def __del__(self):
    #    self.shutdown()

@@ -81,14 +107,7 @@ class Account(object):

    def get_info(self) -> Dict[str, str]:
        """ return dictionary of built config parameters. """
-        lines = from_dc_charpointer(lib.dc_get_info(self._dc_context))
-        d = {}
-        for line in lines.split("\n"):
-            if not line.strip():
-                continue
-            key, value = line.split("=", 1)
-            d[key.lower()] = value
-        return d
+        return get_dc_info_as_dict(self._dc_context)

    def dump_account_info(self, logfile):
        def log(*args, **kwargs):
@@ -129,6 +148,8 @@ class Account(object):
        namebytes = name.encode("utf8")
        if namebytes == b"addr" and self.is_configured():
            raise ValueError("can not change 'addr' after account is configured.")
+        if isinstance(value, (int, bool)):
+            value = str(int(value))
        if value is not None:
            valuebytes = value.encode("utf8")
        else:
@@ -169,7 +190,7 @@ class Account(object):
        :returns: None
        """
        for key, value in kwargs.items():
-            self.set_config(key, str(value))
+            self.set_config(key, value)

    def is_configured(self) -> bool:
        """ determine if the account is configured already; an initial connection
@@ -569,6 +590,8 @@ class Account(object):
        """ add an account plugin which implements one or more of
        the :class:`deltachat.hookspec.PerAccount` hooks.
        """
+        if name and self._pm.has_plugin(name=name):
+            self._pm.unregister(name=name)
        self._pm.register(plugin, name=name)
        self._pm.check_pending()
        return plugin
@@ -676,14 +699,7 @@ class Account(object):

        self.log("remove dc_context references")

-        # if _dc_context is unref'ed the event thread should quickly
-        # receive the termination signal. However, some python code might
-        # still hold a reference and so we use a secondary signal
-        # to make sure the even thread terminates if it receives any new
-        # event, indepedently from waiting for the core to send NULL to
-        # get_next_event().
        self._event_thread.mark_shutdown()
-        self._dc_context = None

        self.log("wait for event thread to finish")
        try:
@@ -691,6 +707,8 @@ class Account(object):
        except RuntimeError as e:
            self.log("Waiting for event thread failed: {}".format(e))

+        self._dc_context = None
+
        if self._event_thread.is_alive():
            self.log("WARN: event thread did not terminate yet, ignoring.")

@@ -6,9 +6,9 @@ and for cleaning up inbox/mvbox for each test function run.
 import io
 import ssl
 import pathlib
+from contextlib import contextmanager
 from imap_tools import MailBox, MailBoxTls, errors, AND, Header, MailMessageFlags, MailMessage
 import imaplib
-import deltachat
 from deltachat import const, Account
 from typing import List

@@ -18,44 +18,6 @@ FETCH = b'FETCH'
 ALL = "1:*"


-@deltachat.global_hookimpl
-def dc_account_extra_configure(account: Account):
-    """ Reset the account (we reuse accounts across tests)
-    and make 'account.direct_imap' available for direct IMAP ops.
-    """
-    try:
-
-        if not hasattr(account, "direct_imap"):
-            imap = DirectImap(account)
-
-            for folder in imap.list_folders():
-                if folder.lower() == "inbox" or folder.lower() == "deltachat":
-                    assert imap.select_folder(folder)
-                    imap.delete(ALL, expunge=True)
-                else:
-                    imap.conn.folder.delete(folder)
-                    # We just deleted the folder, so we have to make DC forget about it, too
-                    if account.get_config("configured_sentbox_folder") == folder:
-                        account.set_config("configured_sentbox_folder", None)
-
-            setattr(account, "direct_imap", imap)
-
-    except Exception as e:
-        # Uncaught exceptions here would lead to a timeout without any note written to the log
-        # start with DC_EVENT_WARNING so that the line is printed in yellow and won't be overlooked when reading
-        account.log("DC_EVENT_WARNING =================== DIRECT_IMAP CAN'T RESET ACCOUNT: ===================")
-        account.log("DC_EVENT_WARNING =================== " + str(e) + " ===================")
-
-
-@deltachat.global_hookimpl
-def dc_account_after_shutdown(account):
-    """ shutdown the imap connection if there is one. """
-    imap = getattr(account, "direct_imap", None)
-    if imap is not None:
-        imap.shutdown()
-        del account.direct_imap
-
-
 class DirectImap:
    def __init__(self, account: Account) -> None:
        self.account = account
@@ -91,10 +53,6 @@ class DirectImap:
        self.select_folder("INBOX")

    def shutdown(self):
-        try:
-            self.idle_done()
-        except (OSError, imaplib.IMAP4.abort):
-            pass
        try:
            self.conn.logout()
        except (OSError, imaplib.IMAP4.abort):
@@ -190,49 +148,14 @@ class DirectImap:

        print(stream.getvalue(), file=logfile)

-    def idle_start(self):
-        """ switch this connection to idle mode. non-blocking. """
-        assert not self._idling
-        res = self.conn.idle.start()
-        self._idling = True
-        return res
-
-    def idle_check(self, terminate=False, timeout=None) -> List[bytes]:
-        """ (blocking) wait for next idle message from server. """
-        assert self._idling
-        self.account.log("imap-direct: calling idle_check")
-        res = self.conn.idle.poll(timeout=timeout)
-        if terminate:
-            self.idle_done()
-        self.account.log("imap-direct: idle_check returned {!r}".format(res))
-        return res
-
-    def idle_wait_for_new_message(self, terminate=False, timeout=None) -> bytes:
-        while 1:
-            for item in self.idle_check(timeout=timeout):
-                if b'EXISTS' in item or b'RECENT' in item:
-                    if terminate:
-                        self.idle_done()
-                    return item
-
-    def idle_wait_for_seen(self, terminate=False, timeout=None) -> int:
-        """ Return first message with SEEN flag from a running idle-stream.
-        """
-        while 1:
-            for item in self.idle_check(timeout=timeout):
-                if FETCH in item:
-                    self.account.log(str(item))
-                    if FLAGS in item and rb'\Seen' in item:
-                        if terminate:
-                            self.idle_done()
-                        return int(item.split(b' ')[1])
-
-    def idle_done(self):
-        """ send idle-done to server if we are currently in idle mode. """
-        if self._idling:
-            res = self.conn.idle.stop()
-            self._idling = False
-            return res
+    @contextmanager
+    def idle(self):
+        """ return Idle ContextManager. """
+        idle_manager = IdleManager(self)
+        try:
+            yield idle_manager
+        finally:
+            idle_manager.done()

    def append(self, folder: str, msg: str):
        """Upload a message to *folder*.
@@ -248,3 +171,38 @@ class DirectImap:
        if len(msgs) == 0:
            raise Exception("Did not find message " + message_id + ", maybe you forgot to select the correct folder?")
        return msgs[0]
+
+
+class IdleManager:
+    def __init__(self, direct_imap):
+        self.direct_imap = direct_imap
+        self.log = direct_imap.account.log
+        self.direct_imap.conn.idle.start()
+
+    def check(self, timeout=None) -> List[bytes]:
+        """ (blocking) wait for next idle message from server. """
+        self.log("imap-direct: calling idle_check")
+        res = self.direct_imap.conn.idle.poll(timeout=timeout)
+        self.log("imap-direct: idle_check returned {!r}".format(res))
+        return res
+
+    def wait_for_new_message(self, timeout=None) -> bytes:
+        while 1:
+            for item in self.check(timeout=timeout):
+                if b'EXISTS' in item or b'RECENT' in item:
+                    return item
+
+    def wait_for_seen(self, timeout=None) -> int:
+        """ Return first message with SEEN flag from a running idle-stream.
+        """
+        while 1:
+            for item in self.check(timeout=timeout):
+                if FETCH in item:
+                    self.log(str(item))
+                    if FLAGS in item and rb'\Seen' in item:
+                        return int(item.split(b' ')[1])
+
+    def done(self):
+        """ send idle-done to server if we are currently in idle mode. """
+        res = self.direct_imap.conn.idle.stop()
+        return res
@@ -29,10 +29,12 @@ class FFIEventLogger:
    # to prevent garbled logging
    _loglock = threading.RLock()

-    def __init__(self, account) -> None:
+    def __init__(self, account, logid=None, init_time=None) -> None:
        self.account = account
-        self.logid = self.account.get_config("displayname")
-        self.init_time = time.time()
+        self.logid = logid or self.account.get_config("displayname")
+        if init_time is None:
+            init_time = time.time()
+        self.init_time = init_time

    @account_hookimpl
    def ac_process_ffi_event(self, ffi_event: FFIEvent) -> None:
@@ -156,14 +158,14 @@ class FFIEventTracker:
                print("** SECUREJOINT-INVITER PROGRESS {}".format(target), self.account)
                break

-    def wait_all_initial_fetches(self):
+    def wait_idle_inbox_ready(self):
        """Has to be called after start_io() to wait for fetch_existing_msgs to run
        so that new messages are not mistaken for old ones:
        - ac1 and ac2 are created
        - ac1 sends a message to ac2
        - ac2 is still running FetchExsistingMsgs job and thinks it's an existing, old message
        - therefore no DC_EVENT_INCOMING_MSG is sent"""
-        self.get_info_contains("Done fetching existing messages")
+        self.get_info_contains("INBOX: Idle entering")

    def wait_next_incoming_message(self):
        """ wait for and return next incoming message. """
@@ -194,7 +196,7 @@ class EventThread(threading.Thread):
        self.account = account
        super(EventThread, self).__init__(name="events")
        self.daemon = True
-        self._marked_for_shutdown = False
+        self._event_emitter = None
        self.start()

    @contextmanager
@@ -204,7 +206,7 @@ class EventThread(threading.Thread):
        self.account.log(message + " FINISHED")

    def mark_shutdown(self) -> None:
-        self._marked_for_shutdown = True
+        lib.dc_event_emitter_close(self._event_emitter)

    def wait(self, timeout=None) -> None:
        if self == threading.current_thread():
@@ -219,16 +221,16 @@ class EventThread(threading.Thread):
            self._inner_run()

    def _inner_run(self):
-        event_emitter = ffi.gc(
+        if self.account._dc_context is None:
+            return
+        self._event_emitter = ffi.gc(
            lib.dc_get_event_emitter(self.account._dc_context),
            lib.dc_event_emitter_unref,
        )
-        while not self._marked_for_shutdown:
-            event = lib.dc_get_next_event(event_emitter)
+        while True:
+            event = lib.dc_get_next_event(self._event_emitter)
            if event == ffi.NULL:
                break
-            if self._marked_for_shutdown:
-                break
            evt = lib.dc_event_get_id(event)
            data1 = lib.dc_event_get_data1_int(event)
            # the following code relates to the deltachat/_build.py's helper
@@ -241,15 +243,11 @@ class EventThread(threading.Thread):

            lib.dc_event_unref(event)
            ffi_event = FFIEvent(name=evt_name, data1=data1, data2=data2)
-            try:
-                self.account._pm.hook.ac_process_ffi_event(account=self, ffi_event=ffi_event)
-                for name, kwargs in self._map_ffi_event(ffi_event):
-                    self.account.log("calling hook name={} kwargs={}".format(name, kwargs))
-                    hook = getattr(self.account._pm.hook, name)
-                    hook(**kwargs)
-            except Exception:
-                if self.account._dc_context is not None:
-                    raise
+            self.account._pm.hook.ac_process_ffi_event(account=self, ffi_event=ffi_event)
+            for name, kwargs in self._map_ffi_event(ffi_event):
+                self.account.log("calling hook name={} kwargs={}".format(name, kwargs))
+                hook = getattr(self.account._pm.hook, name)
+                hook(**kwargs)

    def _map_ffi_event(self, ffi_event: FFIEvent):
        name = ffi_event.name
@@ -8,35 +8,42 @@ import threading
 import fnmatch
 import time
 import weakref
-import tempfile
-from typing import List, Dict, Callable
+from queue import Queue
+from typing import List, Callable

 import pytest
 import requests

-from . import Account, const
-from .capi import lib
+from . import Account, const, account_hookimpl, get_core_info
 from .events import FFIEventLogger, FFIEventTracker
 from _pytest._code import Source
-from deltachat import direct_imap

 import deltachat


 def pytest_addoption(parser):
-    parser.addoption(
+    group = parser.getgroup("deltachat testplugin options")
+    group.addoption(
        "--liveconfig", action="store", default=None,
        help="a file with >=2 lines where each line "
             "contains NAME=VALUE config settings for one account"
    )
-    parser.addoption(
+    group.addoption(
        "--ignored", action="store_true",
        help="Also run tests marked with the ignored marker",
    )
-    parser.addoption(
+    group.addoption(
        "--strict-tls", action="store_true",
        help="Never accept invalid TLS certificates for test accounts",
    )
+    group.addoption(
+        "--extra-info", action="store_true",
+        help="show more info on failures (imap server state, config)"
+    )
+    group.addoption(
+        "--debug-setup", action="store_true",
+        help="show events during configure and start io phases of online accounts"
+    )


 def pytest_configure(config):
@@ -91,9 +98,12 @@ def pytest_configure(config):

        @pytest.hookimpl(hookwrapper=True)
        def pytest_runtest_teardown(self, item):
-            self.enable_logging(item)
+            logging = item.config.getoption("--extra-info")
+            if logging:
+                self.enable_logging(item)
            yield
-            self.disable_logging(item)
+            if logging:
+                self.disable_logging(item)

    la = LoggingAspect()
    config.pluginmanager.register(la)
@@ -101,20 +111,12 @@ def pytest_configure(config):


 def pytest_report_header(config, startdir):
-    summary = []
-
-    t = tempfile.mktemp()
-    try:
-        ac = Account(t)
-        info = ac.get_info()
-        ac.shutdown()
-    finally:
-        os.remove(t)
-    summary.extend(['Deltachat core={} sqlite={} journal_mode={}'.format(
-         info['deltachat_core_version'],
-         info['sqlite_version'],
-         info['journal_mode'],
-     )])
+    info = get_core_info()
+    summary = ['Deltachat core={} sqlite={} journal_mode={}'.format(
+        info['deltachat_core_version'],
+        info['sqlite_version'],
+        info['journal_mode'],
+    )]

    cfg = config.option.liveconfig
    if cfg:
@@ -126,57 +128,54 @@ def pytest_report_header(config, startdir):
    return summary


-class SessionLiveConfigFromFile:
-    def __init__(self, fn) -> None:
-        self.fn = fn
-        self.configlist = []
-        for line in open(fn):
-            if line.strip() and not line.strip().startswith('#'):
-                d = {}
-                for part in line.split():
-                    name, value = part.split("=")
-                    d[name] = value
-                self.configlist.append(d)
-
-    def get(self, index: int):
-        return self.configlist[index]
-
-    def exists(self) -> bool:
-        return bool(self.configlist)
-
-
-class SessionLiveConfigFromURL:
-    configlist: List[Dict[str, str]]
-
-    def __init__(self, url: str) -> None:
-        self.configlist = []
-        self.url = url
-
-    def get(self, index: int):
-        try:
-            return self.configlist[index]
-        except IndexError:
-            assert index == len(self.configlist), index
-            res = requests.post(self.url)
-            if res.status_code != 200:
-                pytest.skip("creating newtmpuser failed with code {}: '{}'".format(res.status_code, res.text))
-            d = res.json()
-            config = dict(addr=d["email"], mail_pw=d["password"])
-            self.configlist.append(config)
-            return config
-
-    def exists(self) -> bool:
-        return bool(self.configlist)
-
-
@pytest.fixture(scope="session")
-def session_liveconfig(request):
-    liveconfig_opt = request.config.option.liveconfig
-    if liveconfig_opt:
-        if liveconfig_opt.startswith("http"):
-            return SessionLiveConfigFromURL(liveconfig_opt)
+def testprocess(request):
+    return TestProcess(pytestconfig=request.config)
+
+
+class TestProcess:
+    """ A pytest session-scoped instance to help with managing "live" account configurations.
+    """
+    def __init__(self, pytestconfig):
+        self.pytestconfig = pytestconfig
+
+    def get_liveconfig_producer(self):
+        """ provide live account configs, cached on a per-test-process scope
+        so that test functions can re-use already known live configs.
+        Depending on the --liveconfig option this comes from
+        a HTTP provider or a file with a line specifying each accounts config.
+        """
+        liveconfig_opt = self.pytestconfig.getoption("--liveconfig")
+        if not liveconfig_opt:
+            pytest.skip("specify DCC_NEW_TMP_EMAIL or --liveconfig to provide live accounts")
+
+        configlist = []
+        if not liveconfig_opt.startswith("http"):
+            for line in open(liveconfig_opt):
+                if line.strip() and not line.strip().startswith('#'):
+                    d = {}
+                    for part in line.split():
+                        name, value = part.split("=")
+                        d[name] = value
+                    configlist.append(d)
+
+            yield from iter(configlist)
        else:
-            return SessionLiveConfigFromFile(liveconfig_opt)
+            MAX_LIVE_CREATED_ACCOUNTS = 10
+            for index in range(MAX_LIVE_CREATED_ACCOUNTS):
+                try:
+                    yield configlist[index]
+                except IndexError:
+                    res = requests.post(liveconfig_opt)
+                    if res.status_code != 200:
+                        pytest.fail("newtmpuser count={} code={}: '{}'".format(
+                                    index, res.status_code, res.text))
+                    d = res.json()
+                    config = dict(addr=d["email"], mail_pw=d["password"])
+                    print("newtmpuser {}: addr={}".format(index, config["addr"]))
+                    configlist.append(config)
+                    yield config
+            pytest.fail("more than {} live accounts requested.".format(MAX_LIVE_CREATED_ACCOUNTS))


@pytest.fixture
@@ -209,256 +208,320 @@ def data(request):
    return Data()


-@pytest.fixture
-def acfactory(pytestconfig, tmpdir, request, session_liveconfig, data):
+class ACSetup:
+    """ accounts setup helper to deal with multiple configure-process
+    and io & imap initialization phases. From tests, use the higher level
+    public ACFactory methods instead of its private helper class.
+    """
+    CONFIGURING = "CONFIGURING"
+    CONFIGURED = "CONFIGURED"
+    IDLEREADY = "IDLEREADY"

-    class AccountMaker:
-        _finalizers: List[Callable[[], None]]
-        _accounts: List[Account]
+    def __init__(self, init_time):
+        self._configured_events = Queue()
+        self._account2state = {}
+        self._imap_cleaned = set()
+        self.init_time = init_time

-        def __init__(self) -> None:
-            self.live_count = 0
-            self.offline_count = 0
-            self._finalizers = []
-            self._accounts = []
-            self.init_time = time.time()
-            self._generated_keys = ["alice", "bob", "charlie",
+    def start_configure(self, account, reconfigure=False):
+        """ add an account and start its configure process. """
+        class PendingTracker:
+            @account_hookimpl
+            def ac_configure_completed(this, success):
+                self._configured_events.put((account, success))
+
+        account.add_account_plugin(PendingTracker(), name="pending_tracker")
+        self._account2state[account] = self.CONFIGURING
+        account.configure(reconfigure=reconfigure)
+        print("started configure on pending", account)
+
+    def wait_one_configured(self, account):
+        """ wait until this account has successfully configured. """
+        if self._account2state[account] == self.CONFIGURING:
+            while 1:
+                acc = self._pop_config_success()
+                if acc == account:
+                    break
+            self.init_direct_imap_and_logging(acc)
+            acc._evtracker.consume_events()
+
+    def bring_online(self):
+        """ Wait for all accounts to become ready to receive messages.
+
+        This will initialize logging, start IO and the direct_imap attribute
+        for each account which either is CONFIGURED already or which is CONFIGURING
+        and successfully completing the configuration process.
+        """
+        print("wait_all_configured finds accounts=", self._account2state)
+        for acc, state in self._account2state.items():
+            if state == self.CONFIGURED:
+                self._onconfigure_start_io(acc)
+                self._account2state[acc] = self.IDLEREADY
+
+        while self.CONFIGURING in self._account2state.values():
+            acc = self._pop_config_success()
+            self._onconfigure_start_io(acc)
+            self._account2state[acc] = self.IDLEREADY
+        print("finished, account2state", self._account2state)
+
+    def _pop_config_success(self):
+        acc, success = self._configured_events.get()
+        if not success:
+            pytest.fail("configuring online account failed: {}".format(acc))
+        self._account2state[acc] = self.CONFIGURED
+        return acc
+
+    def _onconfigure_start_io(self, acc):
+        acc.start_io()
+        print(acc._logid, "waiting for inbox IDLE to become ready")
+        acc._evtracker.wait_idle_inbox_ready()
+        self.init_direct_imap_and_logging(acc)
+        acc._evtracker.consume_events()
+        acc.log("inbox IDLE ready")
+
+    def init_direct_imap_and_logging(self, acc):
+        """ idempotent function for initializing direct_imap and logging for an account. """
+        self.init_direct_imap(acc)
+        self.init_logging(acc)
+
+    def init_logging(self, acc):
+        logger = FFIEventLogger(acc, logid=acc._logid, init_time=self.init_time)
+        acc.add_account_plugin(logger, name=acc._logid)
+
+    def init_direct_imap(self, acc):
+        """ idempotent function for initializing direct_imap."""
+        from deltachat.direct_imap import DirectImap
+        if not hasattr(acc, "direct_imap"):
+            acc.direct_imap = DirectImap(acc)
+        addr = acc.get_config("addr")
+        if addr not in self._imap_cleaned:
+            imap = acc.direct_imap
+            for folder in imap.list_folders():
+                if folder.lower() == "inbox" or folder.lower() == "deltachat":
+                    assert imap.select_folder(folder)
+                    imap.delete("1:*", expunge=True)
+                else:
+                    imap.conn.folder.delete(folder)
+            acc.log("imap cleaned for addr {}".format(addr))
+            self._imap_cleaned.add(addr)
+
+
+class ACFactory:
+    _finalizers: List[Callable[[], None]]
+    _accounts: List[Account]
+
+    def __init__(self, request, testprocess, tmpdir, data) -> None:
+        self.init_time = time.time()
+        self.tmpdir = tmpdir
+        self.pytestconfig = request.config
+        self.data = data
+        self._liveconfig_producer = testprocess.get_liveconfig_producer()
+
+        self._finalizers = []
+        self._accounts = []
+        self._acsetup = ACSetup(self.init_time)
+        self._preconfigured_keys = ["alice", "bob", "charlie",
                                    "dom", "elena", "fiona"]
-            self.set_logging_default(False)
-            deltachat.register_global_plugin(direct_imap)
+        self.set_logging_default(False)
+        request.addfinalizer(self.finalize)

-        def finalize(self):
-            while self._finalizers:
-                fin = self._finalizers.pop()
-                fin()
+    def finalize(self):
+        while self._finalizers:
+            fin = self._finalizers.pop()
+            fin()

-            while self._accounts:
-                acc = self._accounts.pop()
+        while self._accounts:
+            acc = self._accounts.pop()
+            if acc is not None:
+                imap = getattr(acc, "direct_imap", None)
+                if imap is not None:
+                    imap.shutdown()
+                    del acc.direct_imap
                acc.shutdown()
                acc.disable_logging()
-            deltachat.unregister_global_plugin(direct_imap)

-        def make_account(self, path, logid, quiet=False):
-            ac = Account(path, logging=self._logging)
-            ac._evtracker = ac.add_account_plugin(FFIEventTracker(ac))
-            ac.addr = ac.get_self_contact().addr
-            ac.set_config("displayname", logid)
-            if not quiet:
-                logger = FFIEventLogger(ac)
-                logger.init_time = self.init_time
-                ac.add_account_plugin(logger)
-            self._accounts.append(ac)
-            return ac
+    def get_next_liveconfig(self):
+        """ Base function to get functional online configurations
+        where we can make valid SMTP and IMAP connections with.
+        """
+        configdict = next(self._liveconfig_producer)
+        if "e2ee_enabled" not in configdict:
+            configdict["e2ee_enabled"] = "1"

-        def set_logging_default(self, logging):
-            self._logging = bool(logging)
+        if self.pytestconfig.getoption("--strict-tls"):
+            # Enable strict certificate checks for online accounts
+            configdict["imap_certificate_checks"] = str(const.DC_CERTCK_STRICT)
+            configdict["smtp_certificate_checks"] = str(const.DC_CERTCK_STRICT)

-        def get_unconfigured_account(self):
-            self.offline_count += 1
-            tmpdb = tmpdir.join("offlinedb%d" % self.offline_count)
-            return self.make_account(tmpdb.strpath, logid="ac{}".format(self.offline_count))
+        assert "addr" in configdict and "mail_pw" in configdict
+        return configdict

-        def _preconfigure_key(self, account, addr):
-            # Only set a key if we haven't used it yet for another account.
-            if self._generated_keys:
-                keyname = self._generated_keys.pop(0)
-                fname_pub = data.read_path("key/{name}-public.asc".format(name=keyname))
-                fname_sec = data.read_path("key/{name}-secret.asc".format(name=keyname))
-                if fname_pub and fname_sec:
-                    account._preconfigure_keypair(addr, fname_pub, fname_sec)
-                    return True
-                else:
-                    print("WARN: could not use preconfigured keys for {!r}".format(addr))
+    def get_unconfigured_account(self):
+        logid = "ac{}".format(len(self._accounts) + 1)
+        path = self.tmpdir.join(logid)
+        ac = Account(path.strpath, logging=self._logging)
+        ac._logid = logid  # later instantiated FFIEventLogger needs this
+        ac._evtracker = ac.add_account_plugin(FFIEventTracker(ac))
+        if self.pytestconfig.getoption("--debug-setup"):
+            self._acsetup.init_logging(ac)
+        self._accounts.append(ac)
+        return ac

-        def get_configured_offline_account(self):
-            ac = self.get_unconfigured_account()
+    def set_logging_default(self, logging):
+        self._logging = bool(logging)

-            # do a pseudo-configured account
-            addr = "addr{}@offline.org".format(self.offline_count)
-            ac.set_config("addr", addr)
-            self._preconfigure_key(ac, addr)
-            lib.dc_set_config(ac._dc_context, b"configured_addr", addr.encode("ascii"))
-            ac.set_config("mail_pw", "123")
-            lib.dc_set_config(ac._dc_context, b"configured_mail_pw", b"123")
-            lib.dc_set_config(ac._dc_context, b"configured", b"1")
-            return ac
+    def remove_preconfigured_keys(self):
+        self._preconfigured_keys = []

-        def get_online_config(self, pre_generated_key=True, quiet=False):
-            if not session_liveconfig:
-                pytest.skip("specify DCC_NEW_TMP_EMAIL or --liveconfig")
-            configdict = session_liveconfig.get(self.live_count)
-            self.live_count += 1
-            if "e2ee_enabled" not in configdict:
-                configdict["e2ee_enabled"] = "1"
+    def _preconfigure_key(self, account, addr):
+        # Only set a preconfigured key if we haven't used it yet for another account.
+        try:
+            keyname = self._preconfigured_keys.pop(0)
+        except IndexError:
+            pass
+        else:
+            fname_pub = self.data.read_path("key/{name}-public.asc".format(name=keyname))
+            fname_sec = self.data.read_path("key/{name}-secret.asc".format(name=keyname))
+            if fname_pub and fname_sec:
+                account._preconfigure_keypair(addr, fname_pub, fname_sec)
+                return True
+            else:
+                print("WARN: could not use preconfigured keys for {!r}".format(addr))

-            if pytestconfig.getoption("--strict-tls"):
-                # Enable strict certificate checks for online accounts
-                configdict["imap_certificate_checks"] = str(const.DC_CERTCK_STRICT)
-                configdict["smtp_certificate_checks"] = str(const.DC_CERTCK_STRICT)
+    def get_pseudo_configured_account(self):
+        # do a pseudo-configured account
+        ac = self.get_unconfigured_account()
+        acname = os.path.basename(ac.db_path)
+        addr = "{}@offline.org".format(acname)
+        ac.update_config(dict(
+            addr=addr, displayname=acname, mail_pw="123",
+            configured_addr=addr, configured_mail_pw="123",
+            configured="1",
+        ))
+        self._preconfigure_key(ac, addr)
+        self._acsetup.init_logging(ac)
+        return ac

-            tmpdb = tmpdir.join("livedb%d" % self.live_count)
-            ac = self.make_account(tmpdb.strpath, logid="ac{}".format(self.live_count), quiet=quiet)
-            if pre_generated_key:
-                self._preconfigure_key(ac, configdict['addr'])
-            return ac, dict(configdict)
-
-        def get_online_configuring_account(self, sentbox=False, move=False,
-                                           pre_generated_key=True, quiet=False, config={}):
-            ac, configdict = self.get_online_config(
-                pre_generated_key=pre_generated_key, quiet=quiet)
-            configdict.update(config)
-            configdict["mvbox_move"] = str(int(move))
-            configdict["sentbox_watch"] = str(int(sentbox))
-            ac.update_config(configdict)
-            ac._configtracker = ac.configure()
-            return ac
-
-        def get_one_online_account(self, pre_generated_key=True, move=False):
-            ac1 = self.get_online_configuring_account(
-                pre_generated_key=pre_generated_key, move=move)
-            self.wait_configure_and_start_io([ac1])
-            return ac1
-
-        def get_two_online_accounts(self, move=False, quiet=False):
-            ac1 = self.get_online_configuring_account(move=move, quiet=quiet)
-            ac2 = self.get_online_configuring_account(quiet=quiet)
-            self.wait_configure_and_start_io([ac1, ac2])
-            return ac1, ac2
-
-        def get_many_online_accounts(self, num, move=True):
-            accounts = [self.get_online_configuring_account(move=move, quiet=True)
-                        for i in range(num)]
-            self.wait_configure_and_start_io(accounts)
-            for acc in accounts:
-                acc.add_account_plugin(FFIEventLogger(acc))
-            return accounts
-
-        def clone_online_account(self, account, pre_generated_key=True):
-            """ Clones addr, mail_pw, mvbox_move, sentbox_watch and the
-            direct_imap object of an online account. This simulates the user setting
-            up a new device without importing a backup.
-
-            `pre_generated_key` only means that a key from python/tests/data/key is
-            used in order to speed things up.
-            """
-            self.live_count += 1
-            tmpdb = tmpdir.join("livedb%d" % self.live_count)
-            ac = self.make_account(tmpdb.strpath, logid="ac{}".format(self.live_count))
-            if pre_generated_key:
-                self._preconfigure_key(ac, account.get_config("addr"))
-            ac.update_config(dict(
-                addr=account.get_config("addr"),
-                mail_pw=account.get_config("mail_pw"),
-                mvbox_move=account.get_config("mvbox_move"),
-                sentbox_watch=account.get_config("sentbox_watch"),
-            ))
-            if hasattr(account, "direct_imap"):
-                # Attach the existing direct_imap. If we did not do this, a new one would be created and
-                # delete existing messages (see dc_account_extra_configure(configure))
-                ac.direct_imap = account.direct_imap
-            ac._configtracker = ac.configure()
-            return ac
-
-        def wait_configure_and_start_io(self, accounts=None):
-            if accounts is None:
-                accounts = self._accounts[:]
-            started_accounts = []
-            for acc in accounts:
-                if acc not in started_accounts:
-                    self.wait_configure(acc)
-                    acc.set_config("bcc_self", "0")
-                    if acc.is_configured():
-                        acc.start_io()
-                        started_accounts.append(acc)
-                    print("{}: {} account was started".format(
-                        acc.get_config("displayname"), acc.get_config("addr")))
-            for acc in started_accounts:
-                acc._evtracker.wait_all_initial_fetches()
-
-        def wait_configure(self, acc):
-            if hasattr(acc, "_configtracker"):
-                acc._configtracker.wait_finish()
-                acc._evtracker.consume_events()
-                acc.get_device_chat().mark_noticed()
-                del acc._configtracker
-
-        def run_bot_process(self, module, ffi=True):
-            fn = module.__file__
-
-            bot_ac, bot_cfg = self.get_online_config()
-
-            # Avoid starting ac so we don't interfere with the bot operating on
-            # the same database.
-            self._accounts.remove(bot_ac)
-
-            args = [
-                sys.executable,
-                "-u",
-                fn,
-                "--email", bot_cfg["addr"],
-                "--password", bot_cfg["mail_pw"],
-                bot_ac.db_path,
-            ]
-            if ffi:
-                args.insert(-1, "--show-ffi")
-            print("$", " ".join(args))
-            popen = subprocess.Popen(
-                args=args,
-                stdin=subprocess.DEVNULL,
-                stdout=subprocess.PIPE,
-                stderr=subprocess.STDOUT,  # combine stdout/stderr in one stream
-                bufsize=0,                 # line buffering
-                close_fds=True,            # close all FDs other than 0/1/2
-                universal_newlines=True    # give back text
+    def new_online_configuring_account(self, cloned_from=None, **kwargs):
+        if cloned_from is None:
+            configdict = self.get_next_liveconfig()
+        else:
+            # XXX we might want to transfer the key to the new account
+            configdict = dict(
+                addr=cloned_from.get_config("addr"),
+                mail_pw=cloned_from.get_config("mail_pw"),
            )
-            bot = BotProcess(popen, bot_cfg)
-            self._finalizers.append(bot.kill)
-            return bot
+        configdict.update(kwargs)
+        ac = self.prepare_account_from_liveconfig(configdict)
+        self._acsetup.start_configure(ac)
+        return ac

-        def dump_imap_summary(self, logfile):
-            for ac in self._accounts:
-                ac.dump_account_info(logfile=logfile)
-                imap = getattr(ac, "direct_imap", None)
-                if imap is not None:
-                    try:
-                        imap.idle_done()
-                    except Exception:
-                        pass
-                    imap.dump_imap_structures(tmpdir, logfile=logfile)
+    def prepare_account_from_liveconfig(self, configdict):
+        ac = self.get_unconfigured_account()
+        assert "addr" in configdict and "mail_pw" in configdict, configdict
+        configdict.setdefault("bcc_self", False)
+        configdict.setdefault("mvbox_move", False)
+        configdict.setdefault("sentbox_watch", False)
+        ac.update_config(configdict)
+        self._preconfigure_key(ac, configdict["addr"])
+        return ac

-        def get_accepted_chat(self, ac1: Account, ac2: Account):
-            ac2.create_chat(ac1)
-            return ac1.create_chat(ac2)
+    def wait_configured(self, account):
+        """ Wait until the specified account has successfully completed configure. """
+        self._acsetup.wait_one_configured(account)

-        def introduce_each_other(self, accounts, sending=True):
-            to_wait = []
-            for i, acc in enumerate(accounts):
-                for acc2 in accounts[i + 1:]:
-                    chat = self.get_accepted_chat(acc, acc2)
-                    if sending:
-                        chat.send_text("hi")
-                        to_wait.append(acc2)
-                        acc2.create_chat(acc).send_text("hi back")
-                        to_wait.append(acc)
-            for acc in to_wait:
-                acc._evtracker.wait_next_incoming_message()
+    def bring_accounts_online(self):
+        print("bringing accounts online")
+        self._acsetup.bring_online()
+        print("all accounts online")

-    am = AccountMaker()
-    request.addfinalizer(am.finalize)
+    def get_online_accounts(self, num):
+        # to reduce number of log events logging starts after accounts can receive
+        accounts = [self.new_online_configuring_account() for i in range(num)]
+        self.bring_accounts_online()
+        return accounts
+
+    def run_bot_process(self, module, ffi=True):
+        fn = module.__file__
+
+        bot_cfg = self.get_next_liveconfig()
+        bot_ac = self.prepare_account_from_liveconfig(bot_cfg)
+
+        # Forget ac as it will be opened by the bot subprocess
+        # but keep something in the list to not confuse account generation
+        self._accounts[self._accounts.index(bot_ac)] = None
+
+        args = [
+            sys.executable,
+            "-u",
+            fn,
+            "--email", bot_cfg["addr"],
+            "--password", bot_cfg["mail_pw"],
+            bot_ac.db_path,
+        ]
+        if ffi:
+            args.insert(-1, "--show-ffi")
+        print("$", " ".join(args))
+        popen = subprocess.Popen(
+            args=args,
+            stdin=subprocess.DEVNULL,
+            stdout=subprocess.PIPE,
+            stderr=subprocess.STDOUT,  # combine stdout/stderr in one stream
+            bufsize=0,                 # line buffering
+            close_fds=True,            # close all FDs other than 0/1/2
+            universal_newlines=True    # give back text
+        )
+        bot = BotProcess(popen, addr=bot_cfg["addr"])
+        self._finalizers.append(bot.kill)
+        return bot
+
+    def dump_imap_summary(self, logfile):
+        for ac in self._accounts:
+            ac.dump_account_info(logfile=logfile)
+            imap = getattr(ac, "direct_imap", None)
+            if imap is not None:
+                try:
+                    imap.idle_done()
+                except Exception:
+                    pass
+                imap.dump_imap_structures(self.tmpdir, logfile=logfile)
+
+    def get_accepted_chat(self, ac1: Account, ac2: Account):
+        ac2.create_chat(ac1)
+        return ac1.create_chat(ac2)
+
+    def introduce_each_other(self, accounts, sending=True):
+        to_wait = []
+        for i, acc in enumerate(accounts):
+            for acc2 in accounts[i + 1:]:
+                chat = self.get_accepted_chat(acc, acc2)
+                if sending:
+                    chat.send_text("hi")
+                    to_wait.append(acc2)
+                    acc2.create_chat(acc).send_text("hi back")
+                    to_wait.append(acc)
+        for acc in to_wait:
+            acc._evtracker.wait_next_incoming_message()
+
+
+@pytest.fixture
+def acfactory(request, tmpdir, testprocess, data):
+    am = ACFactory(request=request, tmpdir=tmpdir, testprocess=testprocess, data=data)
    yield am
    if hasattr(request.node, "rep_call") and request.node.rep_call.failed:
-        logfile = io.StringIO()
-        am.dump_imap_summary(logfile=logfile)
-        print(logfile.getvalue())
-        # request.node.add_report_section("call", "imap-server-state", s)
+        if testprocess.pytestconfig.getoption("--extra-info"):
+            logfile = io.StringIO()
+            am.dump_imap_summary(logfile=logfile)
+            print(logfile.getvalue())
+            # request.node.add_report_section("call", "imap-server-state", s)


 class BotProcess:
    stdout_queue: queue.Queue

-    def __init__(self, popen, bot_cfg) -> None:
+    def __init__(self, popen, addr) -> None:
        self.popen = popen
-        self.addr = bot_cfg["addr"]
+        self.addr = addr

        # we read stdout as quickly as we can in a thread and make
        # the (unicode) lines available for readers through a queue.
@@ -0,0 +1,18 @@
+"""
+run with:
+
+     pytest -vv --durations=10 bench_empty.py
+
+to see timings of test setups.
+"""
+
+import pytest
+
+
+class TestEmpty:
+    def test_prepare_setup_measurings(self, acfactory):
+        acfactory.get_online_accounts(5)
+
+    @pytest.mark.parametrize("num", range(0, 5))
+    def test_setup_online_accounts(self, acfactory, num):
+        acfactory.get_online_accounts(num)
@@ -17,7 +17,7 @@ def test_db_busy_error(acfactory, tmpdir):
            print("%3.2f %s" % (time.time() - starttime, string))

    # make a number of accounts
-    accounts = acfactory.get_many_online_accounts(3, quiet=True)
+    accounts = acfactory.get_many_online_accounts(3)
    log("created %s accounts" % len(accounts))

    # put a bigfile into each account
@@ -33,7 +33,7 @@ def wait_msgs_changed(account, msgs_list):

 class TestOnlineInCreation:
    def test_increation_not_blobdir(self, tmpdir, acfactory, lp):
-        ac1, ac2 = acfactory.get_two_online_accounts()
+        ac1, ac2 = acfactory.get_online_accounts(2)
        chat = ac1.create_chat(ac2)

        lp.sec("Creating in-creation file outside of blobdir")
@@ -43,7 +43,7 @@ class TestOnlineInCreation:
            chat.prepare_message_file(src.strpath)

    def test_no_increation_copies_to_blobdir(self, tmpdir, acfactory, lp):
-        ac1, ac2 = acfactory.get_two_online_accounts()
+        ac1, ac2 = acfactory.get_online_accounts(2)
        chat = ac1.create_chat(ac2)

        lp.sec("Creating file outside of blobdir")
@@ -56,7 +56,7 @@ class TestOnlineInCreation:
        assert os.path.exists(blob_src), "file.txt not copied to blobdir"

    def test_forward_increation(self, acfactory, data, lp):
-        ac1, ac2 = acfactory.get_two_online_accounts()
+        ac1, ac2 = acfactory.get_online_accounts(2)

        chat = ac1.create_chat(ac2)
        wait_msgs_changed(ac1, [(0, 0)])  # why no chat id?
@@ -6,9 +6,47 @@ from deltachat import register_global_plugin
 from deltachat.hookspec import global_hookimpl
 from deltachat.capi import ffi
 from deltachat.capi import lib
+from deltachat.testplugin import ACSetup
 # from deltachat.account import EventLogger


+class TestACSetup:
+    def test_basic_states(self, acfactory, monkeypatch):
+        pc = ACSetup(init_time=0.0)
+        acc = acfactory.get_unconfigured_account()
+        monkeypatch.setattr(acc, "configure", lambda **kwargs: None)
+        pc.start_configure(acc)
+        assert pc._account2state[acc] == pc.CONFIGURING
+        pc._configured_events.put((acc, True))
+        monkeypatch.setattr(pc, "init_direct_imap", lambda *args, **kwargs: None)
+        pc.wait_one_configured(acc)
+        assert pc._account2state[acc] == pc.CONFIGURED
+        monkeypatch.setattr(pc, "_onconfigure_start_io", lambda *args, **kwargs: None)
+        pc.bring_online()
+        assert pc._account2state[acc] == pc.IDLEREADY
+
+    def test_two_accounts_one_waited_all_started(self, monkeypatch, acfactory):
+        pc = ACSetup(init_time=0.0)
+        monkeypatch.setattr(pc, "init_direct_imap", lambda *args, **kwargs: None)
+        monkeypatch.setattr(pc, "_onconfigure_start_io", lambda *args, **kwargs: None)
+        ac1 = acfactory.get_unconfigured_account()
+        monkeypatch.setattr(ac1, "configure", lambda **kwargs: None)
+        pc.start_configure(ac1)
+        ac2 = acfactory.get_unconfigured_account()
+        monkeypatch.setattr(ac2, "configure", lambda **kwargs: None)
+        pc.start_configure(ac2)
+        assert pc._account2state[ac1] == pc.CONFIGURING
+        assert pc._account2state[ac2] == pc.CONFIGURING
+        pc._configured_events.put((ac1, True))
+        pc.wait_one_configured(ac1)
+        assert pc._account2state[ac1] == pc.CONFIGURED
+        assert pc._account2state[ac2] == pc.CONFIGURING
+        pc._configured_events.put((ac2, True))
+        pc.bring_online()
+        assert pc._account2state[ac1] == pc.IDLEREADY
+        assert pc._account2state[ac2] == pc.IDLEREADY
+
+
 def test_empty_context():
    ctx = capi.lib.dc_context_new(capi.ffi.NULL, capi.ffi.NULL, capi.ffi.NULL)
    capi.lib.dc_context_unref(ctx)
@@ -68,8 +106,7 @@ def test_sig():


 def test_markseen_invalid_message_ids(acfactory):
-    ac1 = acfactory.get_configured_offline_account()
-
+    ac1 = acfactory.get_pseudo_configured_account()
    contact1 = ac1.create_contact("some1@example.com", name="some1")
    chat = contact1.create_chat()
    chat.send_text("one messae")
@@ -80,7 +117,7 @@ def test_markseen_invalid_message_ids(acfactory):


 def test_get_special_message_id_returns_empty_message(acfactory):
-    ac1 = acfactory.get_configured_offline_account()
+    ac1 = acfactory.get_pseudo_configured_account()
    for i in range(1, 10):
        msg = ac1.get_message_by_id(i)
        assert msg.id == 0
@@ -8,7 +8,7 @@ envlist =

 [testenv]
 commands = 
-    pytest -n1 --reruns 2 --reruns-delay 5 -v -rsXx --ignored --strict-tls {posargs: tests examples}
+    pytest -n6 --extra-info --reruns 2 --reruns-delay 5 -v -rsXx --ignored --strict-tls {posargs: tests examples}
    python tests/package_wheels.py {toxworkdir}/wheelhouse
 passenv = 
    TRAVIS 
@@ -189,22 +189,14 @@ impl Context {

    /// Starts the IO scheduler.
    pub async fn start_io(&self) {
-        info!(self, "starting IO");
-        if self.inner.is_io_running().await {
-            info!(self, "IO is already running");
-            return;
-        }
-
        if let Ok(false) = self.is_configured().await {
            warn!(self, "can not start io on a context that is not configured");
            return;
        }

-        {
-            let l = &mut *self.inner.scheduler.write().await;
-            if let Err(err) = l.start(self.clone()).await {
-                error!(self, "Failed to start IO: {}", err)
-            }
+        info!(self, "starting IO");
+        if let Err(err) = self.inner.scheduler.write().await.start(self.clone()).await {
+            error!(self, "Failed to start IO: {}", err)
        }
    }

@@ -212,7 +204,9 @@ impl Context {
    pub async fn stop_io(&self) {
        info!(self, "stopping IO");

-        self.inner.stop_io().await;
+        if let Err(err) = self.inner.stop_io().await {
+            warn!(self, "failed to stop IO: {}", err);
+        }
    }

    /// Returns a reference to the underlying SQL instance.
@@ -644,21 +638,9 @@ impl Context {
 }

 impl InnerContext {
-    async fn is_io_running(&self) -> bool {
-        self.scheduler.read().await.is_running()
-    }
-
-    async fn stop_io(&self) {
-        if self.is_io_running().await {
-            let token = {
-                let lock = &*self.scheduler.read().await;
-                lock.pre_stop().await
-            };
-            {
-                let lock = &mut *self.scheduler.write().await;
-                lock.stop(token).await;
-            }
-        }
+    async fn stop_io(&self) -> Result<()> {
+        self.scheduler.write().await.stop().await?;
+        Ok(())
    }
 }

@@ -64,15 +64,20 @@ impl Events {
 pub struct EventEmitter(Receiver<Event>);

 impl EventEmitter {
-    /// Blocking recv of an event. Return `None` if the `Sender` has been droped.
+    /// Blocking recv of an event. Return `None` if the `Sender` has been dropped.
    pub fn recv_sync(&self) -> Option<Event> {
        async_std::task::block_on(self.recv())
    }

-    /// Async recv of an event. Return `None` if the `Sender` has been droped.
+    /// Async recv of an event. Return `None` if the `Sender` has been dropped.
    pub async fn recv(&self) -> Option<Event> {
        self.0.recv().await.ok()
    }
+
+    /// Closes event emitter.
+    pub async fn close(&mut self) {
+        self.0.close();
+    }
 }

 impl async_std::stream::Stream for EventEmitter {
@@ -14,7 +14,6 @@ use crate::context::Context;
 use crate::dc_tools::time;
 use crate::events::EventType;
 use crate::imap::Imap;
-use crate::location;
 use crate::message::{Message, MsgId};
 use crate::mimefactory::MimeFactory;
 use crate::param::{Param, Params};
@@ -90,8 +89,6 @@ pub enum Action {
    ResyncFolders = 300,

    // Jobs in the SMTP-thread, range from DC_SMTP_THREAD..DC_SMTP_THREAD+999
-    MaybeSendLocations = 5005, // low priority ...
-    MaybeSendLocationsEnded = 5007,
    SendMdn = 5010,
 }

@@ -105,8 +102,6 @@ impl From<Action> for Thread {
            UpdateRecentQuota => Thread::Imap,
            DownloadMsg => Thread::Imap,

-            MaybeSendLocations => Thread::Smtp,
-            MaybeSendLocationsEnded => Thread::Smtp,
            SendMdn => Thread::Smtp,
        }
    }
@@ -571,10 +566,6 @@ async fn perform_job_action(

    let try_res = match job.action {
        Action::SendMdn => job.send_mdn(context, connection.smtp()).await,
-        Action::MaybeSendLocations => location::job_maybe_send_locations(context, job).await,
-        Action::MaybeSendLocationsEnded => {
-            location::job_maybe_send_locations_ended(context, job).await
-        }
        Action::ResyncFolders => job.resync_folders(context, connection.inbox()).await,
        Action::FetchExistingMsgs => job.fetch_existing_msgs(context, connection.inbox()).await,
        Action::UpdateRecentQuota => match context.update_recent_quota(connection.inbox()).await {
@@ -647,7 +638,7 @@ pub async fn add(context: &Context, job: Job) -> Result<()> {
                info!(context, "interrupt: imap");
                context.interrupt_inbox(InterruptInfo::new(false)).await;
            }
-            Action::MaybeSendLocations | Action::MaybeSendLocationsEnded | Action::SendMdn => {
+            Action::SendMdn => {
                info!(context, "interrupt: smtp");
                context.interrupt_smtp(InterruptInfo::new(false)).await;
            }
@@ -91,16 +91,17 @@ impl DcKey for SignedPublicKey {
    type KeyType = SignedPublicKey;

    async fn load_self(context: &Context) -> Result<Self::KeyType> {
+        let addr = context.get_primary_self_addr().await?;
        match context
            .sql
            .query_row_optional(
                r#"
            SELECT public_key
              FROM keypairs
-             WHERE addr=(SELECT value FROM config WHERE keyname="configured_addr")
+             WHERE addr=?
               AND is_default=1;
            "#,
-                paramsv![],
+                paramsv![addr],
                |row| {
                    let bytes: Vec<u8> = row.get(0)?;
                    Ok(bytes)
@@ -1,19 +1,20 @@
 //! Location handling.
 use std::convert::TryFrom;

-use anyhow::{ensure, Result};
+use anyhow::{ensure, Context as _, Result};
+use async_std::channel::Receiver;
+use async_std::future::timeout;
 use bitflags::bitflags;
 use quick_xml::events::{BytesEnd, BytesStart, BytesText};
+use std::time::Duration;

 use crate::chat::{self, ChatId};
 use crate::contact::ContactId;
 use crate::context::Context;
-use crate::dc_tools::time;
+use crate::dc_tools::{duration_to_str, time};
 use crate::events::EventType;
-use crate::job::{self, Job};
 use crate::message::{Message, MsgId, Viewtype};
 use crate::mimeparser::SystemMessage;
-use crate::param::Params;
 use crate::stock_str;

 /// Location record
@@ -226,32 +227,11 @@ pub async fn send_locations_to_chat(
    }
    context.emit_event(EventType::ChatModified(chat_id));
    if 0 != seconds {
-        schedule_maybe_send_locations(context, false).await?;
-        job::add(
-            context,
-            job::Job::new(
-                job::Action::MaybeSendLocationsEnded,
-                chat_id.to_u32(),
-                Params::new(),
-                seconds + 1,
-            ),
-        )
-        .await?;
+        context.interrupt_location().await;
    }
    Ok(())
 }

-async fn schedule_maybe_send_locations(context: &Context, force_schedule: bool) -> Result<()> {
-    if force_schedule || !job::action_exists(context, job::Action::MaybeSendLocations).await? {
-        job::add(
-            context,
-            job::Job::new(job::Action::MaybeSendLocations, 0, Params::new(), 60),
-        )
-        .await?;
-    };
-    Ok(())
-}
-
 /// Returns whether `chat_id` or any chat is sending locations.
 ///
 /// If `chat_id` is `Some` only that chat is checked, otherwise returns `true` if any chat
@@ -318,13 +298,13 @@ pub async fn set(context: &Context, latitude: f64, longitude: f64, accuracy: f64
            ).await {
                warn!(context, "failed to store location {:?}", err);
            } else {
+                info!(context, "stored location for chat {}", chat_id);
                continue_streaming = true;
            }
        }
        if continue_streaming {
            context.emit_event(EventType::LocationChanged(Some(ContactId::SELF)));
        };
-        schedule_maybe_send_locations(context, false).await.ok();
    }

    continue_streaming
@@ -607,147 +587,140 @@ pub(crate) async fn save(
    Ok(newest_location_id)
 }

-pub(crate) async fn job_maybe_send_locations(context: &Context, _job: &Job) -> job::Status {
-    let now = time();
-    let mut continue_streaming = false;
-    info!(
-        context,
-        " ----------------- MAYBE_SEND_LOCATIONS -------------- ",
-    );
+pub(crate) async fn location_loop(context: &Context, interrupt_receiver: Receiver<()>) {
+    loop {
+        let next_event = match maybe_send_locations(context).await {
+            Err(err) => {
+                warn!(context, "maybe_send_locations failed: {}", err);
+                Some(60) // Retry one minute later.
+            }
+            Ok(next_event) => next_event,
+        };

+        let duration = if let Some(next_event) = next_event {
+            Duration::from_secs(next_event)
+        } else {
+            Duration::from_secs(86400)
+        };
+
+        info!(
+            context,
+            "Location loop is waiting for {} or interrupt",
+            duration_to_str(duration)
+        );
+        timeout(duration, interrupt_receiver.recv()).await.ok();
+    }
+}
+
+/// Returns number of seconds until the next time location streaming for some chat ends
+/// automatically.
+async fn maybe_send_locations(context: &Context) -> Result<Option<u64>> {
+    let mut next_event: Option<u64> = None;
+
+    let now = time();
    let rows = context
        .sql
        .query_map(
-            "SELECT id, locations_send_begin, locations_last_sent \
-         FROM chats \
-         WHERE locations_send_until>?;",
-            paramsv![now],
+            "SELECT id, locations_send_begin, locations_send_until, locations_last_sent
+             FROM chats
+             WHERE locations_send_until>0",
+            [],
            |row| {
                let chat_id: ChatId = row.get(0)?;
                let locations_send_begin: i64 = row.get(1)?;
-                let locations_last_sent: i64 = row.get(2)?;
-                continue_streaming = true;
-
-                // be a bit tolerant as the timer may not align exactly with time(NULL)
-                if now - locations_last_sent < (60 - 3) {
-                    Ok(None)
-                } else {
-                    Ok(Some((chat_id, locations_send_begin, locations_last_sent)))
-                }
+                let locations_send_until: i64 = row.get(2)?;
+                let locations_last_sent: i64 = row.get(3)?;
+                Ok((
+                    chat_id,
+                    locations_send_begin,
+                    locations_send_until,
+                    locations_last_sent,
+                ))
            },
            |rows| {
-                rows.filter_map(|v| v.transpose())
-                    .collect::<std::result::Result<Vec<_>, _>>()
+                rows.collect::<std::result::Result<Vec<_>, _>>()
                    .map_err(Into::into)
            },
        )
-        .await;
+        .await
+        .context("failed to query location streaming chats")?;

-    if let Ok(rows) = rows {
-        let mut msgs = Vec::new();
-
-        {
-            let conn = job_try!(context.sql.get_conn().await);
-
-            let mut stmt_locations = job_try!(conn.prepare_cached(
-                "SELECT id \
+    for (chat_id, locations_send_begin, locations_send_until, locations_last_sent) in rows {
+        if locations_send_begin > 0 && locations_send_until > now {
+            let can_send = now > locations_last_sent + 60;
+            let has_locations = context
+                .sql
+                .exists(
+                    "SELECT COUNT(id) \
     FROM locations \
     WHERE from_id=? \
     AND timestamp>=? \
     AND timestamp>? \
-     AND independent=0 \
-             ORDER BY timestamp;",
-            ));
+     AND independent=0",
+                    paramsv![ContactId::SELF, locations_send_begin, locations_last_sent,],
+                )
+                .await?;

-            for (chat_id, locations_send_begin, locations_last_sent) in &rows {
-                if !stmt_locations
-                    .exists(paramsv![
-                        ContactId::SELF,
-                        *locations_send_begin,
-                        *locations_last_sent,
-                    ])
-                    .unwrap_or_default()
-                {
-                    // if there is no new location, there's nothing to send.
-                    // however, maybe we want to bypass this test eg. 15 minutes
-                } else {
-                    // pending locations are attached automatically to every message,
+            next_event = next_event
+                .into_iter()
+                .chain(u64::try_from(locations_send_until - now).into_iter())
+                .min();
+
+            if has_locations {
+                if can_send {
+                    // Send location-only message.
+                    // Pending locations are attached automatically to every message,
                    // so also to this empty text message.
-                    // DC_CMD_LOCATION is only needed to create a nicer subject.
-                    //
-                    // for optimisation and to avoid flooding the sending queue,
-                    // we could sending these messages only if we're really online.
-                    // the easiest way to determine this, is to check for an empty message queue.
-                    // (might not be 100%, however, as positions are sent combined later
-                    // and dc_set_location() is typically called periodically, this is ok)
+                    info!(
+                        context,
+                        "Chat {} has pending locations, sending them.", chat_id
+                    );
                    let mut msg = Message::new(Viewtype::Text);
                    msg.hidden = true;
                    msg.param.set_cmd(SystemMessage::LocationOnly);
-                    msgs.push((*chat_id, msg));
+                    chat::send_msg(context, chat_id, &mut msg).await?;
+                } else {
+                    // Wait until pending locations can be sent.
+                    info!(
+                        context,
+                        "Chat {} has pending locations, but they can't be sent yet.", chat_id
+                    );
+                    next_event = next_event
+                        .into_iter()
+                        .chain(u64::try_from(locations_last_sent + 61 - now).into_iter())
+                        .min();
                }
+            } else {
+                info!(
+                    context,
+                    "Chat {} has location streaming enabled, but no pending locations.", chat_id
+                );
            }
-        }
-
-        for (chat_id, mut msg) in msgs.into_iter() {
-            // TODO: better error handling
-            chat::send_msg(context, chat_id, &mut msg)
-                .await
-                .unwrap_or_default();
-        }
-    }
-
-    if continue_streaming {
-        job_try!(schedule_maybe_send_locations(context, true).await);
-    }
-    job::Status::Finished(Ok(()))
-}
-
-pub(crate) async fn job_maybe_send_locations_ended(
-    context: &Context,
-    job: &mut Job,
-) -> job::Status {
-    // this function is called when location-streaming _might_ have ended for a chat.
-    // the function checks, if location-streaming is really ended;
-    // if so, a device-message is added if not yet done.
-
-    let chat_id = ChatId::new(job.foreign_id);
-
-    let (send_begin, send_until) = job_try!(
-        context
-            .sql
-            .query_row(
-                "SELECT locations_send_begin, locations_send_until  FROM chats  WHERE id=?",
-                paramsv![chat_id],
-                |row| Ok((row.get::<_, i64>(0)?, row.get::<_, i64>(1)?)),
-            )
-            .await
-    );
-
-    let now = time();
-    if !(send_begin != 0 && now <= send_until) {
-        // still streaming -
-        // may happen as several calls to dc_send_locations_to_chat()
-        // do not un-schedule pending DC_MAYBE_SEND_LOC_ENDED jobs
-        if !(send_begin == 0 && send_until == 0) {
-            // not streaming, device-message already sent
-            job_try!(
-                context
-                    .sql
-                    .execute(
-                        "UPDATE chats \
-                             SET locations_send_begin=0, locations_send_until=0 \
- WHERE id=?",
-                        paramsv![chat_id],
-                    )
-                    .await
+        } else {
+            // Location streaming was either explicitly disabled (locations_send_begin = 0) or
+            // locations_send_until is in the past.
+            info!(
+                context,
+                "Disabling location streaming for chat {}.", chat_id
            );
+            context
+                .sql
+                .execute(
+                    "UPDATE chats \
+                         SET locations_send_begin=0, locations_send_until=0 \
+                         WHERE id=?",
+                    paramsv![chat_id],
+                )
+                .await
+                .context("failed to disable location streaming")?;

            let stock_str = stock_str::msg_location_disabled(context).await;
-            job_try!(chat::add_info_msg(context, chat_id, &stock_str, now).await);
+            chat::add_info_msg(context, chat_id, &stock_str, now).await?;
            context.emit_event(EventType::ChatModified(chat_id));
        }
    }
-    job::Status::Finished(Ok(()))
+
+    Ok(next_event)
 }

 #[cfg(test)]
@@ -44,7 +44,7 @@ static P_AKTIVIX_ORG: Lazy<Provider> = Lazy::new(|| Provider {
            protocol: Smtp,
            socket: Starttls,
            hostname: "newyear.aktivix.org",
-            port: 25,
+            port: 587,
            username_pattern: Email,
        },
    ],
@@ -851,10 +851,6 @@ static P_NAUTA_CU: Lazy<Provider> = Lazy::new(|| Provider {
            key: Config::MvboxMove,
            value: "0",
        },
-        ConfigDefault {
-            key: Config::E2eeEnabled,
-            value: "0",
-        },
        ConfigDefault {
            key: Config::MediaQuality,
            value: "1",
@@ -983,7 +979,7 @@ static P_QQ: Lazy<Provider> = Lazy::new(|| {
    overview_page: "https://providers.delta.chat/qq",
    server: vec![
        Server { protocol: Imap, socket: Ssl, hostname: "imap.qq.com", port: 993, username_pattern: Emaillocalpart },
-        Server { protocol: Smtp, socket: Starttls, hostname: "smtp.qq.com", port: 465, username_pattern: Email },
+        Server { protocol: Smtp, socket: Ssl, hostname: "smtp.qq.com", port: 465, username_pattern: Email },
    ],
    config_defaults: None,
    strict_tls: true,
@@ -1402,7 +1398,9 @@ static P_YGGMAIL: Lazy<Provider> = Lazy::new(|| {
        Server { protocol: Imap, socket: Plain, hostname: "localhost", port: 1143, username_pattern: Email },
        Server { protocol: Smtp, socket: Plain, hostname: "localhost", port: 1025, username_pattern: Email },
    ],
-    config_defaults: None,
+    config_defaults: Some(vec![
+        ConfigDefault { key: Config::MvboxMove, value: "0" },
+    ]),
    strict_tls: true,
    max_smtp_rcpt_to: None,
    oauth2_authorizer: None,
@@ -1853,4 +1851,4 @@ pub(crate) static PROVIDER_IDS: Lazy<HashMap<&'static str, &'static Provider>> =
 });

 pub static PROVIDER_UPDATED: Lazy<chrono::NaiveDate> =
-    Lazy::new(|| chrono::NaiveDate::from_ymd(2022, 1, 31));
+    Lazy::new(|| chrono::NaiveDate::from_ymd(2022, 5, 3));
@@ -12,6 +12,7 @@ use crate::dc_tools::time;
 use crate::ephemeral::{self, delete_expired_imap_messages};
 use crate::imap::Imap;
 use crate::job::{self, Thread};
+use crate::location;
 use crate::log::LogExt;
 use crate::smtp::{send_smtp_messages, Smtp};
 use crate::sql;
@@ -20,8 +21,6 @@ use self::connectivity::ConnectivityStore;

 pub(crate) mod connectivity;

-pub(crate) struct StopToken;
-
 /// Job and connection scheduler.
 #[derive(Debug)]
 #[allow(clippy::large_enum_variant)]
@@ -38,6 +37,8 @@ pub(crate) enum Scheduler {
        smtp_handle: Option<task::JoinHandle<()>>,
        ephemeral_handle: Option<task::JoinHandle<()>>,
        ephemeral_interrupt_send: Sender<()>,
+        location_handle: Option<task::JoinHandle<()>>,
+        location_interrupt_send: Sender<()>,
    },
 }

@@ -67,6 +68,10 @@ impl Context {
    pub(crate) async fn interrupt_ephemeral_task(&self) {
        self.scheduler.read().await.interrupt_ephemeral_task().await;
    }
+
+    pub(crate) async fn interrupt_location(&self) {
+        self.scheduler.read().await.interrupt_location().await;
+    }
 }

 async fn inbox_loop(ctx: Context, started: Sender<()>, inbox_handlers: ImapConnectionHandlers) {
@@ -76,16 +81,15 @@ async fn inbox_loop(ctx: Context, started: Sender<()>, inbox_handlers: ImapConne
    let ImapConnectionHandlers {
        mut connection,
        stop_receiver,
-        shutdown_sender,
    } = inbox_handlers;

    let ctx1 = ctx.clone();
    let fut = async move {
-        started
-            .send(())
-            .await
-            .expect("inbox loop, missing started receiver");
        let ctx = ctx1;
+        if let Err(err) = started.send(()).await {
+            warn!(ctx, "inbox loop, missing started receiver: {}", err);
+            return;
+        };

        let mut info = InterruptInfo::default();
        loop {
@@ -131,10 +135,6 @@ async fn inbox_loop(ctx: Context, started: Sender<()>, inbox_handlers: ImapConne
        })
        .race(fut)
        .await;
-    shutdown_sender
-        .send(())
-        .await
-        .expect("inbox loop, missing shutdown receiver");
 }

 async fn fetch_idle(ctx: &Context, connection: &mut Imap, folder: Config) -> InterruptInfo {
@@ -259,17 +259,16 @@ async fn simple_imap_loop(
    let ImapConnectionHandlers {
        mut connection,
        stop_receiver,
-        shutdown_sender,
    } = inbox_handlers;

    let ctx1 = ctx.clone();

    let fut = async move {
-        started
-            .send(())
-            .await
-            .expect("simple imap loop, missing started receive");
        let ctx = ctx1;
+        if let Err(err) = started.send(()).await {
+            warn!(&ctx, "simple imap loop, missing started receiver: {}", err);
+            return;
+        }

        loop {
            fetch_idle(&ctx, &mut connection, folder).await;
@@ -283,10 +282,6 @@ async fn simple_imap_loop(
        })
        .race(fut)
        .await;
-    shutdown_sender
-        .send(())
-        .await
-        .expect("simple imap loop, missing shutdown receiver");
 }

 async fn smtp_loop(ctx: Context, started: Sender<()>, smtp_handlers: SmtpConnectionHandlers) {
@@ -296,17 +291,16 @@ async fn smtp_loop(ctx: Context, started: Sender<()>, smtp_handlers: SmtpConnect
    let SmtpConnectionHandlers {
        mut connection,
        stop_receiver,
-        shutdown_sender,
        idle_interrupt_receiver,
    } = smtp_handlers;

    let ctx1 = ctx.clone();
    let fut = async move {
-        started
-            .send(())
-            .await
-            .expect("smtp loop, missing started receiver");
        let ctx = ctx1;
+        if let Err(err) = started.send(()).await {
+            warn!(&ctx, "smtp loop, missing started receiver: {}", err);
+            return;
+        }

        let mut timeout = None;
        let mut interrupt_info = Default::default();
@@ -378,15 +372,15 @@ async fn smtp_loop(ctx: Context, started: Sender<()>, smtp_handlers: SmtpConnect
        })
        .race(fut)
        .await;
-    shutdown_sender
-        .send(())
-        .await
-        .expect("smtp loop, missing shutdown receiver");
 }

 impl Scheduler {
-    /// Start the scheduler, panics if it is already running.
+    /// Start the scheduler, returns error if it is already running.
    pub async fn start(&mut self, ctx: Context) -> Result<()> {
+        if self.is_running() {
+            bail!("scheduler is already stopped");
+        }
+
        let (mvbox, mvbox_handlers) = ImapConnectionState::new(&ctx).await?;
        let (sentbox, sentbox_handlers) = ImapConnectionState::new(&ctx).await?;
        let (smtp, smtp_handlers) = SmtpConnectionState::new();
@@ -399,6 +393,7 @@ impl Scheduler {
        let mut sentbox_handle = None;
        let (smtp_start_send, smtp_start_recv) = channel::bounded(1);
        let (ephemeral_interrupt_send, ephemeral_interrupt_recv) = channel::bounded(1);
+        let (location_interrupt_send, location_interrupt_recv) = channel::bounded(1);

        let inbox_handle = {
            let ctx = ctx.clone();
@@ -422,7 +417,7 @@ impl Scheduler {
            mvbox_start_send
                .send(())
                .await
-                .expect("mvbox start send, missing receiver");
+                .context("mvbox start send, missing receiver")?;
            mvbox_handlers
                .connection
                .connectivity
@@ -445,7 +440,7 @@ impl Scheduler {
            sentbox_start_send
                .send(())
                .await
-                .expect("sentbox start send, missing receiver");
+                .context("sentbox start send, missing receiver")?;
            sentbox_handlers
                .connection
                .connectivity
@@ -467,6 +462,13 @@ impl Scheduler {
            }))
        };

+        let location_handle = {
+            let ctx = ctx.clone();
+            Some(task::spawn(async move {
+                location::location_loop(&ctx, location_interrupt_recv).await;
+            }))
+        };
+
        *self = Scheduler::Running {
            inbox,
            mvbox,
@@ -478,6 +480,8 @@ impl Scheduler {
            smtp_handle,
            ephemeral_handle,
            ephemeral_interrupt_send,
+            location_handle,
+            location_interrupt_send,
        };

        // wait for all loops to be started
@@ -553,11 +557,21 @@ impl Scheduler {
        }
    }

-    /// Halts the scheduler, must be called first, and then `stop`.
-    pub(crate) async fn pre_stop(&self) -> StopToken {
+    async fn interrupt_location(&self) {
+        if let Scheduler::Running {
+            ref location_interrupt_send,
+            ..
+        } = self
+        {
+            location_interrupt_send.try_send(()).ok();
+        }
+    }
+
+    /// Halt the scheduler.
+    pub(crate) async fn stop(&mut self) -> Result<()> {
        match self {
            Scheduler::Stopped => {
-                panic!("WARN: already stopped");
+                bail!("scheduler is already stopped");
            }
            Scheduler::Running {
                inbox,
@@ -568,40 +582,23 @@ impl Scheduler {
                sentbox_handle,
                smtp,
                smtp_handle,
+                ephemeral_handle,
+                location_handle,
                ..
            } => {
                if inbox_handle.is_some() {
-                    inbox.stop().await;
+                    inbox.stop().await?;
                }
                if mvbox_handle.is_some() {
-                    mvbox.stop().await;
+                    mvbox.stop().await?;
                }
                if sentbox_handle.is_some() {
-                    sentbox.stop().await;
+                    sentbox.stop().await?;
                }
                if smtp_handle.is_some() {
-                    smtp.stop().await;
+                    smtp.stop().await?;
                }

-                StopToken
-            }
-        }
-    }
-
-    /// Halt the scheduler, must only be called after pre_stop.
-    pub(crate) async fn stop(&mut self, _t: StopToken) {
-        match self {
-            Scheduler::Stopped => {
-                panic!("WARN: already stopped");
-            }
-            Scheduler::Running {
-                inbox_handle,
-                mvbox_handle,
-                sentbox_handle,
-                smtp_handle,
-                ephemeral_handle,
-                ..
-            } => {
                if let Some(handle) = inbox_handle.take() {
                    handle.await;
                }
@@ -617,8 +614,12 @@ impl Scheduler {
                if let Some(handle) = ephemeral_handle.take() {
                    handle.cancel().await;
                }
+                if let Some(handle) = location_handle.take() {
+                    handle.cancel().await;
+                }

                *self = Scheduler::Stopped;
+                Ok(())
            }
        }
    }
@@ -632,8 +633,6 @@ impl Scheduler {
 /// Connection state logic shared between imap and smtp connections.
 #[derive(Debug)]
 struct ConnectionState {
-    /// Channel to notify that shutdown has completed.
-    shutdown_receiver: Receiver<()>,
    /// Channel to interrupt the whole connection.
    stop_sender: Sender<()>,
    /// Channel to interrupt idle.
@@ -644,14 +643,13 @@ struct ConnectionState {

 impl ConnectionState {
    /// Shutdown this connection completely.
-    async fn stop(&self) {
+    async fn stop(&self) -> Result<()> {
        // Trigger shutdown of the run loop.
        self.stop_sender
            .send(())
            .await
-            .expect("stop, missing receiver");
-        // Wait for a notification that the run loop has been shutdown.
-        self.shutdown_receiver.recv().await.ok();
+            .context("failed to stop, missing receiver")?;
+        Ok(())
    }

    async fn interrupt(&self, info: InterruptInfo) {
@@ -668,18 +666,15 @@ pub(crate) struct SmtpConnectionState {
 impl SmtpConnectionState {
    fn new() -> (Self, SmtpConnectionHandlers) {
        let (stop_sender, stop_receiver) = channel::bounded(1);
-        let (shutdown_sender, shutdown_receiver) = channel::bounded(1);
        let (idle_interrupt_sender, idle_interrupt_receiver) = channel::bounded(1);

        let handlers = SmtpConnectionHandlers {
            connection: Smtp::new(),
            stop_receiver,
-            shutdown_sender,
            idle_interrupt_receiver,
        };

        let state = ConnectionState {
-            shutdown_receiver,
            stop_sender,
            idle_interrupt_sender,
            connectivity: handlers.connection.connectivity.clone(),
@@ -696,15 +691,15 @@ impl SmtpConnectionState {
    }

    /// Shutdown this connection completely.
-    async fn stop(&self) {
-        self.state.stop().await;
+    async fn stop(&self) -> Result<()> {
+        self.state.stop().await?;
+        Ok(())
    }
 }

 struct SmtpConnectionHandlers {
    connection: Smtp,
    stop_receiver: Receiver<()>,
-    shutdown_sender: Sender<()>,
    idle_interrupt_receiver: Receiver<InterruptInfo>,
 }

@@ -717,17 +712,14 @@ impl ImapConnectionState {
    /// Construct a new connection.
    async fn new(context: &Context) -> Result<(Self, ImapConnectionHandlers)> {
        let (stop_sender, stop_receiver) = channel::bounded(1);
-        let (shutdown_sender, shutdown_receiver) = channel::bounded(1);
        let (idle_interrupt_sender, idle_interrupt_receiver) = channel::bounded(1);

        let handlers = ImapConnectionHandlers {
            connection: Imap::new_configured(context, idle_interrupt_receiver).await?,
            stop_receiver,
-            shutdown_sender,
        };

        let state = ConnectionState {
-            shutdown_receiver,
            stop_sender,
            idle_interrupt_sender,
            connectivity: handlers.connection.connectivity.clone(),
@@ -744,8 +736,9 @@ impl ImapConnectionState {
    }

    /// Shutdown this connection completely.
-    async fn stop(&self) {
-        self.state.stop().await;
+    async fn stop(&self) -> Result<()> {
+        self.state.stop().await?;
+        Ok(())
    }
 }

@@ -753,7 +746,6 @@ impl ImapConnectionState {
 struct ImapConnectionHandlers {
    connection: Imap,
    stop_receiver: Receiver<()>,
-    shutdown_sender: Sender<()>,
 }

 #[derive(Default, Debug)]
@@ -420,7 +420,7 @@ pub(crate) async fn send_msg_to_smtp(
    }
    info!(
        context,
-        "Retry number {} to send message {} over SMTP", retries, msg_id
+        "Try number {} to send message {} over SMTP", retries, msg_id
    );

    let recipients_list = recipients
Author	SHA1	Message	Date
link2xt	69ea56eb75	python: attempt to use dc_event_emitter_close()	2022-05-03 22:44:26 +00:00
link2xt	eb3242e077	Add dc_event_emitter_close() API dc_event_emitter_close() can be used to terminate event loop from another thread without dereferencing the context which may be in use by the event loop at the same time.	2022-05-03 22:44:26 +00:00
bjoern	95a54a43ff	update provider database (#3284 ) ran `./src/provider/update.py ../provider-db/_providers/ > src/provider/data.rs`	2022-05-03 23:57:53 +02:00
holger krekel	ca59cbc898	make direct_imap idle usage safe from tests, so we never miss finishing direct_imap idle	2022-05-03 22:56:23 +02:00
holger krekel	21f72ad845	new --debug-setup option to show events during account setup.	2022-05-03 22:50:16 +02:00
holger krekel	40fd98d580	show logs for pseudo configured accounts (for online ones they are instantiated later)	2022-05-03 22:50:16 +02:00
holger krekel	0422d751d8	avoid instatiating a full blown Account object just for showing some core info	2022-05-03 19:35:03 +02:00
holger krekel	e44f68db83	prevent one strange traceback probably caused by ongoing shutdown	2022-05-03 19:35:03 +02:00
holger krekel	88cbf42c39	fix various tests	2022-05-03 19:35:03 +02:00
holger krekel	45157f79a3	fix examples	2022-05-03 19:35:03 +02:00
holger krekel	f5157392b6	some renaming and some docstrings	2022-05-03 19:35:03 +02:00
holger krekel	f631ec3a7c	shift startio/init machinery into PendingConfigure class	2022-05-03 19:35:03 +02:00
holger krekel	a9b2750ec9	slight renamings	2022-05-03 19:35:03 +02:00
holger krekel	7fbdb42695	remove _configtracker and write a tested "PendingConfigure" helper class for managing configure/started states for test accounts	2022-05-03 19:35:03 +02:00
holger krekel	026dd4480e	add --extra-info CLI option (defaults to False for interactive runs)	2022-05-03 19:35:03 +02:00
holger krekel	b373e7acba	more renames	2022-05-03 19:35:03 +02:00
holger krekel	d2ca54c167	provide for meaningful names for bringing accounts online for test functions	2022-05-03 19:35:03 +02:00
holger krekel	e070284a09	strike setting "displayname" in test plugin and refine two tests which implicitely relied on this	2022-05-03 19:35:03 +02:00
holger krekel	e4e02ea3c0	improve error messages during live creation of accounts	2022-05-03 19:35:03 +02:00
holger krekel	a947980eb6	slight reorg for creating accounts from liveconfig	2022-05-03 19:35:03 +02:00
holger krekel	f11c3dd3e3	fix bench test	2022-05-03 19:35:03 +02:00
holger krekel	394067be63	refactor session_liveconfig_producer to become a more geneal testprocess management object	2022-05-03 19:35:03 +02:00
holger krekel	2494613583	- perform direct_imap init in testplugin instead of global deltachat plugin, probably also helping to avoid some segfeaults during teardown - some API renaming on the side (too hard to split into separate commit, sorry)	2022-05-03 19:35:03 +02:00
holger krekel	77c60e7450	refine test support function name	2022-05-03 19:35:03 +02:00
holger krekel	04dd2d93d0	trim online account creation to a single get_online_accounts() function	2022-05-03 19:35:03 +02:00
holger krekel	5e5710ecce	streamline configuration handling for test accounts, removing one layer of flags	2022-05-03 19:35:03 +02:00
holger krekel	c1b33a66c4	refactor "quiet" parameter away and provide more precise logging-control	2022-05-03 19:35:03 +02:00
holger krekel	0e3165d724	unify get_unconfigured_account and make_account	2022-05-03 19:35:03 +02:00
holger krekel	3e12eab0b5	remove make_account from get_online_config and rename the latter to get_next_liveconfig to avoid one indirection	2022-05-03 19:35:03 +02:00
holger krekel	87365e4a43	- remove superflous early set_configs to separate config / account making better - avoid low-level dc_* API access from testplugin	2022-05-03 19:35:03 +02:00
holger krekel	3e16a47ff2	remove unnccesary distinction between offline/online accounts in make_account, simplifying api/usage	2022-05-03 19:35:03 +02:00
holger krekel	874054148e	streamline session_live_config implementation and usage	2022-05-03 19:35:03 +02:00
holger krekel	5e39a13bf6	refine waiting for initial startup waiting for "INBOX: Idle" ready this slows down initialization for tests but provides more stability in my runs	2022-05-03 19:35:03 +02:00
holger krekel	720c1b5eca	normalize make_account usage	2022-05-03 19:35:03 +02:00
holger krekel	c7c1a04c6a	move large inlined AccountMaker (renamed to ACFactory) to proper class instead of being defined in closure	2022-05-03 19:35:03 +02:00
holger krekel	fd5b224ba0	simplify test setup api by removing pre_generated_keys arguments	2022-05-03 19:35:03 +02:00
holger krekel	0b80ade3ae	fix "retry" wording, it isn't but a try :)	2022-05-03 09:31:38 +02:00
bjoern	e1c3e95307	prepare 1.79 (#3281 ) * update changelog for 1.79.0 * bump version to 1.79.0	2022-05-02 16:18:00 +02:00
link2xt	904e8966c0	Replace location jobs with async location loop Locations are now sent in the background regardless of whether SMTP loop is interrupted or not.	2022-05-01 23:08:34 +00:00
link2xt	3a10f0155f	Remove panics from the scheduler and simplify start/stop_io() Hold scheduler lock during the whole procedure of scheduler starting and stopping. This ensures that two processes can't get two read locks in parallel and start loops or send the stop signal twice. Also remove shutdown channels: it is enough to wait for the loop handle without receiving a shutdown signal from the end of the loop.	2022-04-30 18:08:01 +00:00
Hocuri	4c9cc4f3d4	Hopefully make test_connectivity() less flaky	2022-04-30 18:07:39 +02:00
link2xt	48f2c4e14b	Correctly escape messages consisting of a dot in SMTP protocol Actual bugfix is in the async-smtp crate.	2022-04-30 13:30:31 +00:00
holger krekel	f41df327a9	add a little bench file -- "pytest tests/bench_empty.py --durations=10" will tell you how much overhead there is	2022-04-30 13:27:02 +02:00
holger krekel	3f9e3038b7	strike last hard-coded ref to configured_addr	2022-04-30 13:26:50 +02:00