feat(util): add CoalescingWriter for background file/snapshot writes

One background thread for blocking write jobs, for the settings.json / wallets.json saves and the tx-history snapshot that later commits move off the UI thread. Jobs are keyed (a file path, "tx-history"): - a submit for a key that is still queued replaces that job and keeps its place in line (latest wins); - keys run first in, first out, one job at a time, so two writes of one file never overlap and an older snapshot never lands after a newer one; - submitAndWait() blocks until that job (or a newer one for the key) has run and returns its result, for writes the caller needs on disk; - flush() waits with a bound; stop() runs everything queued, then joins; - the thread starts on first use and sleeps when idle; with no thread (none could be created, or after stop()) a job runs inline. persistFileAtomically() routes Platform::writeFileAtomically through a writer, or writes synchronously when there is none. Nothing uses it yet. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
2026-10-09 21:09:48 -05:00
parent cee8b9d7f6
commit 3f1194a84b
4 changed files with 487 additions and 0 deletions
@@ -0,0 +1,190 @@
+// DragonX Wallet - ImGui Edition
+// Copyright 2024-2026 The Hush Developers
+// Released under the GPLv3
+
+#include "coalescing_writer.h"
+#include "logger.h"
+#include "platform.h"
+
+#include <exception>
+#include <utility>
+
+namespace dragonx {
+namespace util {
+
+bool CoalescingWriter::runGuarded(const std::string& key, Work& work)
+{
+    try {
+        return work();
+    } catch (const std::exception& e) {
+        DEBUG_LOGF("[CoalescingWriter] job '%s' threw: %s\n", key.c_str(), e.what());
+    } catch (...) {
+        DEBUG_LOGF("[CoalescingWriter] job '%s' threw\n", key.c_str());
+    }
+    return false;
+}
+
+std::uint64_t CoalescingWriter::enqueueLocked(const std::string& key, Work& work, Work& superseded)
+{
+    // Inline when the worker is gone, or stop() came before any thread existed (a thread started now would
+    // never be joined).
+    if (exited_ || (stop_ && !started_)) return 0;
+    if (!started_) {
+        try {
+            thread_ = std::thread(&CoalescingWriter::run, this);
+            started_ = true;
+        } catch (...) {
+            return 0;   // no thread: run it inline rather than lose it
+        }
+    }
+    const std::uint64_t ticket = ++seq_;
+    auto it = pending_.find(key);
+    if (it != pending_.end()) {
+        superseded = std::exchange(it->second.work, std::move(work));   // latest wins, keeps its place
+        it->second.ticket = ticket;
+        ++superseded_;
+    } else {
+        pending_.emplace(key, Job{std::move(work), ticket});
+        order_.push_back(key);
+    }
+    return ticket;
+}
+
+void CoalescingWriter::recordLocked(const std::string& key, std::uint64_t ticket, bool ok)
+{
+    Done& d = done_[key];
+    if (ticket >= d.ticket) { d.ticket = ticket; d.ok = ok; }
+    failed_[key] = !ok;
+    if (!ok) ++failures_;
+    ++completed_;
+}
+
+bool CoalescingWriter::runInline(const std::string& key, Work& work)
+{
+    // Only reached when no worker can run it: exited_ (every queued job already finished) or no thread was ever
+    // started (nothing queued), so nothing can overtake this write.
+    const bool ok = runGuarded(key, work);
+    work = nullptr;
+    std::lock_guard<std::mutex> lk(mutex_);
+    recordLocked(key, ++seq_, ok);
+    doneCv_.notify_all();
+    return ok;
+}
+
+void CoalescingWriter::submit(const std::string& key, Work work)
+{
+    if (!work) return;
+    Work superseded;   // destroyed after the lock is released (may own large captures / key copies)
+    std::uint64_t ticket = 0;
+    {
+        std::lock_guard<std::mutex> lk(mutex_);
+        ticket = enqueueLocked(key, work, superseded);
+    }
+    if (ticket) { cv_.notify_one(); return; }
+    runInline(key, work);
+}
+
+bool CoalescingWriter::submitAndWait(const std::string& key, Work work)
+{
+    if (!work) return false;
+    Work superseded;
+    std::uint64_t ticket = 0;
+    {
+        std::lock_guard<std::mutex> lk(mutex_);
+        ticket = enqueueLocked(key, work, superseded);
+    }
+    superseded = nullptr;
+    if (!ticket) return runInline(key, work);
+    cv_.notify_one();
+    std::unique_lock<std::mutex> lk(mutex_);
+    doneCv_.wait(lk, [&] {
+        auto d = done_.find(key);
+        return (d != done_.end() && d->second.ticket >= ticket) || exited_;   // run() drains before exiting
+    });
+    auto d = done_.find(key);
+    return d != done_.end() && d->second.ticket >= ticket && d->second.ok;
+}
+
+bool CoalescingWriter::flush(std::chrono::milliseconds timeout)
+{
+    std::unique_lock<std::mutex> lk(mutex_);
+    const bool idle = doneCv_.wait_for(lk, timeout, [this] { return order_.empty() && !running_; });
+    if (!idle)
+        DEBUG_LOGF("[CoalescingWriter] flush timed out with %zu job(s) pending\n",
+                   order_.size() + (running_ ? 1u : 0u));
+    return idle;
+}
+
+void CoalescingWriter::stop()
+{
+    std::thread t;
+    {
+        std::lock_guard<std::mutex> lk(mutex_);
+        stop_ = true;
+        t = std::move(thread_);
+    }
+    cv_.notify_all();
+    if (t.joinable()) t.join();   // run() drains order_ before it exits
+}
+
+bool CoalescingWriter::takeFailure(const std::string& key)
+{
+    std::lock_guard<std::mutex> lk(mutex_);
+    auto it = failed_.find(key);
+    if (it == failed_.end() || !it->second) return false;
+    it->second = false;
+    return true;
+}
+
+std::size_t CoalescingWriter::pendingCount() const
+{
+    std::lock_guard<std::mutex> lk(mutex_);
+    return order_.size() + (running_ ? 1u : 0u);
+}
+std::size_t CoalescingWriter::supersededCount() const { std::lock_guard<std::mutex> lk(mutex_); return superseded_; }
+std::size_t CoalescingWriter::completedCount() const  { std::lock_guard<std::mutex> lk(mutex_); return completed_; }
+std::uint64_t CoalescingWriter::failureCount() const  { std::lock_guard<std::mutex> lk(mutex_); return failures_; }
+
+void CoalescingWriter::run()
+{
+    std::unique_lock<std::mutex> lk(mutex_);
+    for (;;) {
+        cv_.wait(lk, [this] { return !order_.empty() || stop_; });
+        if (order_.empty()) break;   // stop requested and nothing left: decided under the lock
+        std::string key = std::move(order_.front());
+        order_.pop_front();
+        auto it = pending_.find(key);
+        Job job = std::move(it->second);
+        pending_.erase(it);
+        running_ = true;
+        lk.unlock();
+
+        const bool ok = runGuarded(key, job.work);
+        job.work = nullptr;   // release the job's captures (e.g. a SnapshotJob's key copy) before re-locking
+
+        lk.lock();
+        running_ = false;
+        recordLocked(key, job.ticket, ok);
+        doneCv_.notify_all();
+    }
+    exited_ = true;
+    lk.unlock();
+    doneCv_.notify_all();
+}
+
+bool persistFileAtomically(CoalescingWriter* writer, bool waitForDisk, const std::string& path,
+                           std::string content, bool restrictPermissions)
+{
+    if (!writer) return Platform::writeFileAtomically(path, content, restrictPermissions);
+    auto work = [path, content = std::move(content), restrictPermissions]() {
+        const bool ok = Platform::writeFileAtomically(path, content, restrictPermissions);
+        if (!ok) DEBUG_LOGF("[CoalescingWriter] could not write %s\n", path.c_str());
+        return ok;
+    };
+    if (waitForDisk) return writer->submitAndWait(path, std::move(work));
+    writer->submit(path, std::move(work));
+    return true;
+}
+
+} // namespace util
+} // namespace dragonx
@@ -0,0 +1,91 @@
+// DragonX Wallet - ImGui Edition
+// Copyright 2024-2026 The Hush Developers
+// Released under the GPLv3
+//
+// CoalescingWriter - one background thread that runs blocking write jobs (settings.json / wallets.json atomic
+// writes, the tx-history SQLite snapshot) off the UI thread. Jobs are keyed (a file path, "tx-history"): a submit
+// under a key that is still QUEUED replaces that job and keeps its place in line (latest wins); a submit while that
+// key's job is RUNNING queues behind it. Keys run FIFO by first submit, one job at a time, so two writes of one key
+// never overlap and an older snapshot never lands after a newer one. stop() runs everything queued, then joins;
+// afterwards (or if no thread can be created) a submit runs the job inline on the caller, strictly after every
+// queued job finished. Thread-safe (one mutex); in the App every producer is the UI thread. A job must not call
+// into its own writer.
+#pragma once
+
+#include <chrono>
+#include <condition_variable>
+#include <cstddef>
+#include <cstdint>
+#include <deque>
+#include <functional>
+#include <map>
+#include <mutex>
+#include <string>
+#include <thread>
+
+namespace dragonx {
+namespace util {
+
+class CoalescingWriter {
+public:
+    using Work = std::function<bool()>;   // true on success; a throw counts as a failure
+
+    CoalescingWriter() = default;
+    ~CoalescingWriter() { stop(); }
+    CoalescingWriter(const CoalescingWriter&) = delete;
+    CoalescingWriter& operator=(const CoalescingWriter&) = delete;
+
+    // Queue `work` under `key` (the thread starts on first use). Never blocks on I/O.
+    void submit(const std::string& key, Work work);
+    // Queue, then block until this job (or a newer one for `key`) has run; returns that job's result. For writes
+    // whose caller must know they are on disk (Settings' durable saves). Waits for at most the job in flight plus
+    // the queued ones ahead of it.
+    bool submitAndWait(const std::string& key, Work work);
+    // Wait until nothing is queued or running. True if idle within `timeout`.
+    bool flush(std::chrono::milliseconds timeout);
+    // Run everything queued, then join. Idempotent.
+    void stop();
+
+    bool takeFailure(const std::string& key);   // true once if the last completed job for `key` failed
+    std::size_t pendingCount() const;           // queued + running
+    std::size_t supersededCount() const;        // queued jobs replaced by a newer submit
+    std::size_t completedCount() const;
+    std::uint64_t failureCount() const;
+
+private:
+    struct Job  { Work work; std::uint64_t ticket = 0; };
+    struct Done { std::uint64_t ticket = 0; bool ok = false; };
+
+    // Queues `work` (moved from) and returns its ticket, or 0 when it must run inline (no thread / worker gone).
+    std::uint64_t enqueueLocked(const std::string& key, Work& work, Work& superseded);
+    bool runInline(const std::string& key, Work& work);
+    void recordLocked(const std::string& key, std::uint64_t ticket, bool ok);
+    void run();
+    static bool runGuarded(const std::string& key, Work& work);
+
+    mutable std::mutex mutex_;
+    std::condition_variable cv_;        // worker: work queued or stop
+    std::condition_variable doneCv_;    // flush() / submitAndWait(): a job finished, or the worker left
+    std::map<std::string, Job> pending_;
+    std::deque<std::string> order_;     // keys in first-submit order; same key set as pending_
+    std::map<std::string, Done> done_;  // newest finished ticket per key
+    std::map<std::string, bool> failed_;
+    std::thread thread_;
+    std::uint64_t seq_ = 0;
+    bool started_ = false;   // a worker thread was created
+    bool stop_ = false;      // stop() was called
+    bool exited_ = false;    // the worker drained everything and left (decided under mutex_)
+    bool running_ = false;
+    std::size_t superseded_ = 0;
+    std::size_t completed_ = 0;
+    std::uint64_t failures_ = 0;
+};
+
+// The persistence policy for Settings / WalletIndex: no writer -> write synchronously (tests, tools, early
+// startup); otherwise queue the atomic write (latest wins per path), or with waitForDisk queue it and wait for it.
+// Returns the write's result, or true once queued (a failed queued write is logged and counted).
+bool persistFileAtomically(CoalescingWriter* writer, bool waitForDisk, const std::string& path,
+                           std::string content, bool restrictPermissions);
+
+} // namespace util
+} // namespace dragonx