From cb35fb8f59db12963aebae629f05c45d26f98499 Mon Sep 17 00:00:00 2001 From: Jason Rhinelander Date: Thu, 10 Sep 2026 16:11:43 -0300 Subject: [PATCH 1/5] Give Core's components a threading contract, and enforce it Core's components expose database and config access with no stated thread contract, while every Client method carefully dispatches onto Core's loop. Nothing distinguishes the two halves: `client.set_display_name(name, cb)` is correct and `client.core.configs.user_profile().set_name(name)` compiles just as cleanly and is a data race. The header comment on `globals` invites the second outright ("can also be used by the application to persist settings"). The database itself is not the problem -- session-sqlite hands each thread its own connection and says to use it that way. What is unsynchronised is everything a component holds beside its tables: `_adopt_seed` rewrites a std::string and a secure_buffer that the loop reads while polling, and the config objects are built lazily and then mutated by `merge()`. There is not an atomic or a mutex anywhere in globals, configs or devices. So `CoreComponent` grows `on_loop()`, and every method that touches component state asserts it. `on_loop()` is also true during construction, since a component's `init()` necessarily runs on the constructing thread and no other thread can have reached it yet. Where an application legitimately calls one of these from its own thread, the method now comes in the two forms Client already uses -- a `failable_function` and a `block_t` -- and nothing else. Not three: `call_get` runs the job inline when it is already the loop thread, so code already there uses the blocking form and pays nothing. That covers `create_account`, `restore_account`, `device_info`, `update_info` and `build_link_request`. `Globals`' get/set/erase stay as they are and say why: one self-contained query each, touching nothing cached. `failable_function` and `block_t` move to so Core can use the same convention; `session::client` re-exports them, so nothing that names them changes. The assertion immediately found a live bug, fixed here too because the suite does not pass without it. `Network` builds its own `quic::Loop`, so a `send_request` completion handler runs on the network's thread -- and `Core::_send_poll` called `_handle_poll_response` straight from one, merging configs and flushing their dumps off Core's loop on every single poll. It is now marshalled, and `TestHelper::drain` lets the tests that drive a response by hand wait for it the way production does. Two related lifetime fixes fall out of the same reasoning. Core gains a `JobQueue` of its own, declared last so it stops -- cancelling outstanding component work -- before the components those jobs reach are destroyed, and while `_loop` is still alive to process the stop. And Client's `_jq` existed to cancel its deferred work on destruction but carried exactly one job: `_async` and nine other sites deferred onto the loop's own queue instead, which is not emptied until `~Loop`, the last thing `~Core` does. Every one of those jobs holds `this` and reaches through it into Core's members, so they were being drained throughout the destruction of every one of them. They now go on `_jq`; `call_get` stays on the loop, since its caller is blocked inside it and cannot have gone away. --- include/session/client.hpp | 15 ++++- include/session/client/handler.hpp | 52 ++------------- include/session/core.hpp | 33 +++++++++- include/session/core/component.hpp | 100 ++++++++++++++++++++++++++++- include/session/core/configs.hpp | 11 ++++ include/session/core/devices.hpp | 29 ++++++++- include/session/core/globals.hpp | 32 ++++++++- include/session/handler.hpp | 55 ++++++++++++++++ src/client/client.cpp | 40 ++++++------ src/core.cpp | 26 ++++++-- src/core/component.cpp | 25 ++++++++ src/core/configs.cpp | 19 ++++-- src/core/devices.cpp | 38 +++++++++-- src/core/globals.cpp | 29 ++++++++- tests/test_core_devices.cpp | 74 ++++++++++----------- tests/test_core_globals.cpp | 6 +- tests/test_helper.hpp | 11 ++++ tests/test_poll.cpp | 18 ++++-- 18 files changed, 474 insertions(+), 139 deletions(-) create mode 100644 include/session/handler.hpp diff --git a/include/session/client.hpp b/include/session/client.hpp index 0d2240b78..480a23f6f 100644 --- a/include/session/client.hpp +++ b/include/session/client.hpp @@ -1181,7 +1181,11 @@ class Client { // leave them waiting for an answer that is never coming. template void _async(Produce produce, Cb cb) { - loop.call([this, produce = std::move(produce), cb = std::move(cb)]() mutable { + // `_jq`, not `loop`: this captures `this` and runs later, and ~Client has to be able to + // throw it away. On the loop's own queue it would instead run during Core's destruction + // -- the loop thread keeps draining until ~Loop, which is the *last* thing ~Core does -- + // reaching a Client whose components have already gone. + _jq.call([this, produce = std::move(produce), cb = std::move(cb)]() mutable { using Result = decltype(produce()); try { if constexpr (std::is_void_v) { @@ -1256,6 +1260,15 @@ class Client { // would mean reporting a change to the subscribers of a Client that is going away, against a // Core whose database is already being torn down. // + // **Everything this class defers must go here**, not on `loop`. The loop's own queue is not + // emptied until `~Loop`, which is the last thing `~Core` does, so a job left on it keeps being + // drained by the loop thread throughout the destruction of every Core member -- and every one + // of these jobs holds `this` and reaches through it into those members. Stopping this queue + // is the first thing `~Client` does, while all of that is still whole. + // + // `loop.call_get()` is the exception and stays as it is: the calling thread is blocked inside + // it, so there is no window in which the caller can have gone away. + // // Declared after `core` -- the one thing that belongs below it -- because a JobQueue needs its // loop alive in order to stop, so it has to be destroyed while Core still exists. oxen::quic::JobQueue _jq{loop}; diff --git a/include/session/client/handler.hpp b/include/session/client/handler.hpp index cdad8cd12..b473f8fbc 100644 --- a/include/session/client/handler.hpp +++ b/include/session/client/handler.hpp @@ -1,29 +1,16 @@ #pragma once #include -#include -#include +#include namespace session::client { -/// Passed where a handler would go, to say "wait until this is done and give me the answer" -/// instead. -/// -/// Every asynchronous method has a blocking twin taking one of these. The work is the same and -/// happens in the same place -- on Core's loop -- so the only difference is who waits: the twin -/// blocks the calling thread until the answer is ready, and *throws* what the handler form would -/// have reported through its `error` argument. -/// -/// A tag rather than a second class, and rather than an overload with no handler at all, because -/// the point is that it be visible where it is used. Blocking is a decision about the calling -/// thread, so it belongs at the call site: a render loop must not do it, and a review can grep for -/// it, neither of which works when the choice was made wherever the variable was declared. -/// -/// Calling one from a Client handler is safe rather than a deadlock -- the loop runs the work -/// inline when it is already the current thread -- but it is still waiting, and anything else the -/// loop owes is waiting behind it. -struct await_t {}; -inline constexpr await_t await{}; +/// These are `session::`'s, and are used unqualified throughout this namespace. They live one +/// level up because Core's components hand out the same shapes with the same convention, and Core +/// cannot depend on Client. See for what each one means. +using session::await; +using session::await_t; +using session::failable_function; /// Runs a job on the application's own thread. /// @@ -42,29 +29,4 @@ inline constexpr await_t await{}; /// program with no loop of its own wants. using dispatcher = std::function)>; -namespace detail { - template - struct failable_function; - - template - struct failable_function { - using type = std::function error, A...)>; - }; -} // namespace detail - -/// A handler an application passes to one of the asynchronous methods, written in terms of what -/// that method produces: `failable_function` is a handler taking a -/// message id. -/// -/// What it adds is the leading `error` argument every one of them carries -- unset when the call -/// succeeded, and otherwise saying what went wrong. Every such handler is invoked exactly once, -/// unless the Client is destroyed before its work runs, so a caller is never left waiting on an -/// answer that is not coming; the error argument is how a failure says so, since a call that has -/// been dispatched has no caller left to throw to. -/// -/// Written as an alias rather than spelled out at each declaration so that the convention is stated -/// once and the argument cannot be forgotten or put in the wrong place. -template -using failable_function = typename detail::failable_function::type; - } // namespace session::client diff --git a/include/session/core.hpp b/include/session/core.hpp index d1ee78625..67b65a615 100644 --- a/include/session/core.hpp +++ b/include/session/core.hpp @@ -592,9 +592,17 @@ class Core { /// The event loop this account's work runs on. /// /// Everything Core does off the caller's thread — polling, send completion, and therefore every - /// callback it fires — happens here. A layer above Core dispatches its own database work onto - /// it with `loop().call(...)` so that all access is serialised onto one thread, rather than - /// relying on the database being safe to touch from several. + /// callback it fires — happens here. + /// + /// The database itself does not need this: `sqlite::Database` is a pool that hands each thread + /// its own connection, so a self-contained query is safe from anywhere. What needs the loop is + /// everything a component holds *beside* its tables — the cached account keys, the config + /// objects, and the lazy construction of both — none of which is synchronised and all of which + /// polling touches. `detail::CoreComponent` says which methods that covers, and they assert it + /// in a debug build. + /// + /// A layer above Core that keeps its own tables dispatches its own work here for the same + /// reason it would anywhere else: to serialise its own state, not the database's. /// /// `call()` runs the job inline when the caller is already on this thread, so a single-threaded /// application pays nothing for the indirection. @@ -618,6 +626,10 @@ class Core { // Global value storage. This are used by some components, but can also be used by the // application to persist settings. + // + // `get_*`/`set`/`erase` are self-contained queries and are safe to call from any thread; the + // rest of Globals is not. See the threading note on `detail::CoreComponent`, which applies + // to every component below as well. Globals globals{*this}; // Session Pro-related capabilities @@ -637,6 +649,21 @@ class Core { // is_final=true to flush any actions that are deferred until the end of a fetch. void receive_messages( std::span messages, config::Namespace ns, bool is_final); + + private: + // Set at the end of init(), i.e. once construction is complete and another thread could + // reach a component. Read by CoreComponent::on_loop() so that the components' `init()`, + // which necessarily runs on the constructing thread, is not treated as misuse. + bool _constructed = false; + + // Where component work runs. A queue of our own rather than the loop's shared one so that + // whatever is still outstanding is *cancelled* when Core goes away instead of running against + // components that are already destroyed -- the same reason Client keeps its own. + // + // Declared last so it is destroyed first, before the components its jobs reach. It has to be + // destroyed while `_loop` is still alive, which it is: `_loop` is declared first and so is + // destroyed last. + quic::JobQueue _jq{_loop}; }; } // namespace session::core diff --git a/include/session/core/component.hpp b/include/session/core/component.hpp index 450c8a95a..6a5a9b3f1 100644 --- a/include/session/core/component.hpp +++ b/include/session/core/component.hpp @@ -1,10 +1,19 @@ #pragma once +#include +#include +#include +#include +#include +#include +#include + namespace session::sqlite { class Connection; } namespace oxen::quic { class Loop; +class JobQueue; } // namespace oxen::quic namespace session::core { @@ -14,9 +23,32 @@ class Core; struct callbacks; namespace detail { - // Internal base class bridge between Core and the various components of core. This bridge - // can be used to allow components to access selected private parts of core, such as the - // database, without needing components to be direct friends of Core. + + // Reports a component operation that threw, so that `async` does not have to make the logging + // category reachable from this header. + void log_component_failure(const std::exception& e); + + /// Internal base class bridge between Core and the various components of core. This bridge + /// can be used to allow components to access selected private parts of core, such as the + /// database, without needing components to be direct friends of Core. + /// + /// ## Threading + /// + /// **A component's own state is Core's loop's, not the caller's.** The database underneath is + /// a thread-safe pool and does not care which thread reads it, but a component is more than + /// its tables: it caches account key material, holds the config objects, and lazily builds + /// both. None of that is synchronised, and Core's loop touches all of it while polling. So + /// anything a component does beyond a self-contained query has to happen on the loop, and + /// every such method asserts `on_loop()` in a debug build. + /// + /// A component method that only reads or writes the database through `conn()` is exempt: the + /// pool hands the calling thread its own connection, and two threads doing that concurrently + /// is the arrangement it exists for. Those methods say so individually. + /// + /// `async()` is how a component offers work to a caller on another thread, and mirrors what + /// `session::client::Client` does with the same `failable_function` convention. Note that + /// Core has no application dispatcher -- that is Client's -- so a handler passed here runs on + /// the loop, exactly as `core::callbacks` do. class CoreComponent { protected: friend class core::Core; @@ -32,6 +64,42 @@ namespace detail { // Returns the event loop for scheduling async work. quic::Loop& loop(); + // Returns Core's job queue, which is where component work belongs. + // + // A queue rather than the loop directly so that Core can *cancel* whatever is still + // outstanding when it goes away, instead of letting those jobs run against components + // that are already being destroyed. Stopping the queue does not stop the loop, which + // Core does not own exclusively once a Network is attached. + quic::JobQueue& jq(); + + // Puts `job` on that queue, running it inline if this already is the loop thread. + // + // Out of line, and taking an erased job rather than being a template, so that `async` + // below can be defined here without this header pulling libquic in front of every + // consumer of core.hpp. A `wait_t` overload wants `jq().call_get()` and its return type, + // so those are written in the component's own translation unit, which includes the loop. + void enqueue(std::function job); + + /// True when it is safe for the calling thread to touch this component's own state: + /// either it is the loop thread, or Core is still being constructed and no other thread + /// can have reached the component yet. + /// + /// Written for `assert(on_loop())` and compiled away with it. + [[nodiscard]] bool on_loop() const; + + /// Runs `produce` on Core's job queue and reports what it produced to `cb`, or reports the + /// reason it could not. + /// + /// This is what makes "`cb` is invoked exactly once" true for everything except a Core + /// that is destroyed with the job still queued, which cancels it: the work is database and + /// config access, which throws on a disk error, and by then the caller's stack is gone -- + /// so the handler they gave us is the only way left to tell them. + /// + /// On failure a handler taking a value is given a default-constructed one alongside the + /// error, which it is being told not to read. + template + void async(Produce produce, Cb cb); + explicit CoreComponent(Core& core); // Default component `init()` does nothing; classes can override this if they want to be @@ -41,6 +109,32 @@ namespace detail { virtual void init() {} }; + template + void CoreComponent::async(Produce produce, Cb cb) { + enqueue([produce = std::move(produce), cb = std::move(cb)]() mutable { + using Result = decltype(produce()); + try { + if constexpr (std::is_void_v) { + produce(); + if (cb) + cb(std::nullopt); + } else { + auto result = produce(); + if (cb) + cb(std::nullopt, std::move(result)); + } + } catch (const std::exception& e) { + log_component_failure(e); + if (!cb) + return; + if constexpr (std::is_void_v) + cb(std::string{e.what()}); + else + cb(std::string{e.what()}, Result{}); + } + }); + } + } // namespace detail } // namespace session::core diff --git a/include/session/core/configs.hpp b/include/session/core/configs.hpp index 32f5d2995..0267fe8b7 100644 --- a/include/session/core/configs.hpp +++ b/include/session/core/configs.hpp @@ -37,6 +37,17 @@ namespace session::core { /// them before an account exists in any case: a network cannot be attached without one, so neither /// polling nor pushing can run, and a direct caller gets the same `no_account` any other /// account-dependent call would throw. +/// +/// **Every method here belongs to Core's loop**, with no exceptions of the kind Globals has: there +/// is no self-contained query on this class. The accessors hand out a reference into an object +/// that `merge()` rewrites on the loop as poll results arrive, and building them is itself lazy, +/// so a caller on another thread races either the construction or the merge. A debug build +/// asserts it. +/// +/// An application does not reach these directly. `session::client::Client` wraps the parts it +/// needs -- `display_name()` and the rest of "our own account" -- and does the hop for the caller; +/// anything else wanting a config from another thread does the same thing, rather than reading one +/// from where it happens to be standing. class Configs : public detail::CoreComponent { friend class session::TestHelper; diff --git a/include/session/core/devices.hpp b/include/session/core/devices.hpp index 15558cd3d..3324bd871 100644 --- a/include/session/core/devices.hpp +++ b/include/session/core/devices.hpp @@ -201,6 +201,11 @@ class Devices final : detail::CoreComponent { // (i.e. we are not in the device group, or all our keys have rotated past this message). std::vector decrypt_device_data(std::span data); + // What both public forms of each dispatch onto the loop, and what the rest of this class + // calls internally, since it is already there. Each asserts it got there. + std::pair _device_info(); + void _update_info(const device::Info& info); + public: // Returns the current device's random identifier, in hex. std::string device_id() const; @@ -215,7 +220,11 @@ class Devices final : detail::CoreComponent { std::span only_device = {}); // Returns *this* device's info and whether it is registered in the device group. - std::pair device_info(); + // + // Reads the device config, which the loop merges into, so it happens on the loop either way: + // code already there uses the `await` form and pays nothing for it. + void device_info(failable_function)> cb); + std::pair device_info(await_t); struct LinkRequestResult { std::vector message; // encrypted bytes to push to Namespace::Devices @@ -227,13 +236,27 @@ class Devices final : detail::CoreComponent { // std::logic_error if it is already registered. The returned message is to be pushed to // Namespace::Devices with a 10-minute TTL. The sas field contains the short authentication // string that should be displayed to the user for verification against the accepting device. - LinkRequestResult build_link_request(); + // + // Reads the device config and writes the pending request, so it happens on the loop either + // way; an application driving a linking screen is on its own thread and wants one of these. + void build_link_request(failable_function cb); + LinkRequestResult build_link_request(await_t); + + private: + LinkRequestResult _build_link_request(); + + public: // Updates this device's info locally to match the given info; if the current device is // registered then this dirties the device config data, requiring a push. // // The state and pk_* fields of the input value are ignored. - void update_info(const device::Info& info); + // + // Filling these in is the application's job -- libsession establishes the group with them + // blank -- and an application is on its own thread when it does. The handler form takes the + // info by value: it outlives the call. + void update_info(device::Info info, failable_function cb); + void update_info(const device::Info& info, await_t); // Creates the account's device group with this device as its only member, if one is owed. // diff --git a/include/session/core/globals.hpp b/include/session/core/globals.hpp index b0cb6a8e9..96c76f0b2 100644 --- a/include/session/core/globals.hpp +++ b/include/session/core/globals.hpp @@ -83,19 +83,35 @@ class Globals final : detail::CoreComponent { // account: that one may already have a group belonging to devices we have not met. void _mark_new_account(); + // What both public forms of each dispatch onto the loop; each asserts it got there. + void _create_account(); + void _restore_account(const predefined_seed& seed); + public: /// Whether this account has an identity yet. /// /// Only ever false when the Core was constructed with defer_account and the database held no /// seed. Until it is true, everything needing the account -- session_id(), account_seed(), /// send_dm(), attaching a network -- throws no_account. + /// + /// Safe to read from any thread once the identity is settled, which is what the rest of this + /// paragraph is about. This and the accessors below read state that `create_account()` and + /// `restore_account()` write, and nothing synchronises the two: an application that reads + /// them while one of those is still in flight on the loop is racing itself. Await the + /// handler, or use the `await` form, and the question does not arise -- an identity never + /// changes again once it exists. bool have_account() const { return _have_account; } /// Generates a fresh account and stores it. /// + /// Two forms and no third: this rewrites the cached key material Core's loop reads, so it + /// happens on the loop either way. Code already there uses the `await` form and pays nothing, + /// since `call_get` runs the job inline when it is already the loop thread. + /// /// @throws std::logic_error if this account already has an identity: adopting a second one /// would orphan every message and key already stored against the first. - void create_account(); + void create_account(failable_function cb); + void create_account(await_t); /// Adopts an existing account seed, as typed from a recovery phrase or transferred from /// another device, and stores it. @@ -104,12 +120,22 @@ class Globals final : detail::CoreComponent { /// is encrypted to the account root key, so the seed must be adopted before /// devices.build_link_request() can be called. /// + /// The handler form takes the seed by value because it outlives the call: it is carried to the + /// loop and zeroed with the job, rather than borrowed from a caller that has already returned. + /// /// @throws std::logic_error if this account already has an identity. - void restore_account(const predefined_seed& seed); + void restore_account(predefined_seed seed, failable_function cb); + void restore_account(const predefined_seed& seed, await_t); public: // Retrieval methods. These query for the given key and, if the type matches, return the given - // value. You get back nullopt if the database key does not exist, or if it contains + // value. You get back nullopt if the database key does not exist, or if it contains a value + // of some other type. + // + // These, `set` and `erase` are the exception to the threading rule on the rest of this class: + // each is one self-contained query against a connection the pool hands the calling thread, and + // touches nothing cached. An application may call them from wherever it likes, and gets + // whatever SQLite's own locking gives it if two threads write the same key at once. std::optional get_integer(std::string_view key); std::optional get_real(std::string_view key); std::optional get_text(std::string_view key); diff --git a/include/session/handler.hpp b/include/session/handler.hpp new file mode 100644 index 000000000..b95db4424 --- /dev/null +++ b/include/session/handler.hpp @@ -0,0 +1,55 @@ +#pragma once + +#include +#include +#include + +namespace session { + +/// Passed where a handler would go, to say "wait until this is done and give me the answer" +/// instead. +/// +/// Every asynchronous method has a blocking twin taking one of these. The work is the same and +/// happens in the same place -- on the owning event loop -- so the only difference is who waits: +/// the twin blocks the calling thread until the answer is ready, and *throws* what the handler +/// form would have reported through its `error` argument. +/// +/// A tag rather than a second class, and rather than an overload with no handler at all, because +/// the point is that it be visible where it is used. Blocking is a decision about the calling +/// thread, so it belongs at the call site: a render loop must not do it, and a review can grep for +/// it, neither of which works when the choice was made wherever the variable was declared. +/// +/// Calling one from a handler is safe rather than a deadlock -- the loop runs the work inline when +/// it is already the current thread -- but it is still waiting, and anything else the loop owes is +/// waiting behind it. That inlining is also what lets code already on the loop use the blocking +/// form and pay nothing for it, which is why there is no third "I am already on the loop" overload +/// of anything. +struct await_t {}; +inline constexpr await_t await{}; + +namespace detail { + template + struct failable_function; + + template + struct failable_function { + using type = std::function error, A...)>; + }; +} // namespace detail + +/// A handler an application passes to one of the asynchronous methods, written in terms of what +/// that method produces: `failable_function` is a handler taking a +/// message id. +/// +/// What it adds is the leading `error` argument every one of them carries -- unset when the call +/// succeeded, and otherwise saying what went wrong. Every such handler is invoked exactly once, +/// unless the object it was given to is destroyed before its work runs, so a caller is never left +/// waiting on an answer that is not coming; the error argument is how a failure says so, since a +/// call that has been dispatched has no caller left to throw to. +/// +/// Written as an alias rather than spelled out at each declaration so that the convention is stated +/// once and the argument cannot be forgotten or put in the wrong place. +template +using failable_function = typename detail::failable_function::type; + +} // namespace session diff --git a/src/client/client.cpp b/src/client/client.cpp index 2dfd30b59..619e94468 100644 --- a/src/client/client.cpp +++ b/src/client/client.cpp @@ -450,11 +450,11 @@ void Client::_emit(std::function invoke) { } void Client::set_dispatcher(dispatcher d) { - loop.call([this, d = std::move(d)]() mutable { _dispatcher = std::move(d); }); + _jq.call([this, d = std::move(d)]() mutable { _dispatcher = std::move(d); }); } void Client::set_high_freq_dispatch_interval(std::chrono::milliseconds interval) { - loop.call([this, interval] { _high_freq_dispatch_interval = interval; }); + _jq.call([this, interval] { _high_freq_dispatch_interval = interval; }); } void Client::_dispatch_out(std::function job) { @@ -802,7 +802,7 @@ void Client::profile_picture( const ConversationId& id, std::function)> on_progress, failable_function>)> cb) { - loop.call([this, id, on_progress = std::move(on_progress), cb = std::move(cb)]() mutable { + _jq.call([this, id, on_progress = std::move(on_progress), cb = std::move(cb)]() mutable { try { _profile_picture(id, std::move(on_progress), std::move(cb)); } catch (const std::exception& e) { @@ -970,7 +970,7 @@ void Client::attachment_data( size_t index, std::function on_progress, failable_function)> cb) { - loop.call([this, message_id, index, on_progress = std::move(on_progress), cb]() mutable { + _jq.call([this, message_id, index, on_progress = std::move(on_progress), cb]() mutable { try { _attachment_data(message_id, index, std::move(on_progress), cb); } catch (const std::exception& e) { @@ -1069,7 +1069,7 @@ void Client::_fetch_cached( // Onto the loop before touching the registry -- this arrives on the network thread, and // `_in_flight` is ours. [this, name](int64_t done, int64_t total, std::optional r) { - loop.call([this, name, done, total, r] { + _jq.call([this, name, done, total, r] { auto found = _in_flight.find(name); if (found == _in_flight.end()) return; @@ -1080,7 +1080,7 @@ void Client::_fetch_cached( }); }, [this, name, store = std::move(store)](std::optional error) { - loop.call([this, name, store, error = std::move(error)]() mutable { + _jq.call([this, name, store, error = std::move(error)]() mutable { auto found = _in_flight.find(name); if (found == _in_flight.end()) return; @@ -1262,14 +1262,14 @@ void Client::save_attachment( // Not _async: what that reports is the *start* of the transfer, and the answer a caller wants // is whether the file arrived, which is minutes away. So the callback is carried down to the // download's own completion, and only the failures that happen before it starts come back here. - loop.call([this, - message_id, - index, - dest = std::move(dest), - on_progress = std::move(on_progress), - cb, - notify_sender, - replace]() mutable { + _jq.call([this, + message_id, + index, + dest = std::move(dest), + on_progress = std::move(on_progress), + cb, + notify_sender, + replace]() mutable { try { _save_attachment( message_id, @@ -2411,10 +2411,10 @@ void Client::_prefetch_picture(sqlite::Connection& c, int64_t account, const std auto& [sid, key] = *row; - loop.call_soon([this, - id = ConversationId::dm(sid), - url, - key = std::vector{key.begin(), key.end()}]() mutable { + _jq.call_soon([this, + id = ConversationId::dm(sid), + url, + key = std::vector{key.begin(), key.end()}]() mutable { _fetch_picture(id, std::move(url), std::move(key)); }); } catch (const std::exception& e) { @@ -3796,7 +3796,7 @@ void Client::_upload_next( // Core's loop's alone. Nobody is waiting on a callback here -- this is Client's // own continuation -- so a failure has to be turned into the message failing, which // is what the application is watching. - loop.call([this, client_id, index, on_upload, plaintext_size, result = std::move(result)] { + _jq.call([this, client_id, index, on_upload, plaintext_size, result = std::move(result)] { try { if (auto* err = std::get_if(&result)) { log::warning( @@ -4408,7 +4408,7 @@ void Client::_save_attachment( // we sent would claim they have a file they may never have opened. A note to self // is exempt: there the recipient is us, so saving it really is the recipient // saving it. - loop.call([this, message_id, index, notify_sender] { + _jq.call([this, message_id, index, notify_sender] { if (_saved_by_recipient(message_id)) _record_saved(message_id, index, clock_now_ms()); diff --git a/src/core.cpp b/src/core.cpp index 666159544..f12560d36 100644 --- a/src/core.cpp +++ b/src/core.cpp @@ -85,6 +85,10 @@ void Core::init() { _comp_init.clear(); _update_polling(); + + // Last: from here a poll can run and another thread can hold this Core, so a component + // touched from anywhere but the loop is misuse rather than construction. + _constructed = true; } void Core::register_comp_init(detail::CoreComponent* c) { @@ -122,9 +126,12 @@ void Core::_update_polling() { void Core::set_poll_interval(std::chrono::milliseconds interval) { // Marshalled onto the loop rather than done here: this replaces the ticker, and creating or // stopping a libevent event from a thread that is not the loop's races the loop itself. (Both - // `_poll_interval` and `_poll_ticker` are otherwise only touched there.) `Loop::call` runs it + // `_poll_interval` and `_poll_ticker` are otherwise only touched there.) `call` runs it // inline when we are already on the loop thread, so this costs nothing in that case. - _loop.call([this, interval] { + // + // On our own queue rather than the loop's, so that a interval change still in flight when Core + // goes away is dropped rather than run against a half-destroyed one. + _jq.call([this, interval] { _poll_interval = interval; if (_poll_ticker) { _poll_ticker->stop(); @@ -269,8 +276,19 @@ SELECT h.hash FROM swarm_hashes h JOIN swarm_nodes n ON n.id = h.node return; } - _handle_poll_response( - std::move(node), std::move(namespaces), std::move(*body), round); + // Onto our own queue: this handler runs on the *network's* loop, which is a + // different thread entirely -- Network builds its own quic::Loop -- and handling a + // poll response merges configs and flushes their dumps, which is only safe on + // ours. It also means a response landing after Core has gone is dropped rather + // than run against a Core that is being torn down. + _jq.call([this, + node = std::move(node), + namespaces = std::move(namespaces), + body = std::move(*body), + round]() mutable { + _handle_poll_response( + std::move(node), std::move(namespaces), std::move(body), round); + }); }); } diff --git a/src/core/component.cpp b/src/core/component.cpp index e5828340a..d813bdf65 100644 --- a/src/core/component.cpp +++ b/src/core/component.cpp @@ -1,3 +1,4 @@ +#include #include #include #include @@ -5,6 +6,14 @@ namespace session::core::detail { +namespace log = oxen::log; + +static auto cat = log::Cat("core.comp"); + +void log_component_failure(const std::exception& e) { + log::warning(cat, "Component operation failed: {}", e.what()); +} + sqlite::Connection CoreComponent::conn() { return core.db.conn(); } @@ -17,6 +26,22 @@ quic::Loop& CoreComponent::loop() { return core._loop; } +quic::JobQueue& CoreComponent::jq() { + return core._jq; +} + +void CoreComponent::enqueue(std::function job) { + core._jq.call(std::move(job)); +} + +bool CoreComponent::on_loop() const { + // Before the end of Core's constructor there is no other thread that could have reached a + // component: the loop runs nothing of ours until `init()` starts polling, and the caller + // constructing Core is the only one holding it. Component `init()` therefore runs off the + // loop legitimately, and asserting otherwise would fire on every open. + return !core._constructed || core._loop.inside(); +} + CoreComponent::CoreComponent(Core& core) : core{core} { core.register_comp_init(this); } diff --git a/src/core/configs.cpp b/src/core/configs.cpp index 5aa930be6..ed01e2282 100644 --- a/src/core/configs.cpp +++ b/src/core/configs.cpp @@ -1,6 +1,7 @@ #include "session/core/configs.hpp" #include +#include #include #include #include @@ -33,6 +34,12 @@ Configs::Configs(Core& core) : CoreComponent{core} {} Configs::~Configs() = default; void Configs::_load() { + // Every public entry point on this class reaches the config objects through here, so this is + // the one place the threading rule has to hold: the objects are built lazily, so two threads + // arriving together would race on construction, and once built they are what `merge()` mutates + // on the loop while a reader is walking them. + assert(on_loop()); + if (_loaded) return; @@ -263,11 +270,9 @@ void Configs::_schedule_push() { } void Configs::_arm_push_timer(std::chrono::milliseconds delay) { - loop().call_later(delay, [this, alive = std::weak_ptr{_alive}] { - if (alive.expired()) - return; - _push_if_due(); - }); + // On Core's queue rather than the loop, so stopping the queue deletes the pending timer. The + // `_alive` canary the network callback below still needs is exactly what that spares us here. + jq().call_later(delay, [this] { _push_if_due(); }); } void Configs::_push_if_due() { @@ -293,6 +298,10 @@ void Configs::_push_if_due() { } void Configs::push_now() { + // Its own assert because the early return below reads push state without going through + // _load(), so this is the one path that could otherwise skip the check entirely. + assert(on_loop()); + if (_push_in_flight) return; _send_push(); diff --git a/src/core/devices.cpp b/src/core/devices.cpp index 493d80db2..b67c0f352 100644 --- a/src/core/devices.cpp +++ b/src/core/devices.cpp @@ -3,6 +3,7 @@ #include #include +#include #include #include #include @@ -456,7 +457,16 @@ device::map Devices::devices( return devs; } -std::pair Devices::device_info() { +void Devices::device_info(failable_function)> cb) { + async([this] { return _device_info(); }, std::move(cb)); +} + +std::pair Devices::device_info(await_t) { + return jq().call_get([this] { return _device_info(); }); +} + +std::pair Devices::_device_info() { + assert(on_loop()); auto devs = devices(true, true, true, self_id); if (auto it = devs.find(self_id); it != devs.end()) { // Read the state out before the move: the elements of a braced-init-list are evaluated in @@ -474,8 +484,17 @@ bool device::Info::same_user_fields(const Info& other) const { return fields(*this) == fields(other); } -void Devices::update_info(const device::Info& info) { - auto [current, is_registered] = device_info(); +void Devices::update_info(device::Info info, failable_function cb) { + async([this, info = std::move(info)] { _update_info(info); }, std::move(cb)); +} + +void Devices::update_info(const device::Info& info, await_t) { + jq().call_get([this, &info] { _update_info(info); }); +} + +void Devices::_update_info(const device::Info& info) { + assert(on_loop()); + auto [current, is_registered] = _device_info(); // Early-exit if nothing changed: no seqno bump, no push triggered. // current.seqno == 0 means no row exists yet (default-init sentinel; real rows have seqno >= @@ -1119,8 +1138,17 @@ void Devices::receive_device_group_message(std::span data) { tx.commit(); } -Devices::LinkRequestResult Devices::build_link_request() { - auto [info, is_registered] = device_info(); +void Devices::build_link_request(failable_function cb) { + async([this] { return _build_link_request(); }, std::move(cb)); +} + +Devices::LinkRequestResult Devices::build_link_request(await_t) { + return jq().call_get([this] { return _build_link_request(); }); +} + +Devices::LinkRequestResult Devices::_build_link_request() { + assert(on_loop()); + auto [info, is_registered] = _device_info(); if (is_registered) throw std::logic_error{ diff --git a/src/core/globals.cpp b/src/core/globals.cpp index f410431fb..ffb27ab36 100644 --- a/src/core/globals.cpp +++ b/src/core/globals.cpp @@ -1,6 +1,7 @@ #include #include +#include #include #include #include @@ -123,6 +124,12 @@ bool Globals::erase(std::string_view key) { } void Globals::_adopt_seed(const cleared_b32& seed, bool persist) { + // Everything below this line is cached state that Core's loop reads while polling, rewritten + // in place: the secure buffer reallocates and `_session_id_hex` is reassigned. Reached from + // init() during construction, which on_loop() excuses, and otherwise only from the two public + // account methods. + assert(on_loop()); + // Layout: [ed25519_sk(64) | x25519_sk(32)] = 96 bytes auto rw = _account_seed.resize(96); @@ -178,7 +185,16 @@ void Globals::init() { tx.commit(); } -void Globals::create_account() { +void Globals::create_account(failable_function cb) { + async([this] { _create_account(); }, std::move(cb)); +} + +void Globals::create_account(await_t) { + jq().call_get([this] { _create_account(); }); +} + +void Globals::_create_account() { + assert(on_loop()); if (_have_account) throw std::logic_error{"This account already has an identity"}; auto c = conn(); @@ -202,7 +218,16 @@ void Globals::_mark_new_account() { core.devices._mark_group_owed(); } -void Globals::restore_account(const predefined_seed& seed) { +void Globals::restore_account(predefined_seed seed, failable_function cb) { + async([this, seed = std::move(seed)] { _restore_account(seed); }, std::move(cb)); +} + +void Globals::restore_account(const predefined_seed& seed, await_t) { + jq().call_get([this, &seed] { _restore_account(seed); }); +} + +void Globals::_restore_account(const predefined_seed& seed) { + assert(on_loop()); if (_have_account) throw std::logic_error{"This account already has an identity"}; auto c = conn(); diff --git a/tests/test_core_devices.cpp b/tests/test_core_devices.cpp index 4642b9249..00c7f5b73 100644 --- a/tests/test_core_devices.cpp +++ b/tests/test_core_devices.cpp @@ -56,7 +56,7 @@ TEST_CASE("Devices - initial state", "[core][devices]") { auto c = restored_core(); SECTION("device_info defaults") { - auto [info, is_registered] = c->devices.device_info(); + auto [info, is_registered] = c->devices.device_info(await); // seqno == 0 is the sentinel meaning no row exists yet CHECK(info.seqno == 0); CHECK_FALSE(is_registered); @@ -82,9 +82,9 @@ TEST_CASE("Devices - update_info and same_user_fields", "[core][devices]") { info.description = "test phone"; info.version = {1, 2, 3}; - c->devices.update_info(info); + c->devices.update_info(info, await); - auto [got, is_registered] = c->devices.device_info(); + auto [got, is_registered] = c->devices.device_info(await); CHECK(got.seqno == 1); CHECK(got.type == device::Type::Session_iOS); CHECK(got.description == "test phone"); @@ -99,64 +99,64 @@ TEST_CASE("Devices - update_info and same_user_fields", "[core][devices]") { info.description = "desktop"; info.version = {0, 1, 0}; - c->devices.update_info(info); - CHECK(c->devices.device_info().first.seqno == 1); + c->devices.update_info(info, await); + CHECK(c->devices.device_info(await).first.seqno == 1); - c->devices.update_info(info); // identical — should not bump - CHECK(c->devices.device_info().first.seqno == 1); + c->devices.update_info(info, await); // identical — should not bump + CHECK(c->devices.device_info(await).first.seqno == 1); } SECTION("changed description bumps seqno") { device::Info info{}; info.description = "first"; - c->devices.update_info(info); - CHECK(c->devices.device_info().first.seqno == 1); + c->devices.update_info(info, await); + CHECK(c->devices.device_info(await).first.seqno == 1); info.description = "second"; - c->devices.update_info(info); - CHECK(c->devices.device_info().first.seqno == 2); + c->devices.update_info(info, await); + CHECK(c->devices.device_info(await).first.seqno == 2); } SECTION("changed type bumps seqno") { device::Info info{}; info.type = device::Type::Session_Android; - c->devices.update_info(info); - CHECK(c->devices.device_info().first.seqno == 1); + c->devices.update_info(info, await); + CHECK(c->devices.device_info(await).first.seqno == 1); info.type = device::Type::Session_Desktop; - c->devices.update_info(info); - CHECK(c->devices.device_info().first.seqno == 2); + c->devices.update_info(info, await); + CHECK(c->devices.device_info(await).first.seqno == 2); } SECTION("changed version bumps seqno") { device::Info info{}; info.version = {1, 0, 0}; - c->devices.update_info(info); - CHECK(c->devices.device_info().first.seqno == 1); + c->devices.update_info(info, await); + CHECK(c->devices.device_info(await).first.seqno == 1); info.version = {2, 0, 0}; - c->devices.update_info(info); - CHECK(c->devices.device_info().first.seqno == 2); + c->devices.update_info(info, await); + CHECK(c->devices.device_info(await).first.seqno == 2); } SECTION("extra fields round-trip and participate in comparison") { device::Info info{}; info.extra["custom_key"] = std::string{"hello"}; - c->devices.update_info(info); + c->devices.update_info(info, await); - auto [got, _] = c->devices.device_info(); + auto [got, _] = c->devices.device_info(await); CHECK(got.seqno == 1); REQUIRE(got.extra.count("custom_key")); CHECK(std::get(got.extra.at("custom_key")) == "hello"); // Same extra — no bump - c->devices.update_info(info); - CHECK(c->devices.device_info().first.seqno == 1); + c->devices.update_info(info, await); + CHECK(c->devices.device_info(await).first.seqno == 1); // Changed extra — bump info.extra["custom_key"] = std::string{"world"}; - c->devices.update_info(info); - CHECK(c->devices.device_info().first.seqno == 2); + c->devices.update_info(info, await); + CHECK(c->devices.device_info(await).first.seqno == 2); } SECTION("same_user_fields ignores state/seqno/pk_*") { @@ -181,7 +181,7 @@ TEST_CASE("Devices - update_info and same_user_fields", "[core][devices]") { SECTION("update_info device appears in devices(include_unregistered=true)") { device::Info info{}; info.description = "my device"; - c->devices.update_info(info); + c->devices.update_info(info, await); auto devs = c->devices.devices(false, false, true); CHECK(devs.size() == 1); @@ -432,7 +432,7 @@ TEST_CASE("Devices - build_link_request", "[core][devices]") { auto c = restored_core(); SECTION("returns non-empty message and 21-entry SAS") { - auto result = c->devices.build_link_request(); + auto result = c->devices.build_link_request(await); CHECK_FALSE(result.message.empty()); CHECK(result.sas.size() == 21); for (const auto& s : result.sas) @@ -440,8 +440,8 @@ TEST_CASE("Devices - build_link_request", "[core][devices]") { } SECTION("consecutive calls produce different messages") { - auto r1 = c->devices.build_link_request(); - auto r2 = c->devices.build_link_request(); + auto r1 = c->devices.build_link_request(await); + auto r2 = c->devices.build_link_request(await); CHECK(r1.message != r2.message); } } @@ -504,10 +504,10 @@ TEST_CASE("Devices - establishing the group", "[core][devices]") { SECTION("a generated account establishes a group with itself") { TempCore c; - auto [info, registered] = c->devices.device_info(); + auto [info, registered] = c->devices.device_info(await); CHECK(registered); CHECK(info.state == device::State::Registered); - CHECK(info.id == c->devices.device_info().first.id); + CHECK(info.id == c->devices.device_info(await).first.id); // Exactly one device, and it is us. auto devs = c->devices.devices(true, true, true); @@ -528,7 +528,7 @@ TEST_CASE("Devices - establishing the group", "[core][devices]") { SECTION("a restored account does not") { auto c = restored_core(); - auto [info, registered] = c->devices.device_info(); + auto [info, registered] = c->devices.device_info(await); CHECK_FALSE(registered); CHECK(c->devices.devices(true, true, true).empty()); CHECK_FALSE(c->devices.needs_push().device_group); @@ -541,7 +541,7 @@ TEST_CASE("Devices - establishing the group", "[core][devices]") { fmt::format("{}.db", random::unique_id("test_estab", 7)); { Core c{path}; - auto [info, registered] = c.devices.device_info(); + auto [info, registered] = c.devices.device_info(await); REQUIRE(registered); first_id = info.id; first_seqno = info.seqno; @@ -550,7 +550,7 @@ TEST_CASE("Devices - establishing the group", "[core][devices]") { // Reopened: the flag was cleared the first time, so this must not re-register or // re-mint anything -- a second establish would bump the seqno and mint a second key. Core c{path}; - auto [info, registered] = c.devices.device_info(); + auto [info, registered] = c.devices.device_info(await); CHECK(registered); CHECK(info.id == *first_id); CHECK(info.seqno == first_seqno); @@ -577,7 +577,7 @@ TEST_CASE("Devices - a removal cannot be undone by a message", "[core][devices]" other.pk_x25519 = k.x25519_pub; other.pk_mlkem768 = k.mlkem768_pub; - auto [self, registered] = c->devices.device_info(); + auto [self, registered] = c->devices.device_info(await); REQUIRE(registered); auto deliver = [&](const device::map& m) { @@ -633,7 +633,7 @@ TEST_CASE("Devices - a tombstone for an unknown device is kept", "[core][devices // below is a usable recipient. auto k = c->devices.rotate_device_keys(); - auto [self, registered] = c->devices.device_info(); + auto [self, registered] = c->devices.device_info(await); REQUIRE(registered); auto deliver = [&](const device::map& m) { @@ -687,7 +687,7 @@ TEST_CASE("Devices - a single-recipient group is readable", "[core][devices]") { // The common case: an account with one device, which is what establishing a group produces. // With one recipient every other slot in the message is padding, so nothing else can stand in // for a real entry that was overwritten. - auto [self, registered] = c->devices.device_info(); + auto [self, registered] = c->devices.device_info(await); REQUIRE(registered); auto enc = TestHelper::encrypt_device_data(c->devices, device::map{{self.id, self}}); diff --git a/tests/test_core_globals.cpp b/tests/test_core_globals.cpp index 38f750f17..21167f032 100644 --- a/tests/test_core_globals.cpp +++ b/tests/test_core_globals.cpp @@ -90,12 +90,12 @@ TEST_CASE("Globals: defer_account leaves the account unresolved", "[core][global { Core core{path, defer_account{}}; REQUIRE_FALSE(core.globals.have_account()); - core.globals.create_account(); + core.globals.create_account(await); CHECK(core.globals.have_account()); id = core.globals.session_id_hex(); CHECK(id.starts_with("05")); // Adopting a second identity would orphan everything stored against the first. - CHECK_THROWS_AS(core.globals.create_account(), std::logic_error); + CHECK_THROWS_AS(core.globals.create_account(await), std::logic_error); } // Reopening finds the stored seed, so defer_account is a no-op on an existing account. Core core{path, defer_account{}}; @@ -107,7 +107,7 @@ TEST_CASE("Globals: defer_account leaves the account unresolved", "[core][global std::string restored; { Core core{path, defer_account{}}; - core.globals.restore_account(predefined_seed{seed}); + core.globals.restore_account(predefined_seed{seed}, await); CHECK(core.globals.have_account()); restored = core.globals.session_id_hex(); } diff --git a/tests/test_helper.hpp b/tests/test_helper.hpp index 284095c33..9fa711ef4 100644 --- a/tests/test_helper.hpp +++ b/tests/test_helper.hpp @@ -321,6 +321,17 @@ class TestHelper { public: static void poll(core::Core& core) { core._poll(); } + /// Runs everything already queued on Core's job queue and waits for it. + /// + /// Needed wherever a test drives a network response by hand: in production those arrive on the + /// Network's own loop and Core marshals them onto its queue, so the work is finished a moment + /// after the callback returns rather than during it. A test calling the handler directly has + /// to wait for that in the same way, and the queue is FIFO, so a round-trip through it is + /// enough -- everything posted earlier has run by the time this returns. + static void drain(core::Core& core) { + core._jq.call_get([] {}); + } + /// Puts a swarm straight into the pool's cache. get_swarm consults it first and answers from /// it without touching the network, which is what lets swarm-level behaviour be tested at all: /// a test pool has no seed nodes, so nothing would ever resolve otherwise. diff --git a/tests/test_poll.cpp b/tests/test_poll.cpp index 14e125c60..5e7035f5b 100644 --- a/tests/test_poll.cpp +++ b/tests/test_poll.cpp @@ -113,10 +113,11 @@ TEST_CASE("Core automatic polling", "[core][poll]") { std::ranges::copy(std::as_bytes(seed_acc.seed()), seed_bytes.begin()); } TempCore linker{core::predefined_seed{std::span{seed_bytes}}}; - auto outer_msg = linker->devices.build_link_request().message; + auto outer_msg = linker->devices.build_link_request(await).message; sent.callback( true, false, 200, {}, make_response(*sent.request.body, 21, outer_msg, "hash1").dump()); + TestHelper::drain(*core); // Verify last_hash was stored under this specific node's pubkey. CHECK(TestHelper::namespace_last_hash(*core, 21, mock_net->current_node.remote_pubkey) == @@ -159,6 +160,7 @@ TEST_CASE( {}, make_response(*mock_net->sent_requests[0].request.body, 21, {std::byte{0x01}}, "xyz") .dump()); + TestHelper::drain(*c); CHECK(TestHelper::namespace_last_hash(*c, 21, node_a.remote_pubkey) == "xyz"); CHECK_FALSE(TestHelper::namespace_last_hash(*c, 21, node_b.remote_pubkey).has_value()); @@ -189,6 +191,7 @@ TEST_CASE( {}, make_response(*mock_net->sent_requests[0].request.body, 21, {std::byte{0x02}}, "zyx") .dump()); + TestHelper::drain(*c); CHECK(TestHelper::namespace_last_hash(*c, 21, node_b.remote_pubkey) == "zyx"); // A's hash is untouched. CHECK(TestHelper::namespace_last_hash(*c, 21, node_a.remote_pubkey) == "xyz"); @@ -235,7 +238,7 @@ TEST_CASE("Poll: the sync cursor advances only after the batch is handled", "[co std::ranges::copy(std::as_bytes(seed_acc.seed()), seed_bytes.begin()); } TempCore linker{core::predefined_seed{std::span{seed_bytes}}}; - auto outer_msg = linker->devices.build_link_request().message; + auto outer_msg = linker->devices.build_link_request(await).message; TestHelper::poll(*core); REQUIRE(mock_net->sent_requests.size() == 1); @@ -245,6 +248,7 @@ TEST_CASE("Poll: the sync cursor advances only after the batch is handled", "[co 200, {}, make_response(*mock_net->sent_requests[0].request.body, 21, outer_msg, "hash1").dump()); + TestHelper::drain(*core); REQUIRE(called); CHECK(!hash_during_callback); @@ -287,7 +291,7 @@ TEST_CASE("Poll: a truncated namespace is continued before it is reported final" std::ranges::copy(std::as_bytes(seed_acc.seed()), seed_bytes.begin()); } TempCore linker{core::predefined_seed{std::span{seed_bytes}}}; - auto outer_msg = linker->devices.build_link_request().message; + auto outer_msg = linker->devices.build_link_request(await).message; TestHelper::poll(*core); REQUIRE(mock_net->sent_requests.size() == 1); @@ -296,10 +300,12 @@ TEST_CASE("Poll: a truncated namespace is continued before it is reported final" auto resp = make_response(first, 21, outer_msg, "hash1"); set_more(resp, first, 21); - // Copied out before invoking: the continuation is sent from inside this call, which appends to - // `sent_requests` and can reallocate the vector the callback itself lives in. + // Copied out before invoking: handling the response appends to `sent_requests` -- the batch is + // not final, so a continuation goes out -- which can reallocate the vector the callback itself + // lives in. auto reply = mock_net->sent_requests[0].callback; reply(true, false, 200, {}, resp.dump()); + TestHelper::drain(*core); // The request is stored, but the batch was not final, so nothing has been reported yet. CHECK(calls == 0); @@ -313,6 +319,7 @@ TEST_CASE("Poll: a truncated namespace is continued before it is reported final" // final. auto reply2 = mock_net->sent_requests[1].callback; reply2(true, false, 200, {}, make_empty_response(second).dump()); + TestHelper::drain(*core); CHECK(calls == 1); } @@ -330,6 +337,7 @@ TEST_CASE("Poll: `more` with nothing returned does not continue", "[core][poll]" auto reply = mock_net->sent_requests[0].callback; reply(true, false, 200, {}, resp.dump()); + TestHelper::drain(*core); // There is no new hash to move the cursor to, so another round would ask the same question. CHECK(mock_net->sent_requests.size() == 1); From 7c06ad3904dda805c9e3b932cb7c236540766b1a Mon Sep 17 00:00:00 2001 From: Jason Rhinelander Date: Thu, 10 Sep 2026 16:22:32 -0300 Subject: [PATCH 2/5] Give Core and Client call/call_soon/call_later/call_get Deferring work correctly meant remembering to reach for `_jq` rather than `loop`, and the two read identically at the call site -- which is why nine Client sites and `_async` itself got it wrong. A comment on `_jq` does not fix that; removing the alternative does. `Client::loop` is gone. Everything it was used for now has a wrapper, so the mistake is no longer writable inside Client, and the compiler found every site rather than leaving it to review. Anything that genuinely wants the loop still says `core.loop()`, which stays public and now documents what it costs: a job on the loop's own queue is not discarded until `~Loop`, the last thing `~Core` does. The wrappers are private on Client, whose deferring is all internal, and public on Core, where `loop()` was already reachable and this is the safer spelling of the same thing. `call_get` returns by value rather than perfectly forwarding. A reference handed back through it has outlived the job that produced it, which is the hazard the queue exists to close, so decaying it is the point rather than a limitation. --- include/session/client.hpp | 33 +++++++++++--- include/session/core.hpp | 34 ++++++++++++++ src/client/client.cpp | 90 ++++++++++++++++++------------------- src/client/conversation.cpp | 30 ++++++------- tests/test_helper.hpp | 17 ++++++- 5 files changed, 138 insertions(+), 66 deletions(-) diff --git a/include/session/client.hpp b/include/session/client.hpp index 480a23f6f..fcb9d9a29 100644 --- a/include/session/client.hpp +++ b/include/session/client.hpp @@ -1185,7 +1185,7 @@ class Client { // throw it away. On the loop's own queue it would instead run during Core's destruction // -- the loop thread keeps draining until ~Loop, which is the *last* thing ~Core does -- // reaching a Client whose components have already gone. - _jq.call([this, produce = std::move(produce), cb = std::move(cb)]() mutable { + call([this, produce = std::move(produce), cb = std::move(cb)]() mutable { using Result = decltype(produce()); try { if constexpr (std::is_void_v) { @@ -1251,10 +1251,33 @@ class Client { /// Put anything a Core callback touches *above* this, never below. core::Core core; - /// Helper reference to Core's event loop, which is where this class does its work. - oxen::quic::Loop& loop{core.loop()}; - private: + /// Schedules work on this Client's job queue, which is where everything this class defers + /// belongs -- see `_jq`. Same shapes as `Core`'s: `call` runs inline when already on the loop + /// thread, `call_soon` always queues, `call_later` queues after a delay, and `call_get` blocks + /// until the answer is ready. + /// + /// There is deliberately no `loop` member any more. It read as the obvious way to defer work + /// and was the wrong one, since the loop's own queue outlives every Core member that these + /// jobs reach through `this`. Anything genuinely wanting the loop says `core.loop()`. + template + void call(F&& f) { + _jq.call(std::forward(f)); + } + template + void call_soon(F&& f) { + _jq.call_soon(std::forward(f)); + } + template + void call_later(std::chrono::microseconds delay, F&& f) { + _jq.call_later(delay, std::forward(f)); + } + /// By value, for the same reason as Core's: a reference returned here has escaped the loop. + template + auto call_get(F&& f) { + return _jq.call_get(std::forward(f)); + } + // Client's own queue on Core's loop, rather than the loop's shared one, so that work deferred // here is *cancelled* if the Client is destroyed with it still outstanding. Running it instead // would mean reporting a change to the subscribers of a Client that is going away, against a @@ -1271,7 +1294,7 @@ class Client { // // Declared after `core` -- the one thing that belongs below it -- because a JobQueue needs its // loop alive in order to stop, so it has to be destroyed while Core still exists. - oxen::quic::JobQueue _jq{loop}; + oxen::quic::JobQueue _jq{core.loop()}; }; } // namespace session::client diff --git a/include/session/core.hpp b/include/session/core.hpp index 67b65a615..ebff94c3c 100644 --- a/include/session/core.hpp +++ b/include/session/core.hpp @@ -606,8 +606,42 @@ class Core { /// /// `call()` runs the job inline when the caller is already on this thread, so a single-threaded /// application pays nothing for the indirection. + /// + /// Prefer the `call*` methods below to scheduling on this directly: a job left on the loop's + /// own queue is not discarded until `~Loop`, which is the last thing `~Core` does, so it can + /// still be run against components that have already been destroyed. This is the escape hatch + /// for the cases that genuinely want the loop itself. quic::Loop& loop(); + /// Schedules work on Core's job queue, which is where anything reaching into Core from another + /// thread belongs. + /// + /// `call` runs `f` inline when the caller is already the loop thread and queues it otherwise; + /// `call_soon` queues it either way; `call_later` queues it after a delay; and `call_get` + /// blocks the calling thread until `f` has run, handing back whatever it returned. + /// + /// Unlike `loop()`, work put here is *cancelled* when Core goes away, so a job still + /// outstanding is dropped rather than run against half-destroyed components. Queueing onto a + /// stopped queue throws rather than doing so silently. + template + void call(F&& f) { + _jq.call(std::forward(f)); + } + template + void call_soon(F&& f) { + _jq.call_soon(std::forward(f)); + } + template + void call_later(std::chrono::microseconds delay, F&& f) { + _jq.call_later(delay, std::forward(f)); + } + /// Returns by value, deliberately: a reference handed back here would have outlived the job + /// that produced it, which is the whole hazard this queue exists to close. + template + auto call_get(F&& f) { + return _jq.call_get(std::forward(f)); + } + /// The account database, for a layer built on top of Core that keeps its own tables alongside /// Core's — the same layer that supplies a schema_extension to create them. /// diff --git a/src/client/client.cpp b/src/client/client.cpp index 619e94468..5f32f7f84 100644 --- a/src/client/client.cpp +++ b/src/client/client.cpp @@ -440,7 +440,7 @@ void Client::_init() { // already running by the time a Client is built: the first `_touch` schedules `_flush_pending`, // which steals `_dirty` out from under the reconcile still filling it. if (core.globals.have_account()) - loop.call_get([this] { _reconcile_all(); }); + call_get([this] { _reconcile_all(); }); } // -- Change notification ---------------------------------------------------------------------- @@ -450,11 +450,11 @@ void Client::_emit(std::function invoke) { } void Client::set_dispatcher(dispatcher d) { - _jq.call([this, d = std::move(d)]() mutable { _dispatcher = std::move(d); }); + call([this, d = std::move(d)]() mutable { _dispatcher = std::move(d); }); } void Client::set_high_freq_dispatch_interval(std::chrono::milliseconds interval) { - _jq.call([this, interval] { _high_freq_dispatch_interval = interval; }); + call([this, interval] { _high_freq_dispatch_interval = interval; }); } void Client::_dispatch_out(std::function job) { @@ -592,7 +592,7 @@ void Client::_require_sendable( // alternative is to accept the send, discover inside the loop that the target is not there, and // have only a callback to say so. A caller naming a message that does not exist has made a // mistake at the call site, and that is where it should be reported. - auto found = loop.call_get([this, id, target = *msg.reply_to] { + auto found = call_get([this, id, target = *msg.reply_to] { auto c = core.database().conn(); return c.prepared_maybe_get( R"( @@ -661,7 +661,7 @@ void Client::retry_send( } bool Client::retry_send(int64_t message_id, Conversation::upload_progress on_upload, await_t) { - return loop.call_get([this, message_id, on_upload = std::move(on_upload)] { + return call_get([this, message_id, on_upload = std::move(on_upload)] { return _retry_send(message_id, on_upload); }); } @@ -676,7 +676,7 @@ void Client::message_debug( } std::optional Client::message_debug(int64_t message_id, await_t) { - return loop.call_get([this, message_id] { return _message_debug(message_id); }); + return call_get([this, message_id] { return _message_debug(message_id); }); } void Client::delete_message(int64_t message_id, failable_function cb) { @@ -685,7 +685,7 @@ void Client::delete_message(int64_t message_id, failable_function cb } bool Client::delete_message(int64_t message_id, await_t) { - return loop.call_get( + return call_get( [this, message_id] { return _delete_message(message_id, Deletion::here); }); } @@ -718,7 +718,7 @@ void Client::_sweep_cache() { // `call_get`, not `call`: the destructor joins this thread to know the sweep is over, and a // thread that had only posted the job would finish while the job was still queued. - loop.call_get([this, &attachments, &pictures] { + call_get([this, &attachments, &pictures] { try { _reconcile_cache(std::move(attachments), std::move(pictures)); } catch (const std::exception& e) { @@ -802,7 +802,7 @@ void Client::profile_picture( const ConversationId& id, std::function)> on_progress, failable_function>)> cb) { - _jq.call([this, id, on_progress = std::move(on_progress), cb = std::move(cb)]() mutable { + call([this, id, on_progress = std::move(on_progress), cb = std::move(cb)]() mutable { try { _profile_picture(id, std::move(on_progress), std::move(cb)); } catch (const std::exception& e) { @@ -898,13 +898,13 @@ void Client::set_attachment_cache_limit( _async([this, bytes] { set_limit(core.globals, CACHE_LIMIT_KEY, bytes); }, std::move(cb)); } void Client::set_attachment_cache_limit(std::optional bytes, await_t) { - loop.call_get([this, bytes] { set_limit(core.globals, CACHE_LIMIT_KEY, bytes); }); + call_get([this, bytes] { set_limit(core.globals, CACHE_LIMIT_KEY, bytes); }); } void Client::attachment_cache_limit(failable_function)> cb) { _async([this] { return core.globals.get_integer(CACHE_LIMIT_KEY); }, std::move(cb)); } std::optional Client::attachment_cache_limit(await_t) { - return loop.call_get([this] { return core.globals.get_integer(CACHE_LIMIT_KEY); }); + return call_get([this] { return core.globals.get_integer(CACHE_LIMIT_KEY); }); } void Client::set_auto_download_max_size( @@ -912,13 +912,13 @@ void Client::set_auto_download_max_size( _async([this, bytes] { set_limit(core.globals, AUTO_DL_MAX_KEY, bytes); }, std::move(cb)); } void Client::set_auto_download_max_size(std::optional bytes, await_t) { - loop.call_get([this, bytes] { set_limit(core.globals, AUTO_DL_MAX_KEY, bytes); }); + call_get([this, bytes] { set_limit(core.globals, AUTO_DL_MAX_KEY, bytes); }); } void Client::auto_download_max_size(failable_function)> cb) { _async([this] { return core.globals.get_integer(AUTO_DL_MAX_KEY); }, std::move(cb)); } std::optional Client::auto_download_max_size(await_t) { - return loop.call_get([this] { return core.globals.get_integer(AUTO_DL_MAX_KEY); }); + return call_get([this] { return core.globals.get_integer(AUTO_DL_MAX_KEY); }); } void Client::display_name(failable_function cb) { @@ -927,7 +927,7 @@ void Client::display_name(failable_function cb) { } std::string Client::display_name(await_t) { - return loop.call_get( + return call_get( [this] { return std::string{core.configs.user_profile().get_name().value_or("")}; }); } @@ -937,7 +937,7 @@ void Client::set_display_name(std::string_view name, failable_function c } void Client::set_display_name(std::string_view name, await_t) { - loop.call_get([this, name] { core.configs.user_profile().set_name(name); }); + call_get([this, name] { core.configs.user_profile().set_name(name); }); } void Client::notify_media_saved(failable_function cb) { @@ -945,7 +945,7 @@ void Client::notify_media_saved(failable_function cb) { } bool Client::notify_media_saved(await_t) { - return loop.call_get([this] { return core.configs.user_profile().get_notify_media_saved(); }); + return call_get([this] { return core.configs.user_profile().get_notify_media_saved(); }); } void Client::set_notify_media_saved(bool notify, failable_function cb) { @@ -954,7 +954,7 @@ void Client::set_notify_media_saved(bool notify, failable_function cb) { } void Client::set_notify_media_saved(bool notify, await_t) { - loop.call_get([this, notify] { core.configs.user_profile().set_notify_media_saved(notify); }); + call_get([this, notify] { core.configs.user_profile().set_notify_media_saved(notify); }); } void Client::delete_message_everywhere(int64_t message_id, failable_function cb) { @@ -962,7 +962,7 @@ void Client::delete_message_everywhere(int64_t message_id, failable_function on_progress, failable_function)> cb) { - _jq.call([this, message_id, index, on_progress = std::move(on_progress), cb]() mutable { + call([this, message_id, index, on_progress = std::move(on_progress), cb]() mutable { try { _attachment_data(message_id, index, std::move(on_progress), cb); } catch (const std::exception& e) { @@ -1069,7 +1069,7 @@ void Client::_fetch_cached( // Onto the loop before touching the registry -- this arrives on the network thread, and // `_in_flight` is ours. [this, name](int64_t done, int64_t total, std::optional r) { - _jq.call([this, name, done, total, r] { + call([this, name, done, total, r] { auto found = _in_flight.find(name); if (found == _in_flight.end()) return; @@ -1080,7 +1080,7 @@ void Client::_fetch_cached( }); }, [this, name, store = std::move(store)](std::optional error) { - _jq.call([this, name, store, error = std::move(error)]() mutable { + call([this, name, store, error = std::move(error)]() mutable { auto found = _in_flight.find(name); if (found == _in_flight.end()) return; @@ -1109,7 +1109,7 @@ void Client::set_gallery(int64_t message_id, bool gallery, failable_function cb) { @@ -1117,7 +1117,7 @@ void Client::purge_deleted_message(int64_t message_id, failable_function Client::conversations(await_t) { - return loop.call_get([this] { return _conversations(); }); + return call_get([this] { return _conversations(); }); } std::vector Client::message_requests(await_t) { - return loop.call_get([this] { return _message_requests(); }); + return call_get([this] { return _message_requests(); }); } std::optional Client::conversation(const ConversationId& id, await_t) { - return loop.call_get([this, id] { return _conversation(id); }); + return call_get([this, id] { return _conversation(id); }); } std::optional Client::message(int64_t id, await_t) { - return loop.call_get([this, id] { return _message(id); }); + return call_get([this, id] { return _message(id); }); } int64_t Client::send_message(const ConversationId& id, OutgoingMessage msg, await_t) { @@ -1192,7 +1192,7 @@ int64_t Client::send_message( Conversation::upload_progress on_upload, await_t) { _require_sendable("send_message", id, msg); - return loop.call_get([&] { return _send_message(id, msg, std::move(on_upload)); }); + return call_get([&] { return _send_message(id, msg, std::move(on_upload)); }); } void Client::conversation( @@ -1220,7 +1220,7 @@ void Client::dm( std::optional Client::dm(const ConversationId& id, await_t) { _require_dm("dm", id); - return loop.call_get([this, id] { return as_dm(_conversation(id)); }); + return call_get([this, id] { return as_dm(_conversation(id)); }); } void Client::open_dm( @@ -1233,7 +1233,7 @@ void Client::open_dm( DM Client::open_dm(const ConversationId& id, await_t) { _require_dm("open_dm", id); - return loop.call_get([this, id] { + return call_get([this, id] { return *as_dm(std::optional{_create_conversation(id)}); }); } @@ -1262,14 +1262,14 @@ void Client::save_attachment( // Not _async: what that reports is the *start* of the transfer, and the answer a caller wants // is whether the file arrived, which is minutes away. So the callback is carried down to the // download's own completion, and only the failures that happen before it starts come back here. - _jq.call([this, - message_id, - index, - dest = std::move(dest), - on_progress = std::move(on_progress), - cb, - notify_sender, - replace]() mutable { + call([this, + message_id, + index, + dest = std::move(dest), + on_progress = std::move(on_progress), + cb, + notify_sender, + replace]() mutable { try { _save_attachment( message_id, @@ -2411,10 +2411,10 @@ void Client::_prefetch_picture(sqlite::Connection& c, int64_t account, const std auto& [sid, key] = *row; - _jq.call_soon([this, - id = ConversationId::dm(sid), - url, - key = std::vector{key.begin(), key.end()}]() mutable { + call_soon([this, + id = ConversationId::dm(sid), + url, + key = std::vector{key.begin(), key.end()}]() mutable { _fetch_picture(id, std::move(url), std::move(key)); }); } catch (const std::exception& e) { @@ -3796,7 +3796,7 @@ void Client::_upload_next( // Core's loop's alone. Nobody is waiting on a callback here -- this is Client's // own continuation -- so a failure has to be turned into the message failing, which // is what the application is watching. - _jq.call([this, client_id, index, on_upload, plaintext_size, result = std::move(result)] { + call([this, client_id, index, on_upload, plaintext_size, result = std::move(result)] { try { if (auto* err = std::get_if(&result)) { log::warning( @@ -4408,7 +4408,7 @@ void Client::_save_attachment( // we sent would claim they have a file they may never have opened. A note to self // is exempt: there the recipient is us, so saving it really is the recipient // saving it. - _jq.call([this, message_id, index, notify_sender] { + call([this, message_id, index, notify_sender] { if (_saved_by_recipient(message_id)) _record_saved(message_id, index, clock_now_ms()); diff --git a/src/client/conversation.cpp b/src/client/conversation.cpp index 01713242e..5936ccd47 100644 --- a/src/client/conversation.cpp +++ b/src/client/conversation.cpp @@ -62,7 +62,7 @@ std::vector Conversation::messages( std::vector Conversation::messages( int limit, std::optional before, bool include_deleted, await_t) const { _client->_require_page("messages", limit); - return _client->loop.call_get([this, limit, before, include_deleted] { + return _client->call_get([this, limit, before, include_deleted] { return _client->_messages(id, limit, before, include_deleted); }); } @@ -73,7 +73,7 @@ void Conversation::purge_deleted(failable_function cb) { } size_t Conversation::purge_deleted(await_t) { _client->_require_dm("purge_deleted", id); - return _client->loop.call_get([this] { return _client->_purge_deleted(id); }); + return _client->call_get([this] { return _client->_purge_deleted(id); }); } // -- Read state --------------------------------------------------------------------------------- @@ -88,7 +88,7 @@ void Conversation::mark_read(await_t) { mark_read(std::nullopt, await); } void Conversation::mark_read(std::optional up_to, await_t) { - _client->loop.call_get([this, up_to] { _client->_mark_read(id, up_to); }); + _client->call_get([this, up_to] { _client->_mark_read(id, up_to); }); } void Conversation::set_marked_unread(bool unread, failable_function cb) { @@ -96,7 +96,7 @@ void Conversation::set_marked_unread(bool unread, failable_function cb) [c = _client, id = id, unread] { c->_set_marked_unread(id, unread); }, std::move(cb)); } void Conversation::set_marked_unread(bool unread, await_t) { - _client->loop.call_get([this, unread] { _client->_set_marked_unread(id, unread); }); + _client->call_get([this, unread] { _client->_set_marked_unread(id, unread); }); } // -- Settings ----------------------------------------------------------------------------------- @@ -106,7 +106,7 @@ void Conversation::set_priority(int priority, failable_function cb) { [c = _client, id = id, priority] { c->_set_priority(id, priority); }, std::move(cb)); } void Conversation::set_priority(int priority, await_t) { - _client->loop.call_get([this, priority] { _client->_set_priority(id, priority); }); + _client->call_get([this, priority] { _client->_set_priority(id, priority); }); } void Conversation::set_notifications(config::notify_mode mode, failable_function cb) { @@ -114,7 +114,7 @@ void Conversation::set_notifications(config::notify_mode mode, failable_function [c = _client, id = id, mode] { c->_set_notifications(id, mode); }, std::move(cb)); } void Conversation::set_notifications(config::notify_mode mode, await_t) { - _client->loop.call_get([this, mode] { _client->_set_notifications(id, mode); }); + _client->call_get([this, mode] { _client->_set_notifications(id, mode); }); } void Conversation::set_mute_until(std::chrono::sys_seconds until, failable_function cb) { @@ -122,7 +122,7 @@ void Conversation::set_mute_until(std::chrono::sys_seconds until, failable_funct [c = _client, id = id, until] { c->_set_mute_until(id, until); }, std::move(cb)); } void Conversation::set_mute_until(std::chrono::sys_seconds until, await_t) { - _client->loop.call_get([this, until] { _client->_set_mute_until(id, until); }); + _client->call_get([this, until] { _client->_set_mute_until(id, until); }); } void Conversation::set_expiry( @@ -132,7 +132,7 @@ void Conversation::set_expiry( std::move(cb)); } void Conversation::set_expiry(config::expiration_mode mode, std::chrono::seconds timer, await_t) { - _client->loop.call_get([this, mode, timer] { _client->_set_expiry(id, mode, timer); }); + _client->call_get([this, mode, timer] { _client->_set_expiry(id, mode, timer); }); } void Conversation::set_auto_download(AutoDownload mode, failable_function cb) { @@ -140,7 +140,7 @@ void Conversation::set_auto_download(AutoDownload mode, failable_function_set_auto_download(id, mode); }, std::move(cb)); } void Conversation::set_auto_download(AutoDownload mode, await_t) { - _client->loop.call_get([this, mode] { _client->_set_auto_download(id, mode); }); + _client->call_get([this, mode] { _client->_set_auto_download(id, mode); }); } // -- Sending ------------------------------------------------------------------------------------ @@ -165,7 +165,7 @@ int64_t Conversation::send_message(OutgoingMessage msg, await_t) { } int64_t Conversation::send_message(OutgoingMessage msg, upload_progress on_upload, await_t) { _client->_require_sendable("send_message", id, msg); - return _client->loop.call_get( + return _client->call_get( [&] { return _client->_send_message(id, msg, std::move(on_upload)); }); } @@ -177,7 +177,7 @@ void Conversation::clear_messages(failable_function cb) { } void Conversation::clear_messages(await_t) { _client->_require_dm("clear_messages", id); - _client->loop.call_get([this] { _client->_clear_messages(id); }); + _client->call_get([this] { _client->_clear_messages(id); }); } void Conversation::delete_conversation(failable_function cb) { @@ -194,7 +194,7 @@ void Conversation::delete_conversation(await_t) { } void Conversation::delete_conversation(bool keep_messages, await_t) { _client->_require_dm("delete_conversation", id); - _client->loop.call_get( + _client->call_get( [this, keep_messages] { _client->_delete_conversation(id, keep_messages); }); } @@ -207,7 +207,7 @@ void DM::set_blocked(bool blocked, failable_function cb) { } void DM::set_blocked(bool blocked, await_t) { _client->_require_contact("set_blocked", id); - _client->loop.call_get([this, blocked] { _client->_set_blocked(id, blocked); }); + _client->call_get([this, blocked] { _client->_set_blocked(id, blocked); }); } void DM::set_nickname(std::string_view nickname, failable_function cb) { @@ -220,7 +220,7 @@ void DM::set_nickname(std::string_view nickname, failable_function cb) { } void DM::set_nickname(std::string_view nickname, await_t) { _client->_require_contact("set_nickname", id); - _client->loop.call_get([this, nickname] { _client->_set_nickname(id, nickname); }); + _client->call_get([this, nickname] { _client->_set_nickname(id, nickname); }); } void DM::delete_contact(failable_function cb) { @@ -229,7 +229,7 @@ void DM::delete_contact(failable_function cb) { } void DM::delete_contact(await_t) { _client->_require_contact("delete_contact", id); - _client->loop.call_get([this] { _client->_delete_contact(id); }); + _client->call_get([this] { _client->_delete_contact(id); }); } } // namespace session::client diff --git a/tests/test_helper.hpp b/tests/test_helper.hpp index 9fa711ef4..b6c7d5fef 100644 --- a/tests/test_helper.hpp +++ b/tests/test_helper.hpp @@ -321,6 +321,21 @@ class TestHelper { public: static void poll(core::Core& core) { core._poll(); } + /// Runs `f` on Core's loop and hands back what it returned. + /// + /// Core's components are the loop's, not the caller's, and a test is on its own thread like + /// any other application. Wrap the body of anything reaching into `configs` -- or any other + /// component state -- in one of these; there is no need for one per call, since everything + /// inside runs on the loop for as long as `f` does. + /// + /// This is a test's version of what `Client` does for every one of its own methods. There is + /// deliberately no application-facing equivalent on `Configs`: nothing outside libsession + /// reaches its accessors, and `Client` wraps the parts an application actually wants. + template + static auto on_loop(core::Core& core, F&& f) { + return core.call_get(std::forward(f)); + } + /// Runs everything already queued on Core's job queue and waits for it. /// /// Needed wherever a test drives a network response by hand: in production those arrive on the @@ -329,7 +344,7 @@ class TestHelper { /// to wait for that in the same way, and the queue is FIFO, so a round-trip through it is /// enough -- everything posted earlier has run by the time this returns. static void drain(core::Core& core) { - core._jq.call_get([] {}); + on_loop(core, [] {}); } /// Puts a swarm straight into the pool's cache. get_swarm consults it first and answers from From 1e7423143f5da37e6adf99fa0dcc35b7a84c1a7c Mon Sep 17 00:00:00 2001 From: Jason Rhinelander Date: Thu, 10 Sep 2026 16:53:00 -0300 Subject: [PATCH 3/5] Run the Configs tests on the loop, like everything else that touches them `test_core_configs.cpp` reached into the configs from the test thread, which is what the new assertion is about: a test is an application like any other, and the configs belong to Core's loop. `TestHelper::on_loop` wraps a block rather than a call, since a test case does several config operations in a row and they all want the same excursion. Three places could not simply be wrapped whole, and say so where they are: `reopen()` destroys the Core and so the very loop a wrapper would be running on, and Catch2's GENERATE and SECTION have to stay at test scope because the case is re-run for each. `test_client/`'s shared `*_from_another_device` helpers get the same treatment. The remaining direct reads in `test_client/` are not converted yet: several return a `string_view` into the config, which would outlive the excursion, so they need looking at one at a time rather than wrapping. --- tests/test_client/config_helpers.hpp | 62 +-- tests/test_core_configs.cpp | 670 +++++++++++++++------------ 2 files changed, 408 insertions(+), 324 deletions(-) diff --git a/tests/test_client/config_helpers.hpp b/tests/test_client/config_helpers.hpp index 5ca98d491..a259de5a1 100644 --- a/tests/test_client/config_helpers.hpp +++ b/tests/test_client/config_helpers.hpp @@ -14,15 +14,19 @@ inline std::vector> profile_from_another_device( // built from our own dump once ours has gone out. Starting it from nothing would make a rival // at the same seqno, which is a different scenario entirely -- and one that resolves by merging // the two sets of changes rather than by taking theirs. - auto& ours = c.core.configs.user_profile(); - auto [seqno, messages, obsolete] = ours.push(); - ours.confirm_pushed(seqno, {"ourprofile"}); - - auto seed = c.core.globals.account_seed(); - config::UserProfile theirs{seed.ed25519_secret(), ours.make_dump()}; - change(theirs); - auto [their_seqno, their_messages, their_obsolete] = theirs.push(); - return their_messages; + // On Core's loop, like every other reach into the configs: they are the loop's, and a test is + // on its own thread like any other application. + return TestHelper::on_loop(c.core, [&] { + auto& ours = c.core.configs.user_profile(); + auto [seqno, messages, obsolete] = ours.push(); + ours.confirm_pushed(seqno, {"ourprofile"}); + + auto seed = c.core.globals.account_seed(); + config::UserProfile theirs{seed.ed25519_secret(), ours.make_dump()}; + change(theirs); + auto [their_seqno, their_messages, their_obsolete] = theirs.push(); + return their_messages; + }); } /// Feeds them in as a poll would. A SwarmMessage points at its data rather than owning it, so @@ -67,15 +71,17 @@ inline std::vector> contacts_from_another_device( /// the union of the two, which is right but is never what a test about *removal* wants. inline std::vector> contacts_update_from_another_device( Client& c, const std::function& change) { - auto& ours = c.core.configs.contacts(); - auto [seqno, messages, obsolete] = ours.push(); - ours.confirm_pushed(seqno, {"ourcontacts"}); - - auto seed = c.core.globals.account_seed(); - config::Contacts theirs{seed.ed25519_secret(), ours.make_dump()}; - change(theirs); - auto [their_seqno, their_messages, their_obsolete] = theirs.push(); - return their_messages; + return TestHelper::on_loop(c.core, [&] { + auto& ours = c.core.configs.contacts(); + auto [seqno, messages, obsolete] = ours.push(); + ours.confirm_pushed(seqno, {"ourcontacts"}); + + auto seed = c.core.globals.account_seed(); + config::Contacts theirs{seed.ed25519_secret(), ours.make_dump()}; + change(theirs); + auto [their_seqno, their_messages, their_obsolete] = theirs.push(); + return their_messages; + }); } inline ConversationId dm_from_hex(std::string_view hex) { @@ -106,15 +112,17 @@ inline void insert_message( /// reason as `contacts_update_from_another_device`. inline std::vector> volatile_from_another_device( Client& c, const std::function& change) { - auto& ours = c.core.configs.convo_info_volatile(); - auto [seqno, messages, obsolete] = ours.push(); - ours.confirm_pushed(seqno, {"ourvolatile"}); - - auto seed = c.core.globals.account_seed(); - config::ConvoInfoVolatile theirs{seed.ed25519_secret(), ours.make_dump()}; - change(theirs); - auto [their_seqno, their_messages, their_obsolete] = theirs.push(); - return their_messages; + return TestHelper::on_loop(c.core, [&] { + auto& ours = c.core.configs.convo_info_volatile(); + auto [seqno, messages, obsolete] = ours.push(); + ours.confirm_pushed(seqno, {"ourvolatile"}); + + auto seed = c.core.globals.account_seed(); + config::ConvoInfoVolatile theirs{seed.ed25519_secret(), ours.make_dump()}; + change(theirs); + auto [their_seqno, their_messages, their_obsolete] = theirs.push(); + return their_messages; + }); } inline void merge_volatile(Client& c, const std::vector>& messages) { diff --git a/tests/test_core_configs.cpp b/tests/test_core_configs.cpp index acf35d954..61ac2ecfb 100644 --- a/tests/test_core_configs.cpp +++ b/tests/test_core_configs.cpp @@ -123,19 +123,22 @@ struct PushableCore { TEST_CASE("Configs: a fresh account starts with its defaults", "[core][configs]") { TempCore c{}; - // Not blank: creating an account writes the defaults it should start life with, and note to - // self starting hidden is one of them. It is owed to the swarm precisely because it is shared - // -- the account's other devices have to be told, or they would each invent their own answer. - CHECK(c->configs.user_profile().get_nts_priority() == -1); - CHECK(c->configs.needs_push()); - CHECK(stored_dumps(c) == 1); + TestHelper::on_loop(*c, [&] { + // Not blank: creating an account writes the defaults it should start life with, and note + // to self starting hidden is one of them. It is owed to the swarm precisely because it is + // shared -- the account's other devices have to be told, or they would each invent their + // own answer. + CHECK(c->configs.user_profile().get_nts_priority() == -1); + CHECK(c->configs.needs_push()); + CHECK(stored_dumps(c) == 1); - // Everything not defaulted is still empty. - CHECK_FALSE(c->configs.user_profile().get_name()); - CHECK(c->configs.contacts().size() == 0); + // Everything not defaulted is still empty. + CHECK_FALSE(c->configs.user_profile().get_name()); + CHECK(c->configs.contacts().size() == 0); - settle_new_account(c); - CHECK_FALSE(c->configs.needs_push()); + settle_new_account(c); + CHECK_FALSE(c->configs.needs_push()); + }); } TEST_CASE("Configs: a namespace names exactly one config", "[core][configs]") { @@ -143,115 +146,138 @@ TEST_CASE("Configs: a namespace names exactly one config", "[core][configs]") { auto base = [](auto& conf) { return static_cast(&conf); }; - CHECK(c->configs.for_namespace(config::Namespace::UserProfile) == - base(c->configs.user_profile())); - CHECK(c->configs.for_namespace(config::Namespace::Contacts) == base(c->configs.contacts())); - CHECK(c->configs.for_namespace(config::Namespace::UserGroups) == - base(c->configs.user_groups())); - CHECK(c->configs.for_namespace(config::Namespace::ConvoInfoVolatile) == - base(c->configs.convo_info_volatile())); - - // Local reports UserProfile's namespace, having none of its own, so the lookup must not be - // answering from storage_namespace() -- if it were, one of these two would win arbitrarily. - CHECK(c->configs.for_namespace(config::Namespace::UserProfile) != base(c->configs.local())); - - // A namespace that holds no config at all. - CHECK(c->configs.for_namespace(config::Namespace::Default) == nullptr); + TestHelper::on_loop(*c, [&] { + CHECK(c->configs.for_namespace(config::Namespace::UserProfile) == + base(c->configs.user_profile())); + CHECK(c->configs.for_namespace(config::Namespace::Contacts) == base(c->configs.contacts())); + CHECK(c->configs.for_namespace(config::Namespace::UserGroups) == + base(c->configs.user_groups())); + CHECK(c->configs.for_namespace(config::Namespace::ConvoInfoVolatile) == + base(c->configs.convo_info_volatile())); + + // Local reports UserProfile's namespace, having none of its own, so the lookup must not be + // answering from storage_namespace() -- if it were, one of these two would win + // arbitrarily. + CHECK(c->configs.for_namespace(config::Namespace::UserProfile) != base(c->configs.local())); + + // A namespace that holds no config at all. + CHECK(c->configs.for_namespace(config::Namespace::Default) == nullptr); + }); } TEST_CASE("Configs: a dumped config survives a restart", "[core][configs]") { TempCore c{}; - c->configs.user_profile().set_name("Leia"); - c->configs.local().set_setting("some_toggle", true); - c->configs.store_dumps(); + TestHelper::on_loop(*c, [&] { + c->configs.user_profile().set_name("Leia"); + c->configs.local().set_setting("some_toggle", true); + c->configs.store_dumps(); + }); + // Outside the wrapper on purpose: this destroys the Core, and so the very loop the wrapper + // would be running on. reopen(c); - CHECK(c->configs.user_profile().get_name() == "Leia"); - CHECK(c->configs.local().get_setting("some_toggle") == true); + TestHelper::on_loop(*c, [&] { + CHECK(c->configs.user_profile().get_name() == "Leia"); + CHECK(c->configs.local().get_setting("some_toggle") == true); - // Reloading is not a change, so it owes no new dump. - CHECK_FALSE(c->configs.user_profile().needs_dump()); + // Reloading is not a change, so it owes no new dump. + CHECK_FALSE(c->configs.user_profile().needs_dump()); + }); } TEST_CASE("Configs: merging what another device pushed", "[core][configs]") { TempCore c{}; - auto pushed = push_from_another_device(c, "Padmé"); - auto incoming = as_swarm_messages(pushed); - c->receive_messages(incoming, config::Namespace::UserProfile, true); + TestHelper::on_loop(*c, [&] { + auto pushed = push_from_another_device(c, "Padmé"); + auto incoming = as_swarm_messages(pushed); + c->receive_messages(incoming, config::Namespace::UserProfile, true); - CHECK(c->configs.user_profile().get_name() == "Padmé"); + CHECK(c->configs.user_profile().get_name() == "Padmé"); - // Adopting someone else's config outright is not a change of ours, so there is nothing to push - // back -- but it is a change to what we hold, so it is written out. - CHECK_FALSE(c->configs.needs_push()); - CHECK(stored_dumps(c) == 1); + // Adopting someone else's config outright is not a change of ours, so there is nothing to + // push back -- but it is a change to what we hold, so it is written out. + CHECK_FALSE(c->configs.needs_push()); + CHECK(stored_dumps(c) == 1); + }); reopen(c); - CHECK(c->configs.user_profile().get_name() == "Padmé"); + TestHelper::on_loop(*c, [&] { CHECK(c->configs.user_profile().get_name() == "Padmé"); }); } TEST_CASE("Configs: a local change survives merging a config that predates it", "[core][configs]") { TempCore c{}; - // Our own unpushed change participates in the merge as though it had been pushed, so a config - // from another device that has never heard of it does not erase it. This is what makes "in our - // database but not in the config" mean deleted elsewhere rather than not yet synced. - c->configs.contacts().set(c->configs.contacts().get_or_construct("05" + std::string(64, 'a'))); - REQUIRE(c->configs.contacts().size() == 1); + TestHelper::on_loop(*c, [&] { + // Our own unpushed change participates in the merge as though it had been pushed, so a + // config from another device that has never heard of it does not erase it. This is what + // makes "in our database but not in the config" mean deleted elsewhere rather than not yet + // synced. + c->configs.contacts().set( + c->configs.contacts().get_or_construct("05" + std::string(64, 'a'))); + REQUIRE(c->configs.contacts().size() == 1); - auto seed = c->globals.account_seed(); - config::Contacts theirs{seed.ed25519_secret(), std::nullopt}; - theirs.set(theirs.get_or_construct("05" + std::string(64, 'b'))); - auto [seqno, messages, obsolete] = theirs.push(); + auto seed = c->globals.account_seed(); + config::Contacts theirs{seed.ed25519_secret(), std::nullopt}; + theirs.set(theirs.get_or_construct("05" + std::string(64, 'b'))); + auto [seqno, messages, obsolete] = theirs.push(); - auto incoming = as_swarm_messages(messages); - c->receive_messages(incoming, config::Namespace::Contacts, true); + auto incoming = as_swarm_messages(messages); + c->receive_messages(incoming, config::Namespace::Contacts, true); - // Both contacts are present, and the merged result is ours to push since only we hold it. - CHECK(c->configs.contacts().size() == 2); - CHECK(c->configs.needs_push()); + // Both contacts are present, and the merged result is ours to push since only we hold it. + CHECK(c->configs.contacts().size() == 2); + CHECK(c->configs.needs_push()); + }); } TEST_CASE("Configs: Local is never owed to a swarm", "[core][configs]") { TempCore c{}; - settle_new_account(c); - c->configs.local().set_setting("a_toggle", true); + TestHelper::on_loop(*c, [&] { + settle_new_account(c); - // The change is real -- it is held, and dumped like any other config... - CHECK(c->configs.local().get_setting("a_toggle") == true); - CHECK(c->configs.local().needs_dump()); + c->configs.local().set_setting("a_toggle", true); - // ...but Local has no swarm, so it can never make the account owe a push. It declines on its - // own account (needs_push() is overridden to false) and is also left out of the pushable set, - // so neither alone is load-bearing. - CHECK_FALSE(c->configs.local().needs_push()); - CHECK_FALSE(c->configs.needs_push()); + // The change is real -- it is held, and dumped like any other config... + CHECK(c->configs.local().get_setting("a_toggle") == true); + CHECK(c->configs.local().needs_dump()); + + // ...but Local has no swarm, so it can never make the account owe a push. It declines on + // its own account (needs_push() is overridden to false) and is also left out of the + // pushable set, so neither alone is load-bearing. + CHECK_FALSE(c->configs.local().needs_push()); + CHECK_FALSE(c->configs.needs_push()); + }); } TEST_CASE("Configs: a batch holds back the dump", "[core][configs]") { TempCore c{}; - settle_new_account(c); - // Contacts rather than UserProfile, because a new account has already dumped the latter to - // record its defaults: counting rows only shows the deferral for a config that has none yet. - auto pushed = contacts_from_another_device(c, "05" + std::string(64, 'a')); - auto profile = as_swarm_messages(pushed); - REQUIRE(stored_dumps(c) == 1); + TestHelper::on_loop(*c, [&] { + settle_new_account(c); - { - auto held = c->configs.batch(); - c->receive_messages(profile, config::Namespace::Contacts, true); + // Contacts rather than UserProfile, because a new account has already dumped the latter to + // record its defaults: counting rows only shows the deferral for a config that has none + // yet. + auto pushed = contacts_from_another_device(c, "05" + std::string(64, 'a')); + auto profile = as_swarm_messages(pushed); + REQUIRE(stored_dumps(c) == 1); - // The merge landed, but writing it out is deferred: nothing reads a half-processed batch. - CHECK(c->configs.contacts().size() == 1); - CHECK(stored_dumps(c) == 1); - } + { + auto held = c->configs.batch(); + c->receive_messages(profile, config::Namespace::Contacts, true); - CHECK(stored_dumps(c) == 2); + // The merge landed, but writing it out is deferred: nothing reads a half-processed + // batch. + CHECK(c->configs.contacts().size() == 1); + CHECK(stored_dumps(c) == 1); + } + + CHECK(stored_dumps(c) == 2); + }); } namespace { @@ -275,40 +301,45 @@ TEST_CASE("Configs: a merge that changed something is reported", "[core][configs ChangeWatcher w; TempCore c{w.callbacks()}; - auto pushed = push_from_another_device(c, "Padmé"); - auto incoming = as_swarm_messages(pushed); - c->receive_messages(incoming, config::Namespace::UserProfile, true); + TestHelper::on_loop(*c, [&] { + auto pushed = push_from_another_device(c, "Padmé"); + auto incoming = as_swarm_messages(pushed); + c->receive_messages(incoming, config::Namespace::UserProfile, true); - REQUIRE(w.reported.size() == 1); - CHECK(w.reported[0] == std::vector{config::Namespace::UserProfile}); + REQUIRE(w.reported.size() == 1); + CHECK(w.reported[0] == std::vector{config::Namespace::UserProfile}); - // The same message again changes nothing, and nothing is what gets reported -- otherwise every - // poll that re-fetched the same config would send the application round the houses again. - c->receive_messages(incoming, config::Namespace::UserProfile, true); - CHECK(w.reported.size() == 1); + // The same message again changes nothing, and nothing is what gets reported -- otherwise + // every poll that re-fetched the same config would send the application round the houses + // again. + c->receive_messages(incoming, config::Namespace::UserProfile, true); + CHECK(w.reported.size() == 1); + }); } TEST_CASE("Configs: one batch reports everything it changed, once", "[core][configs][notify]") { ChangeWatcher w; TempCore c{w.callbacks()}; - auto profile_pushed = push_from_another_device(c, "Leia"); - auto profile = as_swarm_messages(profile_pushed); - auto contacts_pushed = contacts_from_another_device(c, "05" + std::string(64, 'a')); - auto contacts = as_swarm_messages(contacts_pushed); + TestHelper::on_loop(*c, [&] { + auto profile_pushed = push_from_another_device(c, "Leia"); + auto profile = as_swarm_messages(profile_pushed); + auto contacts_pushed = contacts_from_another_device(c, "05" + std::string(64, 'a')); + auto contacts = as_swarm_messages(contacts_pushed); - { - auto held = c->configs.batch(); - c->receive_messages(profile, config::Namespace::UserProfile, true); - c->receive_messages(contacts, config::Namespace::Contacts, true); - } + { + auto held = c->configs.batch(); + c->receive_messages(profile, config::Namespace::UserProfile, true); + c->receive_messages(contacts, config::Namespace::Contacts, true); + } - // One notification carrying both, not one per config: a poll can deliver all four, and telling - // the application about each in turn shows it a half-applied state. - REQUIRE(w.reported.size() == 1); - auto changed = w.reported[0]; - std::ranges::sort(changed); - CHECK(changed == std::vector{config::Namespace::UserProfile, config::Namespace::Contacts}); + // One notification carrying both, not one per config: a poll can deliver all four, and + // telling the application about each in turn shows it a half-applied state. + REQUIRE(w.reported.size() == 1); + auto changed = w.reported[0]; + std::ranges::sort(changed); + CHECK(changed == std::vector{config::Namespace::UserProfile, config::Namespace::Contacts}); + }); } TEST_CASE("Configs: a conflicting merge at our own seqno is reported", "[core][configs][notify]") { @@ -316,8 +347,14 @@ TEST_CASE("Configs: a conflicting merge at our own seqno is reported", "[core][c TempCore c{w.callbacks()}; // A local change of our own, at some seqno. - c->configs.contacts().set(c->configs.contacts().get_or_construct("05" + std::string(64, 'a'))); - auto our_seqno = c->configs.contacts().seqno(); + // + // Wrapped in pieces rather than all at once: GENERATE and SECTION below have to stay at test + // scope, since Catch2 re-runs the case for each of them. + auto our_seqno = TestHelper::on_loop(*c, [&] { + c->configs.contacts().set( + c->configs.contacts().get_or_construct("05" + std::string(64, 'a'))); + return c->configs.contacts().seqno(); + }); // Our config can be in either of two states here, and _merge takes a different path for each: // Dirty means the change has not been serialised into a message at all, so nothing else can @@ -326,10 +363,12 @@ TEST_CASE("Configs: a conflicting merge at our own seqno is reported", "[core][c // sections below into all four combinations rather than replacing them. const bool already_a_message = GENERATE(false, true); CAPTURE(already_a_message); - if (already_a_message) - c->configs.contacts().push(); - REQUIRE(c->configs.contacts().is_dirty() == !already_a_message); - REQUIRE(c->configs.contacts().seqno() == our_seqno); + TestHelper::on_loop(*c, [&] { + if (already_a_message) + c->configs.contacts().push(); + REQUIRE(c->configs.contacts().is_dirty() == !already_a_message); + REQUIRE(c->configs.contacts().seqno() == our_seqno); + }); // Another device changed things from the same starting point, so its push carries the *same* // seqno as ours with different contents. Both shapes of disagreement are worth covering: one @@ -348,28 +387,30 @@ TEST_CASE("Configs: a conflicting merge at our own seqno is reported", "[core][c expected_contacts = 2; } - std::vector> pushed; - { - auto seed = c->globals.account_seed(); - config::Contacts theirs{seed.ed25519_secret(), std::nullopt}; - for (const auto& id : theirs_has) - theirs.set(theirs.get_or_construct(id)); - REQUIRE(theirs.seqno() == our_seqno); - auto [seqno, messages, obsolete] = theirs.push(); - pushed = std::move(messages); - } - auto incoming = as_swarm_messages(pushed); - - c->receive_messages(incoming, config::Namespace::Contacts, true); - - // The data changed, so the application has to be told. The risk being checked is that - // resolving two same-numbered configs might leave the seqno where it was, which a seqno - // comparison would then miss. It does not: two distinct messages at one seqno are a conflict - // whatever their contents, and a conflict resolves to one past the highest. - CHECK(c->configs.contacts().size() == expected_contacts); - CHECK(c->configs.contacts().seqno() > our_seqno); - REQUIRE(w.reported.size() == 1); - CHECK(w.reported[0] == std::vector{config::Namespace::Contacts}); + TestHelper::on_loop(*c, [&] { + std::vector> pushed; + { + auto seed = c->globals.account_seed(); + config::Contacts theirs{seed.ed25519_secret(), std::nullopt}; + for (const auto& id : theirs_has) + theirs.set(theirs.get_or_construct(id)); + REQUIRE(theirs.seqno() == our_seqno); + auto [seqno, messages, obsolete] = theirs.push(); + pushed = std::move(messages); + } + auto incoming = as_swarm_messages(pushed); + + c->receive_messages(incoming, config::Namespace::Contacts, true); + + // The data changed, so the application has to be told. The risk being checked is that + // resolving two same-numbered configs might leave the seqno where it was, which a seqno + // comparison would then miss. It does not: two distinct messages at one seqno are a + // conflict whatever their contents, and a conflict resolves to one past the highest. + CHECK(c->configs.contacts().size() == expected_contacts); + CHECK(c->configs.contacts().seqno() > our_seqno); + REQUIRE(w.reported.size() == 1); + CHECK(w.reported[0] == std::vector{config::Namespace::Contacts}); + }); } TEST_CASE("Configs: merging a change identical to our own", "[core][configs][notify]") { @@ -377,21 +418,23 @@ TEST_CASE("Configs: merging a change identical to our own", "[core][configs][not TempCore c{w.callbacks()}; auto contact = "05" + std::string(64, 'a'); - c->configs.contacts().set(c->configs.contacts().get_or_construct(contact)); - auto our_seqno = c->configs.contacts().seqno(); - // Another device made the very same change from the same starting point. - auto pushed = contacts_from_another_device(c, contact); - auto incoming = as_swarm_messages(pushed); - c->receive_messages(incoming, config::Namespace::Contacts, true); + TestHelper::on_loop(*c, [&] { + c->configs.contacts().set(c->configs.contacts().get_or_construct(contact)); - // Nothing is lost: we already held exactly what arrived. - CHECK(c->configs.contacts().size() == 1); - CHECK(c->configs.contacts().get(contact).has_value()); + // Another device made the very same change from the same starting point. + auto pushed = contacts_from_another_device(c, contact); + auto incoming = as_swarm_messages(pushed); + c->receive_messages(incoming, config::Namespace::Contacts, true); + + // Nothing is lost: we already held exactly what arrived. + CHECK(c->configs.contacts().size() == 1); + CHECK(c->configs.contacts().get(contact).has_value()); - // And nothing is owed: agreeing with another device settles clean against that device's - // message rather than leaving us dirty, so it costs no push carrying no changes. - CHECK_FALSE(c->configs.contacts().needs_push()); + // And nothing is owed: agreeing with another device settles clean against that device's + // message rather than leaving us dirty, so it costs no push carrying no changes. + CHECK_FALSE(c->configs.contacts().needs_push()); + }); // The seqno is deliberately not asserted. It currently advances even though the data did not, // because merging while dirty builds a MutableConfigMessage and that constructor increments @@ -411,165 +454,183 @@ TEST_CASE("Configs: an adopted duplicate leaves no gap behind", "[core][configs] ChangeWatcher w; TempCore c{w.callbacks()}; - auto seed = c->globals.account_seed(); - config::Contacts them{seed.ed25519_secret(), std::nullopt}; - - auto a = "05" + std::string(64, 'a'); - auto b = "05" + std::string(64, 'b'); - - // Both devices make the same change from the same starting point. - c->configs.contacts().set(c->configs.contacts().get_or_construct(a)); - them.set(them.get_or_construct(a)); - REQUIRE(c->configs.contacts().seqno() == them.seqno()); - auto agreed_seqno = them.seqno(); - - { - auto [seqno, messages, obsolete] = them.push(); - them.confirm_pushed(seqno, {"theirs1"}); - auto incoming = as_swarm_messages(messages, "theirs"); - c->receive_messages(incoming, config::Namespace::Contacts, true); - } + TestHelper::on_loop(*c, [&] { + auto seed = c->globals.account_seed(); + config::Contacts them{seed.ed25519_secret(), std::nullopt}; + + auto a = "05" + std::string(64, 'a'); + auto b = "05" + std::string(64, 'b'); + + // Both devices make the same change from the same starting point. + c->configs.contacts().set(c->configs.contacts().get_or_construct(a)); + them.set(them.get_or_construct(a)); + REQUIRE(c->configs.contacts().seqno() == them.seqno()); + auto agreed_seqno = them.seqno(); + + { + auto [seqno, messages, obsolete] = them.push(); + them.confirm_pushed(seqno, {"theirs1"}); + auto incoming = as_swarm_messages(messages, "theirs"); + c->receive_messages(incoming, config::Namespace::Contacts, true); + } - // Adopting their identical config leaves us on the seqno the swarm actually holds, rather than - // one past it: no number is consumed that no stored message occupies. - CHECK(c->configs.contacts().seqno() == agreed_seqno); - CHECK_FALSE(c->configs.contacts().needs_push()); + // Adopting their identical config leaves us on the seqno the swarm actually holds, rather + // than one past it: no number is consumed that no stored message occupies. + CHECK(c->configs.contacts().seqno() == agreed_seqno); + CHECK_FALSE(c->configs.contacts().needs_push()); - // So the other device's further change of its own lands on the next number up, with nothing of - // ours already sitting on it. - them.set(them.get_or_construct(b)); - REQUIRE(them.seqno() == agreed_seqno + 1); + // So the other device's further change of its own lands on the next number up, with + // nothing of ours already sitting on it. + them.set(them.get_or_construct(b)); + REQUIRE(them.seqno() == agreed_seqno + 1); - { - auto [seqno, messages, obsolete] = them.push(); - auto incoming = as_swarm_messages(messages, "theirs2-"); - c->receive_messages(incoming, config::Namespace::Contacts, true); - } + { + auto [seqno, messages, obsolete] = them.push(); + auto incoming = as_swarm_messages(messages, "theirs2-"); + c->receive_messages(incoming, config::Namespace::Contacts, true); + } - // Their change arrives intact and ours is still there. - CHECK(c->configs.contacts().size() == 2); - CHECK(c->configs.contacts().get(a).has_value()); - CHECK(c->configs.contacts().get(b).has_value()); - - // ...and it is adopted at its own seqno rather than resolving as a conflict one past both, so - // we are left neither dirty nor owing a push for a change that was never ours. One real - // change, one seqno -- which is also what keeps the "within N" conflict window from spending - // two of its five carrying a single change. - CHECK(c->configs.contacts().seqno() == agreed_seqno + 1); - CHECK_FALSE(c->configs.contacts().needs_push()); + // Their change arrives intact and ours is still there. + CHECK(c->configs.contacts().size() == 2); + CHECK(c->configs.contacts().get(a).has_value()); + CHECK(c->configs.contacts().get(b).has_value()); + + // ...and it is adopted at its own seqno rather than resolving as a conflict one past both, + // so we are left neither dirty nor owing a push for a change that was never ours. One + // real change, one seqno -- which is also what keeps the "within N" conflict window from + // spending two of its five carrying a single change. + CHECK(c->configs.contacts().seqno() == agreed_seqno + 1); + CHECK_FALSE(c->configs.contacts().needs_push()); + }); } TEST_CASE("Configs: a local change is not reported back", "[core][configs][notify]") { ChangeWatcher w; TempCore c{w.callbacks()}; - { - auto held = c->configs.batch(); - c->configs.user_profile().set_name("Leia"); - } + TestHelper::on_loop(*c, [&] { + { + auto held = c->configs.batch(); + c->configs.user_profile().set_name("Leia"); + } - // The application made this change; being told about it would be news to nobody, and would - // invite it to reconcile its own write back over itself. - CHECK(w.reported.empty()); + // The application made this change; being told about it would be news to nobody, and + // would invite it to reconcile its own write back over itself. + CHECK(w.reported.empty()); + }); } TEST_CASE("Configs: a change goes out as one signed sequence", "[core][configs][push]") { PushableCore c; - c->configs.user_profile().set_name("Leia"); - // Local changes too: it must not appear in what goes out. - c->configs.local().set_setting("a_toggle", true); - c->configs.push_now(); - - CHECK(c.only_request().endpoint == "sequence"); - - auto subs = c.subrequests(); - REQUIRE(subs.size() == 1); - CHECK(subs[0]["method"] == "store"); - CHECK(subs[0]["params"]["namespace"] == 2); - CHECK(subs[0]["params"]["ttl"] == std::chrono::milliseconds{30 * 24h}.count()); - // Config namespaces are owner-write, so the store carries a signature and the key to check it. - CHECK(subs[0]["params"].contains("signature")); - CHECK(subs[0]["params"].contains("pubkey_ed25519")); - - // Nothing is obsolete on a first push, so there is nothing to delete. - CHECK_FALSE(subs[0].contains("delete")); + TestHelper::on_loop(*c.core, [&] { + c->configs.user_profile().set_name("Leia"); + // Local changes too: it must not appear in what goes out. + c->configs.local().set_setting("a_toggle", true); + c->configs.push_now(); + + CHECK(c.only_request().endpoint == "sequence"); + + auto subs = c.subrequests(); + REQUIRE(subs.size() == 1); + CHECK(subs[0]["method"] == "store"); + CHECK(subs[0]["params"]["namespace"] == 2); + CHECK(subs[0]["params"]["ttl"] == std::chrono::milliseconds{30 * 24h}.count()); + // Config namespaces are owner-write, so the store carries a signature and the key to + // check it. + CHECK(subs[0]["params"].contains("signature")); + CHECK(subs[0]["params"].contains("pubkey_ed25519")); + + // Nothing is obsolete on a first push, so there is nothing to delete. + CHECK_FALSE(subs[0].contains("delete")); + }); } TEST_CASE("Configs: two dirty configs share one request", "[core][configs][push]") { PushableCore c; - c->configs.user_profile().set_name("Leia"); - c->configs.contacts().set(c->configs.contacts().get_or_construct("05" + std::string(64, 'a'))); - c->configs.push_now(); - - auto subs = c.subrequests(); - REQUIRE(subs.size() == 2); - std::vector namespaces{subs[0]["params"]["namespace"], subs[1]["params"]["namespace"]}; - std::ranges::sort(namespaces); - CHECK(namespaces == std::vector{2, 3}); + TestHelper::on_loop(*c.core, [&] { + c->configs.user_profile().set_name("Leia"); + c->configs.contacts().set( + c->configs.contacts().get_or_construct("05" + std::string(64, 'a'))); + c->configs.push_now(); + + auto subs = c.subrequests(); + REQUIRE(subs.size() == 2); + std::vector namespaces{subs[0]["params"]["namespace"], subs[1]["params"]["namespace"]}; + std::ranges::sort(namespaces); + CHECK(namespaces == std::vector{2, 3}); + }); } TEST_CASE("Configs: a stored push stops being owed", "[core][configs][push]") { PushableCore c; - c->configs.user_profile().set_name("Leia"); - c->configs.push_now(); + TestHelper::on_loop(*c.core, [&] { + c->configs.user_profile().set_name("Leia"); + c->configs.push_now(); - // Handing it to the swarm is not the same as it having arrived, so it is still owed until the - // store is confirmed -- otherwise a failed push would be forgotten. - CHECK(c->configs.needs_push()); + // Handing it to the swarm is not the same as it having arrived, so it is still owed until + // the store is confirmed -- otherwise a failed push would be forgotten. + CHECK(c->configs.needs_push()); - c.answer({"hash1"}); - CHECK_FALSE(c->configs.needs_push()); + c.answer({"hash1"}); + CHECK_FALSE(c->configs.needs_push()); + }); } TEST_CASE("Configs: a rejected store leaves the config dirty", "[core][configs][push]") { PushableCore c; - c->configs.user_profile().set_name("Leia"); - c->configs.push_now(); - c.answer({std::nullopt}); - - // The change is still ours to deliver, and pushing again offers it again. - CHECK(c->configs.needs_push()); - c->configs.push_now(); - CHECK(c.subrequests().size() == 1); + TestHelper::on_loop(*c.core, [&] { + c->configs.user_profile().set_name("Leia"); + c->configs.push_now(); + c.answer({std::nullopt}); + + // The change is still ours to deliver, and pushing again offers it again. + CHECK(c->configs.needs_push()); + c->configs.push_now(); + CHECK(c.subrequests().size() == 1); + }); } TEST_CASE("Configs: the next push deletes what it replaces", "[core][configs][push]") { PushableCore c; - c->configs.user_profile().set_name("Leia"); - c->configs.push_now(); - c.answer({"hash1"}); - - c->configs.user_profile().set_name("Padmé"); - c->configs.push_now(); - - auto subs = c.subrequests(); - REQUIRE(subs.size() == 2); - CHECK(subs[0]["method"] == "store"); - - // The delete goes last: a sequence stops at its first failure, so nothing is removed before - // what replaces it has been stored. - CHECK(subs[1]["method"] == "delete"); - CHECK(subs[1]["params"]["messages"] == nlohmann::json::array({"hash1"})); - CHECK(subs[1]["params"].contains("signature")); + TestHelper::on_loop(*c.core, [&] { + c->configs.user_profile().set_name("Leia"); + c->configs.push_now(); + c.answer({"hash1"}); + + c->configs.user_profile().set_name("Padmé"); + c->configs.push_now(); + + auto subs = c.subrequests(); + REQUIRE(subs.size() == 2); + CHECK(subs[0]["method"] == "store"); + + // The delete goes last: a sequence stops at its first failure, so nothing is removed + // before what replaces it has been stored. + CHECK(subs[1]["method"] == "delete"); + CHECK(subs[1]["params"]["messages"] == nlohmann::json::array({"hash1"})); + CHECK(subs[1]["params"].contains("signature")); + }); } TEST_CASE("Configs: a change schedules a push rather than sending one", "[core][configs][push]") { PushableCore c; - { - auto held = c->configs.batch(); - c->configs.user_profile().set_name("Leia"); - } + TestHelper::on_loop(*c.core, [&] { + { + auto held = c->configs.batch(); + c->configs.user_profile().set_name("Leia"); + } - // Releasing the batch is what notices the change; it schedules rather than sending, so a run of - // changes coalesces into one request. - CHECK(TestHelper::push_scheduled(c->configs)); - CHECK(c.net->sent_requests.empty()); + // Releasing the batch is what notices the change; it schedules rather than sending, so a + // run of changes coalesces into one request. + CHECK(TestHelper::push_scheduled(c->configs)); + CHECK(c.net->sent_requests.empty()); + }); } TEST_CASE("Configs: the debounce waits for quiet, up to a limit", "[core][configs][push]") { @@ -577,32 +638,42 @@ TEST_CASE("Configs: the debounce waits for quiet, up to a limit", "[core][config c->configs.push_debounce = 2s; c->configs.push_max_delay = 10s; - { - auto held = c->configs.batch(); - c->configs.user_profile().set_name("Leia"); - } - REQUIRE(TestHelper::push_scheduled(c->configs)); + // Wrapped per section rather than all at once: Catch2 re-runs the case for each SECTION, so + // they have to stay at test scope. + TestHelper::on_loop(*c.core, [&] { + { + auto held = c->configs.batch(); + c->configs.user_profile().set_name("Leia"); + } + REQUIRE(TestHelper::push_scheduled(c->configs)); + }); SECTION("changes still arriving hold it back") { - TestHelper::backdate_push_state(c->configs, 500ms, 1s); - TestHelper::push_if_due(c->configs); - CHECK(c.net->sent_requests.empty()); - CHECK(TestHelper::push_scheduled(c->configs)); + TestHelper::on_loop(*c.core, [&] { + TestHelper::backdate_push_state(c->configs, 500ms, 1s); + TestHelper::push_if_due(c->configs); + CHECK(c.net->sent_requests.empty()); + CHECK(TestHelper::push_scheduled(c->configs)); + }); } SECTION("quiet for long enough sends it") { - TestHelper::backdate_push_state(c->configs, 3s, 4s); - TestHelper::push_if_due(c->configs); - CHECK(c.net->sent_requests.size() == 1); - CHECK_FALSE(TestHelper::push_scheduled(c->configs)); + TestHelper::on_loop(*c.core, [&] { + TestHelper::backdate_push_state(c->configs, 3s, 4s); + TestHelper::push_if_due(c->configs); + CHECK(c.net->sent_requests.size() == 1); + CHECK_FALSE(TestHelper::push_scheduled(c->configs)); + }); } SECTION("a steady trickle cannot defer it past the cap") { - // Never quiet -- the last change was a moment ago -- but the burst began long enough ago - // that waiting for quiet would mean waiting indefinitely. - TestHelper::backdate_push_state(c->configs, 100ms, 11s); - TestHelper::push_if_due(c->configs); - CHECK(c.net->sent_requests.size() == 1); + TestHelper::on_loop(*c.core, [&] { + // Never quiet -- the last change was a moment ago -- but the burst began long enough + // ago that waiting for quiet would mean waiting indefinitely. + TestHelper::backdate_push_state(c->configs, 100ms, 11s); + TestHelper::push_if_due(c->configs); + CHECK(c.net->sent_requests.size() == 1); + }); } } @@ -610,33 +681,38 @@ TEST_CASE("Configs: pushing can be switched off entirely", "[core][configs][push PushableCore c; c->configs.push_enabled = false; - c->configs.user_profile().set_name("Leia"); - c->configs.push_now(); + TestHelper::on_loop(*c.core, [&] { + c->configs.user_profile().set_name("Leia"); + c->configs.push_now(); - // Nothing goes out... - CHECK(c.net->sent_requests.empty()); + // Nothing goes out... + CHECK(c.net->sent_requests.empty()); - // ...and nothing pretends it did: the change is still held and still owed, so the state reads - // as unpublished rather than as settled. - CHECK(c->configs.user_profile().get_name() == "Leia"); - CHECK(c->configs.needs_push()); + // ...and nothing pretends it did: the change is still held and still owed, so the state + // reads as unpublished rather than as settled. + CHECK(c->configs.user_profile().get_name() == "Leia"); + CHECK(c->configs.needs_push()); - // Switching it back on lets everything accumulated since go out together. - c->configs.push_enabled = true; - c->configs.push_now(); - CHECK(c.net->sent_requests.size() == 1); + // Switching it back on lets everything accumulated since go out together. + c->configs.push_enabled = true; + c->configs.push_now(); + CHECK(c.net->sent_requests.size() == 1); + }); } TEST_CASE("Configs: a push already in flight is not duplicated", "[core][configs][push]") { PushableCore c; - c->configs.user_profile().set_name("Leia"); - c->configs.push_now(); - REQUIRE(c.net->sent_requests.size() == 1); - - // A second change while the first is out must not race it onto the wire; the completion picks - // it up instead. - c->configs.contacts().set(c->configs.contacts().get_or_construct("05" + std::string(64, 'a'))); - c->configs.push_now(); - CHECK(c.net->sent_requests.size() == 1); + TestHelper::on_loop(*c.core, [&] { + c->configs.user_profile().set_name("Leia"); + c->configs.push_now(); + REQUIRE(c.net->sent_requests.size() == 1); + + // A second change while the first is out must not race it onto the wire; the completion + // picks it up instead. + c->configs.contacts().set( + c->configs.contacts().get_or_construct("05" + std::string(64, 'a'))); + c->configs.push_now(); + CHECK(c.net->sent_requests.size() == 1); + }); } From c16f358d87505f8d212a55f4bfb70fd0455418a4 Mon Sep 17 00:00:00 2001 From: Jason Rhinelander Date: Thu, 10 Sep 2026 18:53:20 -0300 Subject: [PATCH 4/5] Finish moving Core's own deferred work onto its queue Two things the poll fix left behind, both the same shape as it. A config push's completion runs on the Network's own loop -- Network builds its own quic::Loop -- and clears `_push_in_flight`, confirms the pushed configs and dumps them, all of which is Configs' state and none of which is safe off Core's loop. The body moves into `_handle_push_response` so the callback can be one line of marshalling, the way `Core::_handle_poll_response` already is, and `Pending` moves to the class with it. Note what does *not* change: the `_alive` canary stays, and there is now a comment saying why. The Network owns these callbacks, so they can outlive Core entirely, and a stopped queue cannot cancel something that was never queued -- the canary is what makes reaching `jq()` safe in the first place, and the queue takes over from there. And `_poll_ticker` was declared with the rest of the polling machinery, so it was destroyed *after* the components a firing poll reaches. It is declared last now, which destroys it first: no poll can be in flight by the time anything it touches is being torn down. Before `_jq` rather than after, so that a poll cannot try to queue its response onto a queue that has already stopped, which throws rather than being ignored. --- include/session/core.hpp | 19 +++- include/session/core/configs.hpp | 30 +++++- src/core/configs.cpp | 172 +++++++++++++++++-------------- 3 files changed, 135 insertions(+), 86 deletions(-) diff --git a/include/session/core.hpp b/include/session/core.hpp index ebff94c3c..27a034683 100644 --- a/include/session/core.hpp +++ b/include/session/core.hpp @@ -326,9 +326,9 @@ class Core { // migrations, and then calls init() on each sub-component. void init(); - // Polling-related members and methods + // Polling-related members and methods. The ticker itself is declared at the bottom of the + // class, with the rest of what has to be torn down before the components it reaches. std::chrono::milliseconds _poll_interval = 20s; - std::shared_ptr _poll_ticker; void _update_polling(); void _poll(); @@ -694,10 +694,19 @@ class Core { // whatever is still outstanding is *cancelled* when Core goes away instead of running against // components that are already destroyed -- the same reason Client keeps its own. // - // Declared last so it is destroyed first, before the components its jobs reach. It has to be - // destroyed while `_loop` is still alive, which it is: `_loop` is declared first and so is - // destroyed last. + // Declared near the bottom so it is destroyed early, before the components its jobs reach. It + // has to be destroyed while `_loop` is still alive, which it is: `_loop` is declared first and + // so is destroyed last. quic::JobQueue _jq{_loop}; + + // Last of all, so it is the *first* thing destroyed: a ticker still running is a poll still + // arriving, and a poll reaches every component. Stopping it before `_jq` rather than after + // also means no poll can try to queue its response onto a queue that has already stopped, + // which throws. + // + // (It lives here rather than beside `_poll_interval` for that reason alone; everything that + // uses it is up there with the rest of the polling machinery.) + std::shared_ptr _poll_ticker; }; } // namespace session::core diff --git a/include/session/core/configs.hpp b/include/session/core/configs.hpp index 0267fe8b7..1cadcd6d1 100644 --- a/include/session/core/configs.hpp +++ b/include/session/core/configs.hpp @@ -1,8 +1,12 @@ #pragma once +#include #include +#include #include +#include #include +#include #include #include "component.hpp" @@ -77,16 +81,36 @@ class Configs : public detail::CoreComponent { bool _push_scheduled = false; bool _push_in_flight = false; - // Deferred work is handed to the event loop, which outlives this component and has no way to - // cancel a call already scheduled. Callbacks capture a weak reference to this and do nothing - // if it has expired, which is what stops a pending push firing into a destroyed Core. + // The network's callbacks outlive this component and are owned by the Network rather than by + // Core's job queue, so cancelling the queue cannot reach them: they capture a weak reference + // to this and do nothing if it has expired. That is what stops a push completing into a + // destroyed Core -- the queue only takes over once the callback has safely got that far. std::shared_ptr _alive = std::make_shared(0); + // Which subrequests belong to which config, so that a result can be matched back to the config + // whose push produced it. A sequence answers positionally, so this is the only link. + struct Pending { + config::ConfigBase* conf; + config::seqno_t seqno; + size_t first; + size_t count; + }; + void _schedule_push(); void _arm_push_timer(std::chrono::milliseconds delay); void _push_if_due(); void _send_push(); + // Applies a config push's answer. Split out from the callback that receives it because that + // one arrives on the Network's own loop, and all of this is Configs' state: the callback + // checks it is still alive and hands this to Core's queue. + void _handle_push_response( + std::vector pending, + bool success, + bool timeout, + int16_t status, + std::optional resp); + // Constructs the configs from their stored dumps, or empty if there are none. Requires an // account; throws globals::no_account if there is not one yet. void _load(); diff --git a/src/core/configs.cpp b/src/core/configs.cpp index ed01e2282..1e00b9fcc 100644 --- a/src/core/configs.cpp +++ b/src/core/configs.cpp @@ -322,15 +322,6 @@ void Configs::_send_push() { return; } - // Which subrequests belong to which config, so that a result can be matched back to the config - // whose push produced it. A sequence answers positionally, so this is the only link. - struct Pending { - config::ConfigBase* conf; - config::seqno_t seqno; - size_t first; - size_t count; - }; - auto now_ms = epoch_ms(clock_now_ms()); auto pubkey_hex = core.globals.session_id_hex(); auto ed25519_hex = core.globals.pubkey_ed25519().hex(); @@ -414,7 +405,9 @@ void Configs::_send_push() { return; if (swarm.empty()) { log::warning(cat, "Cannot push configs: no swarm nodes available"); - _push_in_flight = false; + // Onto Core's queue like everything else here: this handler runs on the + // Network's own loop, which is a different thread, and the flag is ours. + jq().call([this] { _push_in_flight = false; }); return; } @@ -429,78 +422,101 @@ void Configs::_send_push() { bool timeout, int16_t status, auto, - std::optional resp) { + std::optional resp) mutable { + // The canary first, because reaching `jq()` at all means touching + // this: the Network owns this callback, so it can outlive Core, and + // cancelling the queue cannot reach something that was never on it. if (alive.expired()) return; - _push_in_flight = false; - - if (!success || !resp) { - log::warning( - cat, - "Config push failed ({}): {}", - timeout ? "timed out" : "status {}"_format(status), - resp.value_or("no response body")); - return; - } - - // A config is confirmed only if *every* message it split into was - // stored. Confirming a partial push would drop the parts that did - // land from the obsolete list while leaving the config believing it - // is clean, so the missing part would never be sent again. - try { - auto json = nlohmann::json::parse(*resp); - auto results = json.find("results"); - if (results == json.end() || !results->is_array()) { - log::warning(cat, "Config push response carried no results"); - return; - } - - for (const auto& p : pending) { - std::unordered_set hashes; - bool stored = true; - for (size_t i = p.first; stored && i < p.first + p.count; i++) { - if (i >= results->size()) { - stored = false; - break; - } - const auto& r = (*results)[i]; - auto code = r.find("code"); - auto b = r.find("body"); - if (code == r.end() || code->get() != 200 || - b == r.end()) { - stored = false; - break; - } - auto h = b->find("hash"); - if (h == b->end() || !h->is_string()) { - stored = false; - break; - } - hashes.insert(h->get()); - } - - if (!stored) { - log::warning( - cat, - "Config push: {} was not stored, leaving it dirty", - p.conf->encryption_domain()); - continue; - } - p.conf->confirm_pushed(p.seqno, std::move(hashes)); - } - } catch (const std::exception& e) { - log::warning( - cat, "Could not read config push response: {}", e.what()); - return; - } - - // Confirming changes the configs' state, and a change that arrived - // while this was in flight has re-dirtied them. - store_dumps(); - if (needs_push()) - _schedule_push(); + jq().call([this, + pending = std::move(pending), + success, + timeout, + status, + resp = std::move(resp)]() mutable { + _handle_push_response( + std::move(pending), + success, + timeout, + status, + std::move(resp)); + }); }); }); } +void Configs::_handle_push_response( + std::vector pending, + bool success, + bool timeout, + int16_t status, + std::optional resp) { + assert(on_loop()); + + _push_in_flight = false; + + if (!success || !resp) { + log::warning( + cat, + "Config push failed ({}): {}", + timeout ? "timed out" : "status {}"_format(status), + resp.value_or("no response body")); + return; + } + + // A config is confirmed only if *every* message it split into was stored. Confirming a + // partial push would drop the parts that did land from the obsolete list while leaving the + // config believing it is clean, so the missing part would never be sent again. + try { + auto json = nlohmann::json::parse(*resp); + auto results = json.find("results"); + if (results == json.end() || !results->is_array()) { + log::warning(cat, "Config push response carried no results"); + return; + } + + for (const auto& p : pending) { + std::unordered_set hashes; + bool stored = true; + for (size_t i = p.first; stored && i < p.first + p.count; i++) { + if (i >= results->size()) { + stored = false; + break; + } + const auto& r = (*results)[i]; + auto code = r.find("code"); + auto b = r.find("body"); + if (code == r.end() || code->get() != 200 || b == r.end()) { + stored = false; + break; + } + auto h = b->find("hash"); + if (h == b->end() || !h->is_string()) { + stored = false; + break; + } + hashes.insert(h->get()); + } + + if (!stored) { + log::warning( + cat, + "Config push: {} was not stored, leaving it dirty", + p.conf->encryption_domain()); + continue; + } + p.conf->confirm_pushed(p.seqno, std::move(hashes)); + } + } catch (const std::exception& e) { + log::warning(cat, "Could not read config push response: {}", e.what()); + return; + } + + // Confirming changes the configs' state, and a change that arrived while this was in flight + // has re-dirtied them. + store_dumps(); + if (needs_push()) + _schedule_push(); +} + } // namespace session::core From 7ca67d607f7eacc841f716e9815bdf87a09a4134 Mon Sep 17 00:00:00 2001 From: Jason Rhinelander Date: Thu, 10 Sep 2026 19:05:38 -0300 Subject: [PATCH 5/5] Run the rest of the tests on the loop too The remaining reads of `core.configs` from `test_client/`, which is the other half of what the assertion is about: a test is an application like any other, and the configs belong to Core's loop. `in_configs` goes in `common.hpp` rather than `config_helpers.hpp`, since it is not about config reconciliation -- it is for any test reaching past `Client` to check what was written underneath. It hands back a value on purpose: several of these read `get_name()`, which returns a `string_view` into the config, and that is dangling the moment the excursion ends. Those copy inside the lambda. Three sites bound `auto& contacts = c->core.configs.contacts()` and used it across several statements. A reference is exactly what cannot leave the loop, so they read what they need each time instead. `TestHelper::sync_contact` and `sync_convo_volatile` hop for themselves rather than making every caller do it: what they reach is a `_`-form on Client, which is only ever called from inside `_async` in the real thing. The `merge_*` helpers wrap their `receive_messages` for the same reason -- a real poll's merge arrives on the loop. Three tests were already doing this hop by hand with `loop().call_get`, which is where the idea came from; one of those is now spelled the same way as the rest. --- tests/test_client/attachments.cpp | 4 +- tests/test_client/common.hpp | 14 ++++++ tests/test_client/config_helpers.hpp | 13 +++-- tests/test_client/configs.cpp | 69 +++++++++++++++----------- tests/test_client/conversation_api.cpp | 16 ++++-- tests/test_client/receiving.cpp | 11 ++-- tests/test_client/requests.cpp | 8 ++- tests/test_client/volatile.cpp | 10 ++-- tests/test_helper.hpp | 8 ++- 9 files changed, 102 insertions(+), 51 deletions(-) diff --git a/tests/test_client/attachments.cpp b/tests/test_client/attachments.cpp index fe3d553d3..2769108de 100644 --- a/tests/test_client/attachments.cpp +++ b/tests/test_client/attachments.cpp @@ -329,7 +329,7 @@ TEST_CASE( // The account's own answer refuses the notification even when the caller asked for it, so a // client that never grew a setting for this still honours one made on another device. { - c->core.configs.user_profile().set_notify_media_saved(false); + in_configs(*c, [](auto& cfg) { cfg.user_profile().set_notify_media_saved(false); }); auto loud = dir / "still-quiet.bin"; auto waiter = save(loud, true); REQUIRE(serve_downloads(*net, ciphertext) == 1); @@ -338,7 +338,7 @@ TEST_CASE( CHECK(std::filesystem::exists(loud)); sync(*c); CHECK(stores(*net).empty()); - c->core.configs.user_profile().set_notify_media_saved(true); + in_configs(*c, [](auto& cfg) { cfg.user_profile().set_notify_media_saved(true); }); } // A file that fails to authenticate is a failure, not a corrupt file on disk: the ciphertext is diff --git a/tests/test_client/common.hpp b/tests/test_client/common.hpp index 2c2e4d37a..d8db41269 100644 --- a/tests/test_client/common.hpp +++ b/tests/test_client/common.hpp @@ -241,6 +241,20 @@ inline std::string preview_body(const AnyConversation& c) { return p ? p->body : ""; } +/// Reads something out of the account's configs, on Core's loop where they belong. +/// +/// The configs are the loop's -- see `core::detail::CoreComponent` -- and a test is on its own +/// thread like any other application. `Client` wraps the parts an application actually wants; +/// this is for the tests that reach past it to check what was written underneath. +/// +/// **Return a value.** Anything pointing into a config -- a reference, or the `string_view` that +/// getters like `get_name()` hand back -- is dangling the moment this returns, since the config is +/// only ours for as long as the excursion lasts. Copy it inside the lambda. +template +auto in_configs(Client& c, F&& f) { + return TestHelper::on_loop(c.core, [&] { return f(c.core.configs); }); +} + } // namespace client_test using namespace client_test; diff --git a/tests/test_client/config_helpers.hpp b/tests/test_client/config_helpers.hpp index a259de5a1..4b114ee5a 100644 --- a/tests/test_client/config_helpers.hpp +++ b/tests/test_client/config_helpers.hpp @@ -39,7 +39,10 @@ inline void merge_profile(Client& c, const std::vector>& m.data = messages[i]; incoming.push_back(std::move(m)); } - c.core.receive_messages(incoming, config::Namespace::UserProfile, true); + // Merging is config work, so it belongs on Core's loop just as a real poll's would. + TestHelper::on_loop(c.core, [&] { + c.core.receive_messages(incoming, config::Namespace::UserProfile, true); + }); } inline ConversationId self_convo(Client& c) { @@ -133,7 +136,9 @@ inline void merge_volatile(Client& c, const std::vector>& m.data = messages[i]; incoming.push_back(std::move(m)); } - c.core.receive_messages(incoming, config::Namespace::ConvoInfoVolatile, true); + TestHelper::on_loop(c.core, [&] { + c.core.receive_messages(incoming, config::Namespace::ConvoInfoVolatile, true); + }); } inline void merge_contacts(Client& c, const std::vector>& messages) { @@ -144,7 +149,9 @@ inline void merge_contacts(Client& c, const std::vector>& m.data = messages[i]; incoming.push_back(std::move(m)); } - c.core.receive_messages(incoming, config::Namespace::Contacts, true); + TestHelper::on_loop(c.core, [&] { + c.core.receive_messages(incoming, config::Namespace::Contacts, true); + }); } } // namespace client_test diff --git a/tests/test_client/configs.cpp b/tests/test_client/configs.cpp index 9e9f4331d..5e348ac8d 100644 --- a/tests/test_client/configs.cpp +++ b/tests/test_client/configs.cpp @@ -62,11 +62,12 @@ TEST_CASE("Client: re-deriving a contact changes nothing", "[client][configs]") // deriving a config back from those tables is the identity. Anything lost, rounded or // defaulted on the way through shows up here as a config that went dirty -- and a mapping that // dirties on every pass would push a pointless update after every merge, forever. - auto& contacts = c->core.configs.contacts(); - REQUIRE_FALSE(contacts.needs_push()); + REQUIRE_FALSE(in_configs(*c, [](auto& cfg) { return cfg.contacts().needs_push(); })); TestHelper::sync_contact(*c.client, id); - CHECK_FALSE(contacts.needs_push()); - CHECK_FALSE(contacts.needs_dump()); + in_configs(*c, [](auto& cfg) { + CHECK_FALSE(cfg.contacts().needs_push()); + CHECK_FALSE(cfg.contacts().needs_dump()); + }); } TEST_CASE("Client: a contact removed elsewhere takes its history", "[client][configs]") { @@ -129,13 +130,13 @@ TEST_CASE( // Stand in for a crash between committing the row and writing the dump: the tables hold a // contact the config has never heard of. Reconciled inward first, that is indistinguishable // from one deleted elsewhere and would be destroyed with its history. - REQUIRE(c->core.configs.contacts().erase(them)); + REQUIRE(in_configs(*c, [&](auto& cfg) { return cfg.contacts().erase(them); })); c.reopen(); // Startup derives outward before reconciling inward, so it is published rather than deleted. CHECK(c->conversation(id, await)); - CHECK(c->core.configs.contacts().get(them).has_value()); + CHECK(in_configs(*c, [&](auto& cfg) { return cfg.contacts().get(them); }).has_value()); } TEST_CASE("Client: a new account starts with note to self hidden", "[client][configs]") { @@ -145,7 +146,7 @@ TEST_CASE("Client: a new account starts with note to self hidden", "[client][con // Seeded at account creation rather than left at the default, because nts_priority is carried // in the shared UserProfile config: a default of 0 would not merely show the conversation here, // it would make it appear on every other device on the account once they synced. - CHECK(c->core.configs.user_profile().get_nts_priority() == -1); + CHECK(in_configs(*c, [](auto& cfg) { return cfg.user_profile().get_nts_priority(); }) == -1); CHECK_FALSE(listed(*c.client, me)); } @@ -160,8 +161,10 @@ TEST_CASE("Client: writing a note to self reveals it", "[client][configs]") { // Both halves: it is in our own list, and UserProfile says so, which is what stops the other // devices on the account from carrying on hiding it. CHECK(listed(*c.client, me)); - CHECK(c->core.configs.user_profile().get_nts_priority() == 0); - CHECK(c->core.configs.user_profile().needs_push()); + in_configs(*c, [](auto& cfg) { + CHECK(cfg.user_profile().get_nts_priority() == 0); + CHECK(cfg.user_profile().needs_push()); + }); } TEST_CASE("Client: revealing note to self keeps a pin it already had", "[client][configs]") { @@ -175,7 +178,7 @@ TEST_CASE("Client: revealing note to self keeps a pin it already had", "[client] c->send_message(me, {.body = "a reminder"}, await); // Already visible, so there is nothing to reveal and the pin is left where the user put it. - CHECK(c->core.configs.user_profile().get_nts_priority() == 7); + CHECK(in_configs(*c, [](auto& cfg) { return cfg.user_profile().get_nts_priority(); }) == 7); CHECK(c->conversation(me, await)->priority() == 7); } @@ -309,7 +312,7 @@ TEST_CASE("Client: blocking someone makes them a contact", "[client][configs]") auto conn = c->core.database().conn(); conn.prepared_exec("INSERT INTO accounts (session_id) VALUES (?)", id.session_id()); } - REQUIRE_FALSE(c->core.configs.contacts().get(them)); + REQUIRE_FALSE(in_configs(*c, [&](auto& cfg) { return cfg.contacts().get(them); })); // Through Client, not through a DM: there is no conversation here, which is exactly the case // that carve-out exists for. @@ -317,14 +320,14 @@ TEST_CASE("Client: blocking someone makes them a contact", "[client][configs]") // The block has to be synced and the entry is the only place it can live, so blocking makes // one. It does not approve them: refusing someone's messages is not accepting them. - auto entry = c->core.configs.contacts().get(them); + auto entry = in_configs(*c, [&](auto& cfg) { return cfg.contacts().get(them); }); REQUIRE(entry); CHECK(entry->blocked); CHECK_FALSE(entry->approved); c->set_blocked(id, false, await); - REQUIRE(c->core.configs.contacts().get(them)); - CHECK_FALSE(c->core.configs.contacts().get(them)->blocked); + REQUIRE(in_configs(*c, [&](auto& cfg) { return cfg.contacts().get(them); })); + CHECK_FALSE(in_configs(*c, [&](auto& cfg) { return cfg.contacts().get(them); })->blocked); } TEST_CASE("Client: clearing a conversation says when it was cleared", "[client][configs]") { @@ -348,7 +351,7 @@ TEST_CASE("Client: clearing a conversation says when it was cleared", "[client][ // And the moment is recorded rather than the deletion being local, so a device that has been // offline through all of this deletes the same messages when it catches up. - auto entry = c->core.configs.contacts().get(them); + auto entry = in_configs(*c, [&](auto& cfg) { return cfg.contacts().get(them); }); REQUIRE(entry); CHECK(entry->delete_before >= before); } @@ -367,7 +370,7 @@ TEST_CASE("Client: deleting a conversation keeps the contact", "[client][configs CHECK_FALSE(listed(*c.client, id)); CHECK(c->conversation(id, await)->messages(await).empty()); - auto entry = c->core.configs.contacts().get(them); + auto entry = in_configs(*c, [&](auto& cfg) { return cfg.contacts().get(them); }); REQUIRE(entry); // Still a contact, so a message from them brings the conversation back. CHECK(entry->approved); CHECK(entry->priority == -1); // The pin it had is not among the things kept. @@ -385,11 +388,12 @@ TEST_CASE("Client: hiding note to self keeps what is in it", "[client][configs]" CHECK_FALSE(listed(*c.client, me)); CHECK(c->conversation(me, await)->messages(await).size() == 1); - CHECK(c->core.configs.user_profile().get_nts_priority() == -1); + CHECK(in_configs(*c, [](auto& cfg) { return cfg.user_profile().get_nts_priority(); }) == -1); // No instruction to destroy anything, which is the whole difference between hiding a // conversation and deleting one. - CHECK(c->core.configs.user_profile().get_nts_delete_before() == std::chrono::sys_seconds{}); + CHECK(in_configs(*c, [](auto& cfg) { return cfg.user_profile().get_nts_delete_before(); }) == + std::chrono::sys_seconds{}); } TEST_CASE("Client: deleting a contact takes the entry that held the block", "[client][configs]") { @@ -412,7 +416,7 @@ TEST_CASE("Client: deleting a contact takes the entry that held the block", "[cl // No entry means no delete-before instruction is owed: another device merging this drops the // conversation and its history because the contact is gone, not because it was told to. It // also means the block is gone, since the entry was the only thing holding it. - CHECK_FALSE(c->core.configs.contacts().get(them)); + CHECK_FALSE(in_configs(*c, [&](auto& cfg) { return cfg.contacts().get(them); })); auto conn = c->core.database().conn(); CHECK(conn.prepared_get("SELECT count(*) FROM messages") == 0); @@ -453,8 +457,8 @@ TEST_CASE("Client: approval is not walked back by a merge", "[client][configs]") auto id = dm_from_hex(them); c->open_dm(id, await); - REQUIRE(c->core.configs.contacts().get(them)); - REQUIRE(c->core.configs.contacts().get(them)->approved); + REQUIRE(in_configs(*c, [&](auto& cfg) { return cfg.contacts().get(them); })); + REQUIRE(in_configs(*c, [&](auto& cfg) { return cfg.contacts().get(them); })->approved); // Another client clearing both flags on its way to deleting the contact, merged without the // deletion that was to follow. Copied verbatim this would file the conversation back under @@ -480,16 +484,19 @@ TEST_CASE("Client: a delete-before is not walked back", "[client][configs]") { // Another device cleared at a moment this one has not reached yet -- clock skew is enough for // that. Publishing our own, smaller value would tell it to un-delete what it destroyed. - auto& contacts = c->core.configs.contacts(); auto later = std::chrono::floor(clock_now_ms()) + 1h; - auto entry = contacts.get_or_construct(them); - entry.delete_before = later; - contacts.set(entry); + in_configs(*c, [&](auto& cfg) { + auto entry = cfg.contacts().get_or_construct(them); + entry.delete_before = later; + cfg.contacts().set(entry); + }); c->conversation(id, await)->clear_messages(await); - REQUIRE(contacts.get(them)); - CHECK(contacts.get(them)->delete_before == later); + in_configs(*c, [&](auto& cfg) { + REQUIRE(cfg.contacts().get(them)); + CHECK(cfg.contacts().get(them)->delete_before == later); + }); } TEST_CASE("Client: our own profile is account state, not a conversation", "[client][configs]") { @@ -502,7 +509,10 @@ TEST_CASE("Client: our own profile is account state, not a conversation", "[clie c->set_display_name("Leia", await); CHECK(c->display_name(await) == "Leia"); - CHECK(c->core.configs.user_profile().get_name() == "Leia"); + // Copied to a string inside the excursion: get_name() hands back a view into the config. + CHECK(in_configs(*c, [](auto& cfg) { + return std::string{cfg.user_profile().get_name().value_or("")}; + }) == "Leia"); // Still no conversation: setting a name is not writing to yourself. CHECK(c->conversations(await).empty()); @@ -516,7 +526,8 @@ TEST_CASE("Client: the save-notification preference follows the account", "[clie c->set_notify_media_saved(false, await); CHECK_FALSE(c->notify_media_saved(await)); - CHECK_FALSE(c->core.configs.user_profile().get_notify_media_saved()); + CHECK_FALSE( + in_configs(*c, [](auto& cfg) { return cfg.user_profile().get_notify_media_saved(); })); // What another device set reaches us through a merge, like any other profile field. auto pushed = profile_from_another_device( diff --git a/tests/test_client/conversation_api.cpp b/tests/test_client/conversation_api.cpp index 04ad48add..28e22e759 100644 --- a/tests/test_client/conversation_api.cpp +++ b/tests/test_client/conversation_api.cpp @@ -34,7 +34,7 @@ TEST_CASE("Client: a conversation reports the settings it carries", "[client][co // All of it reaches the Contacts config, which is what makes it follow the account rather than // the device. - auto entry = c->core.configs.contacts().get(them); + auto entry = in_configs(*c, [&](auto& cfg) { return cfg.contacts().get(them); }); REQUIRE(entry); CHECK(entry->notifications == config::notify_mode::disabled); CHECK(entry->mute_until == 1700000000); @@ -46,7 +46,9 @@ TEST_CASE("Client: a conversation reports the settings it carries", "[client][co // Clearing the nickname falls back to what they call themselves. c->dm(id, await)->set_nickname("", await); CHECK(convo().dm()->nickname.empty()); - CHECK_FALSE(c->core.configs.contacts().get(them)->nickname == "Bilbo"); + CHECK_FALSE( + in_configs(*c, [&](auto& cfg) { return cfg.contacts().get(them); })->nickname == + "Bilbo"); // A timer without a mode expires nothing, so it is not stored as though it were a setting. c->conversation(id, await)->set_expiry(config::expiration_mode::none, 3600s, await); @@ -203,12 +205,16 @@ TEST_CASE("Client: auto-download is per conversation and stays here", "[client][ // Device-local: nothing about it reaches the config that follows the account. Checked by // deriving the contact outward and finding the config unchanged -- if this were synced, the // setting above would have dirtied it. - auto& contacts = c->core.configs.contacts(); + // Read each time rather than binding the config: a reference to one is only ours for as long + // as the excursion onto the loop lasts. + auto contacts_dirty = [&] { + return in_configs(*c, [](auto& cfg) { return cfg.contacts().needs_push(); }); + }; TestHelper::sync_contact(*c.client, id); - auto before_push = contacts.needs_push(); + auto before_push = contacts_dirty(); c->conversation(id, await)->set_auto_download(AutoDownload::all, await); TestHelper::sync_contact(*c.client, id); - CHECK(contacts.needs_push() == before_push); + CHECK(contacts_dirty() == before_push); // And it survives a restart, being a stored property rather than a session's opinion. c.reopen(); diff --git a/tests/test_client/receiving.cpp b/tests/test_client/receiving.cpp index e248d5357..82c744a2b 100644 --- a/tests/test_client/receiving.cpp +++ b/tests/test_client/receiving.cpp @@ -257,7 +257,9 @@ TEST_CASE("Client: non-conversation content does not create a conversation", "[c own_sid(*c), std::nullopt); core::SwarmMessage sm{encoded, std::move(hash), from_epoch_ms(1000), from_epoch_ms(99999)}; - c->core.receive_messages({&sm, 1}, config::Namespace::Default, true); + TestHelper::on_loop(c->core, [&] { + c->core.receive_messages({&sm, 1}, config::Namespace::Default, true); + }); }; // A typing indicator: valid Content, but nothing that belongs in message history. @@ -703,9 +705,8 @@ TEST_CASE("Client: an unsend request from the author deletes the message", "[cli own_sid(*c), std::nullopt); core::SwarmMessage sm{encoded, std::move(hash), from_epoch_ms(9000), from_epoch_ms(1e12)}; - c->core.loop().call_get([&] { + TestHelper::on_loop(c->core, [&] { c->core.receive_messages({&sm, 1}, config::Namespace::Default, true); - return 0; }); }; @@ -843,7 +844,9 @@ TEST_CASE("Client: a sender's picture arrives with their message", "[client][rec } SECTION("and it reaches the config, so our other devices learn it too") { - auto entry = c->core.configs.contacts().get(oxenc::to_hex(sender.session_id)); + auto entry = in_configs(*c, [&](auto& cfg) { + return cfg.contacts().get(oxenc::to_hex(sender.session_id)); + }); REQUIRE(entry); CHECK(entry->profile_picture.url == "http://fs.example/file/7#pubkey=aa"); CHECK(entry->profile_picture.key == key); diff --git a/tests/test_client/requests.cpp b/tests/test_client/requests.cpp index 076d743e8..f5958825a 100644 --- a/tests/test_client/requests.cpp +++ b/tests/test_client/requests.cpp @@ -27,7 +27,9 @@ TEST_CASE("Client: a stranger's message is a request, not a conversation", "[cli // And it is synced, so a request answered on one device is not still waiting on another. Their // writing to us is what says they approved us; nothing yet says we approved them. - auto entry = c->core.configs.contacts().get(oxenc::to_hex(sender.session_id)); + auto entry = in_configs(*c, [&](auto& cfg) { + return cfg.contacts().get(oxenc::to_hex(sender.session_id)); + }); REQUIRE(entry); CHECK(entry->approved_me); CHECK_FALSE(entry->approved); @@ -51,7 +53,9 @@ TEST_CASE("Client: answering a request accepts it", "[client][requests]") { CHECK(c->message_requests(await).empty()); REQUIRE(c->conversations(await).size() == 1); CHECK_FALSE(c->conversations(await)[0].dm()->request); - CHECK(c->core.configs.contacts().get(oxenc::to_hex(sender.session_id))->approved); + CHECK(in_configs(*c, [&](auto& cfg) { + return cfg.contacts().get(oxenc::to_hex(sender.session_id)); + })->approved); // It left one list and joined the other, which is neither an addition nor a removal to either, // so both are replaced. diff --git a/tests/test_client/volatile.cpp b/tests/test_client/volatile.cpp index 97d1723ff..56c159b5f 100644 --- a/tests/test_client/volatile.cpp +++ b/tests/test_client/volatile.cpp @@ -21,7 +21,7 @@ TEST_CASE("Client: reading a conversation publishes the watermark", "[client][vo c->conversation(id, await)->mark_read(await); CHECK(c->conversation(id, await)->unread() == 0); - auto entry = c->core.configs.convo_info_volatile().get_1to1(hex); + auto entry = in_configs(*c, [&](auto& cfg) { return cfg.convo_info_volatile().get_1to1(hex); }); REQUIRE(entry); CHECK(entry->last_read == newest.time_since_epoch().count()); } @@ -77,7 +77,7 @@ TEST_CASE("Client: a stale watermark cannot unread what we have read", "[client] // ...and we do not publish the stale value back out, either. TestHelper::sync_convo_volatile(*c.client, id); - auto entry = c->core.configs.convo_info_volatile().get_1to1(hex); + auto entry = in_configs(*c, [&](auto& cfg) { return cfg.convo_info_volatile().get_1to1(hex); }); REQUIRE(entry); CHECK(entry->last_read == newest.time_since_epoch().count()); } @@ -98,11 +98,13 @@ TEST_CASE("Client: marking unread syncs, and reading clears it", "[client][volat // Survives having read everything, which is the whole point of it. CHECK(c->conversation(id, await)->marked_unread()); CHECK(c->conversation(id, await)->unread() == 0); - CHECK(c->core.configs.convo_info_volatile().get_1to1(hex)->unread); + CHECK(in_configs(*c, [&](auto& cfg) { return cfg.convo_info_volatile().get_1to1(hex); }) + ->unread); c->conversation(id, await)->mark_read(await); CHECK_FALSE(c->conversation(id, await)->marked_unread()); - CHECK_FALSE(c->core.configs.convo_info_volatile().get_1to1(hex)->unread); + CHECK_FALSE(in_configs(*c, [&](auto& cfg) { return cfg.convo_info_volatile().get_1to1(hex); }) + ->unread); } TEST_CASE("Client: read state for a conversation we do not have is ignored", "[client][volatile]") { diff --git a/tests/test_helper.hpp b/tests/test_helper.hpp index b6c7d5fef..08e5c2dd4 100644 --- a/tests/test_helper.hpp +++ b/tests/test_helper.hpp @@ -373,14 +373,18 @@ class TestHelper { /// /// A template so that this header need not know the client types; it is only ever instantiated /// where they are complete. + /// + /// Both hop onto the loop themselves rather than leaving it to the caller: what they reach is + /// a `_`-form on Client, which is what Client's own methods call from inside `_async`, and it + /// touches the configs. template static void sync_contact(Client& c, const Id& id) { - c._sync_contact(id); + on_loop(c.core, [&] { c._sync_contact(id); }); } template static void sync_convo_volatile(Client& c, const Id& id) { - c._sync_convo_volatile(id); + on_loop(c.core, [&] { c._sync_convo_volatile(id); }); } /// The push debounce, driven by hand. A test that waited out real intervals would be both slow