diff --git a/include/session/client.hpp b/include/session/client.hpp index 0d2240b7..fcb9d9a2 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. + call([this, produce = std::move(produce), cb = std::move(cb)]() mutable { using Result = decltype(produce()); try { if constexpr (std::is_void_v) { @@ -1247,18 +1251,50 @@ 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 // 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}; + oxen::quic::JobQueue _jq{core.loop()}; }; } // namespace session::client diff --git a/include/session/client/handler.hpp b/include/session/client/handler.hpp index cdad8cd1..b473f8fb 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 d1ee7862..27a03468 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(); @@ -592,14 +592,56 @@ 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. + /// + /// 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. /// @@ -618,6 +660,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 +683,30 @@ 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 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/component.hpp b/include/session/core/component.hpp index 450c8a95..6a5a9b3f 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 32f5d299..1cadcd6d 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" @@ -37,6 +41,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; @@ -66,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/include/session/core/devices.hpp b/include/session/core/devices.hpp index 15558cd3..3324bd87 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 b0cb6a8e..96c76f0b 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 00000000..b95db442 --- /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 2dfd30b5..5f32f7f8 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) { - loop.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) { - loop.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) { - loop.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) { - loop.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) { - loop.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) { - loop.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,7 +1262,7 @@ 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, + call([this, message_id, index, dest = std::move(dest), @@ -2411,7 +2411,7 @@ void Client::_prefetch_picture(sqlite::Connection& c, int64_t account, const std auto& [sid, key] = *row; - loop.call_soon([this, + call_soon([this, id = ConversationId::dm(sid), url, key = std::vector{key.begin(), key.end()}]() mutable { @@ -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)] { + 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] { + 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 01713242..5936ccd4 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/src/core.cpp b/src/core.cpp index 66615954..f12560d3 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 e5828340..d813bdf6 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 5aa930be..1e00b9fc 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(); @@ -313,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(); @@ -405,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; } @@ -420,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 diff --git a/src/core/devices.cpp b/src/core/devices.cpp index 493d80db..b67c0f35 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 f410431f..ffb27ab3 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_client/attachments.cpp b/tests/test_client/attachments.cpp index fe3d553d..2769108d 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 2c2e4d37..d8db4126 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 5ca98d49..4b114ee5 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 @@ -35,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) { @@ -67,15 +74,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 +115,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) { @@ -125,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) { @@ -136,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 9e9f4331..5e348ac8 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 04ad48ad..28e22e75 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 e248d535..82c744a2 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 076d743e..f5958825 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 97d1723f..56c159b5 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_core_configs.cpp b/tests/test_core_configs.cpp index acf35d95..61ac2ecf 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); + }); } diff --git a/tests/test_core_devices.cpp b/tests/test_core_devices.cpp index 4642b924..00c7f5b7 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 38f750f1..21167f03 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 284095c3..08e5c2dd 100644 --- a/tests/test_helper.hpp +++ b/tests/test_helper.hpp @@ -321,6 +321,32 @@ 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 + /// 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) { + on_loop(core, [] {}); + } + /// 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. @@ -347,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 diff --git a/tests/test_poll.cpp b/tests/test_poll.cpp index 14e125c6..5e7035f5 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);