Give Core's components a threading contract, and enforce it

include/session/client.hpp

@@ -1181,7 +1181,11 @@ class Client {
     // leave them waiting for an answer that is never coming.
     template <typename Produce, typename Cb>
     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<Result>) {
@@ -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 <typename F>
+    void call(F&& f) {
+        _jq.call(std::forward<F>(f));
+    }
+    template <typename F>
+    void call_soon(F&& f) {
+        _jq.call_soon(std::forward<F>(f));
+    }
+    template <typename F>
+    void call_later(std::chrono::microseconds delay, F&& f) {
+        _jq.call_later(delay, std::forward<F>(f));
+    }
+    /// By value, for the same reason as Core's: a reference returned here has escaped the loop.
+    template <typename F>
+    auto call_get(F&& f) {
+        return _jq.call_get(std::forward<F>(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

include/session/client/handler.hpp

@@ -1,29 +1,16 @@
 #pragma once
 
 #include <functional>
-#include <optional>
-#include <string>
+#include <session/handler.hpp>
 
 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 <session/handler.hpp> 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<void(std::function<void()>)>;
 
-namespace detail {
-    template <typename Sig>
-    struct failable_function;
-
-    template <typename... A>
-    struct failable_function<void(A...)> {
-        using type = std::function<void(std::optional<std::string> 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<void(int64_t message_id)>` 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 <typename Sig>
-using failable_function = typename detail::failable_function<Sig>::type;
-
 }  // namespace session::client

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<oxen::quic::Ticker> _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 <typename F>
+    void call(F&& f) {
+        _jq.call(std::forward<F>(f));
+    }
+    template <typename F>
+    void call_soon(F&& f) {
+        _jq.call_soon(std::forward<F>(f));
+    }
+    template <typename F>
+    void call_later(std::chrono::microseconds delay, F&& f) {
+        _jq.call_later(delay, std::forward<F>(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 <typename F>
+    auto call_get(F&& f) {
+        return _jq.call_get(std::forward<F>(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<const SwarmMessage> 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<oxen::quic::Ticker> _poll_ticker;
 };
 
 }  // namespace session::core

include/session/core/component.hpp

@@ -1,10 +1,19 @@
 #pragma once
 
+#include <exception>
+#include <functional>
+#include <optional>
+#include <session/handler.hpp>
+#include <string>
+#include <type_traits>
+#include <utility>
+
 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<void()> 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 <typename Produce, typename Cb>
+        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 <typename Produce, typename Cb>
+    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<Result>) {
+                    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<Result>)
+                    cb(std::string{e.what()});
+                else
+                    cb(std::string{e.what()}, Result{});
+            }
+        });
+    }
+
 }  // namespace detail
 
 }  // namespace session::core