diff --git a/CMakeLists.txt b/CMakeLists.txt index c70efdd..02b8eea 100644 --- a/CMakeLists.txt +++ b/CMakeLists.txt @@ -559,6 +559,7 @@ set(APP_SOURCES src/data/exchange_info.cpp src/util/logger.cpp src/util/async_task_manager.cpp + src/util/coalescing_writer.cpp src/util/amount_format.cpp src/util/address_validation.cpp src/util/seed_phrase.cpp @@ -705,6 +706,7 @@ set(APP_HEADERS src/util/logger.h src/util/async_task_manager.h src/util/fetch_backoff.h + src/util/coalescing_writer.h src/util/amount_format.h src/util/base64.h src/util/single_instance.h @@ -1213,6 +1215,7 @@ if(BUILD_TESTING) src/resources/embedded_resources.cpp src/util/secure_vault.cpp src/util/platform.cpp + src/util/coalescing_writer.cpp src/util/logger.cpp src/util/lite_server_probe.cpp src/util/pool_registry_core.cpp diff --git a/src/util/coalescing_writer.cpp b/src/util/coalescing_writer.cpp new file mode 100644 index 0000000..e92991e --- /dev/null +++ b/src/util/coalescing_writer.cpp @@ -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 +#include + +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 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 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 lk(mutex_); + ticket = enqueueLocked(key, work, superseded); + } + superseded = nullptr; + if (!ticket) return runInline(key, work); + cv_.notify_one(); + std::unique_lock 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 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 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 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 lk(mutex_); + return order_.size() + (running_ ? 1u : 0u); +} +std::size_t CoalescingWriter::supersededCount() const { std::lock_guard lk(mutex_); return superseded_; } +std::size_t CoalescingWriter::completedCount() const { std::lock_guard lk(mutex_); return completed_; } +std::uint64_t CoalescingWriter::failureCount() const { std::lock_guard lk(mutex_); return failures_; } + +void CoalescingWriter::run() +{ + std::unique_lock 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 diff --git a/src/util/coalescing_writer.h b/src/util/coalescing_writer.h new file mode 100644 index 0000000..e7ce01d --- /dev/null +++ b/src/util/coalescing_writer.h @@ -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 +#include +#include +#include +#include +#include +#include +#include +#include +#include + +namespace dragonx { +namespace util { + +class CoalescingWriter { +public: + using Work = std::function; // 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 pending_; + std::deque order_; // keys in first-submit order; same key set as pending_ + std::map done_; // newest finished ticket per key + std::map 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 diff --git a/tests/test_phase4.cpp b/tests/test_phase4.cpp index 5676779..a5c8877 100644 --- a/tests/test_phase4.cpp +++ b/tests/test_phase4.cpp @@ -18,6 +18,7 @@ #include "util/fetch_backoff.h" #include "util/logger.h" #include "data/transaction_history_cache.h" +#include "util/coalescing_writer.h" #include "data/address_book.h" #include "data/wallet_index.h" #include "util/secure_vault.h" @@ -94,8 +95,11 @@ #include #include #include +#include #include +#include #include +#include #include #include #include @@ -7898,6 +7902,203 @@ void testLiteOfficialServerDetection() EXPECT_FALSE(isOfficialLiteServer("")); } +// CoalescingWriter (SI-1): the shared background writer behind the settings/wallets.json saves and the tx-history +// snapshot. Latest wins per key, distinct keys all run, stop() drains, failures are reported once, a throw is +// contained, and a submit after stop() (or before any thread exists) runs inline. +void testCoalescingWriter() +{ + using dragonx::util::CoalescingWriter; + { // latest wins: hold the worker in job 1, queue three more under the same key, only the last runs + std::mutex m; + std::condition_variable cv; + bool release = false, started = false; + std::vector ran; // written only by the worker; read after flush() (synchronised) + // Declared after the locals its jobs capture by reference: if flush() times out, ~CoalescingWriter joins + // the still-running job before those locals are destroyed. + CoalescingWriter w; + w.submit("k", [&] { + { std::lock_guard l(m); started = true; } + cv.notify_all(); + std::unique_lock l(m); + cv.wait(l, [&] { return release; }); + ran.push_back(1); + return true; + }); + { std::unique_lock l(m); cv.wait(l, [&] { return started; }); } + w.submit("k", [&] { ran.push_back(2); return true; }); + w.submit("k", [&] { ran.push_back(3); return true; }); + w.submit("k", [&] { ran.push_back(4); return true; }); + { std::lock_guard l(m); release = true; } + cv.notify_all(); + EXPECT_TRUE(w.flush(std::chrono::seconds(5))); + EXPECT_EQ(ran.size(), static_cast(2)); + EXPECT_EQ(ran.back(), 4); + EXPECT_EQ(w.supersededCount(), static_cast(2)); + } + { // distinct keys all run; stop() drains; failure reported once; a throw is contained + CoalescingWriter w; + std::atomic n{0}; + w.submit("a", [&] { ++n; return true; }); + w.submit("b", [&] { ++n; return false; }); + w.submit("c", [&]() -> bool { ++n; throw std::runtime_error("boom"); }); + w.stop(); + EXPECT_EQ(n.load(), 3); + EXPECT_FALSE(w.takeFailure("a")); + EXPECT_TRUE(w.takeFailure("b")); + EXPECT_FALSE(w.takeFailure("b")); // cleared + EXPECT_TRUE(w.takeFailure("c")); + bool ranInline = false; // after stop() a late submit runs inline, not dropped + w.submit("late", [&] { ranInline = true; return true; }); + EXPECT_TRUE(ranInline); + w.stop(); // idempotent + } + { // stop() before the first submit: no thread is ever started (nothing left unjoined) + CoalescingWriter w; + w.stop(); + bool ran = false; + w.submit("x", [&] { ran = true; return true; }); + EXPECT_TRUE(ran); + EXPECT_EQ(w.completedCount(), static_cast(1)); + } +} + +// C269: CoalescingWriter ordering, as the settings.json / wallets.json writes use it. Deterministic (jobs are +// gated, no sleeps): latest-wins coalescing, FIFO across keys, submitAndWait ordering + result, bounded flush, +// drain-on-destroy, and real files through persistFileAtomically. +void testCoalescingWriterOrdering() +{ + using dragonx::util::CoalescingWriter; + using ms = std::chrono::milliseconds; + + struct Rec { + std::mutex m; + std::condition_variable cv; + std::vector> calls; + bool hold = false; // jobs block until released + int entered = 0; // jobs that have started + }; + auto r = std::make_shared(); + // A job for key p that "writes" content c: it blocks while held, then records (p, c) and succeeds, or with + // ok=false fails without recording. + auto rec = [r](const std::string& p, const std::string& c, bool ok = true) { + return [r, p, c, ok]() { + std::unique_lock lk(r->m); + ++r->entered; + r->cv.notify_all(); + r->cv.wait(lk, [&] { return !r->hold; }); + if (!ok) return false; + r->calls.emplace_back(p, c); + return true; + }; + }; + auto reset = [r](bool hold) { + std::lock_guard lk(r->m); + r->calls.clear(); r->hold = hold; r->entered = 0; + }; + auto release = [r] { + { std::lock_guard lk(r->m); r->hold = false; } + r->cv.notify_all(); + }; + auto waitEntered = [r](int n) { + std::unique_lock lk(r->m); + return r->cv.wait_for(lk, ms(3000), [&] { return r->entered >= n; }); + }; + using Call = std::pair; + + { // a burst while one write is in flight collapses into ONE follow-up write of the latest content + reset(true); + CoalescingWriter w; + w.submit("a.json", rec("a.json", "v0")); + EXPECT_TRUE(waitEntered(1)); + for (int i = 1; i < 50; ++i) w.submit("a.json", rec("a.json", "v" + std::to_string(i))); + EXPECT_EQ(w.pendingCount(), (size_t)2); // v0 in flight + one coalesced slot + release(); + EXPECT_TRUE(w.flush(ms(3000))); + std::lock_guard lk(r->m); + EXPECT_EQ(r->calls.size(), (size_t)2); + EXPECT_TRUE(r->calls.size() == 2 && r->calls[0] == Call("a.json", "v0") + && r->calls[1] == Call("a.json", "v49")); + } + { // FIFO by first enqueue; a replaced job keeps its place; a re-submit of the in-flight key goes last + reset(true); + CoalescingWriter w; + w.submit("p1", rec("p1", "1a")); + EXPECT_TRUE(waitEntered(1)); + w.submit("p2", rec("p2", "2a")); + w.submit("p3", rec("p3", "3a")); + w.submit("p2", rec("p2", "2b")); + w.submit("p1", rec("p1", "1b")); + release(); + EXPECT_TRUE(w.flush(ms(3000))); + std::lock_guard lk(r->m); + const std::vector want = {{"p1", "1a"}, {"p2", "2b"}, {"p3", "3a"}, {"p1", "1b"}}; + EXPECT_TRUE(r->calls == want); + } + { // submitAndWait: done when it returns, never overtaken by the older queued job + reset(false); + CoalescingWriter w; + w.submit("b.json", rec("b.json", "old")); + EXPECT_TRUE(w.submitAndWait("b.json", rec("b.json", "new"))); + { + std::lock_guard lk(r->m); + EXPECT_TRUE(!r->calls.empty() && r->calls.back() == Call("b.json", "new")); + } + EXPECT_EQ(w.pendingCount(), (size_t)0); + } + { // submitAndWait reports a failed write; a failing key doesn't block others + reset(false); + CoalescingWriter w; + EXPECT_FALSE(w.submitAndWait("c.json", rec("c.json", "x", /*ok=*/false))); + EXPECT_EQ(w.failureCount(), (uint64_t)1); + w.submit("d.json", rec("d.json", "y")); + EXPECT_TRUE(w.flush(ms(3000))); + std::lock_guard lk(r->m); + EXPECT_TRUE(!r->calls.empty() && r->calls.back() == Call("d.json", "y")); + } + { // flush is bounded when a write is wedged, and completes once it unwedges + reset(true); + CoalescingWriter w; + w.submit("e.json", rec("e.json", "z")); + EXPECT_TRUE(waitEntered(1)); + EXPECT_FALSE(w.flush(ms(50))); + release(); + EXPECT_TRUE(w.flush(ms(3000))); + } + { // the destructor runs what is still queued before joining + reset(true); + { + CoalescingWriter w; + w.submit("f.json", rec("f.json", "1")); + EXPECT_TRUE(waitEntered(1)); + w.submit("g.json", rec("g.json", "2")); + release(); + } + std::lock_guard lk(r->m); + EXPECT_TRUE(r->calls.size() == 2 && r->calls[1] == Call("g.json", "2")); + } + + // Real files through persistFileAtomically: atomic, latest wins, no stray .tmp; with no writer it is sync. + { + const fs::path dir = makeTempDir(); + const std::string target = (dir / "n" / "s.json").string(); + { + CoalescingWriter w; + EXPECT_TRUE(dragonx::util::persistFileAtomically(&w, false, target, "{\"k\":1}", true)); + EXPECT_TRUE(dragonx::util::persistFileAtomically(&w, false, target, "{\"k\":2}", true)); + EXPECT_TRUE(w.flush(ms(5000))); + } + std::ifstream in(target); + std::string body((std::istreambuf_iterator(in)), std::istreambuf_iterator()); + EXPECT_EQ(body, std::string("{\"k\":2}")); + EXPECT_FALSE(fs::exists(target + ".tmp")); + const std::string sync = (dir / "sync.json").string(); + EXPECT_TRUE(dragonx::util::persistFileAtomically(nullptr, false, sync, "x", true)); + EXPECT_TRUE(fs::exists(sync)); + in.close(); + removeTree(dir); + } +} + // H7: atomic + durable file writes (settings.json / addressbook.json persistence). void testAtomicFileWrite() { @@ -12404,6 +12605,8 @@ int main() testEffectivePools(); testFormatFeePercent(); testPoolWeightedSelection(); + testCoalescingWriter(); + testCoalescingWriterOrdering(); testAtomicFileWrite(); testHushChatCrypto(); testHushChatReceivePath();