// 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