Merge pull request #97 from frstrtr/btc/v36-livelock-lockyield-watchdog

btc(v36): livelock mirror — lock-yield + think() watchdog (scaffold)

Author: frstrtr

src/impl/btc/node.cpp

@@ -12,6 +12,8 @@
 #include <iomanip>
 #include <random>
 
+#include <execinfo.h>  // backtrace() for think() watchdog stack dump
+
 // Static members for DensePPLNSWindow precomputed decay table
 std::vector<uint64_t> btc::DensePPLNSWindow::s_decay_table;
 uint64_t btc::DensePPLNSWindow::s_decay_per = 0;
@@ -394,6 +396,19 @@ void NodeImpl::processing_shares_phase2(HandleSharesData& data, NetService addr)
     {
         std::unique_lock lock(m_tracker_mutex, std::try_to_lock);
         if (!lock.owns_lock()) {
+            // ── Backpressure (V36 livelock defense-in-depth) ──────────────
+            // If think() is wedged/slow the deferred queue could grow without
+            // bound and blow memory. Cap it: over MAX_PENDING_ADDS we DROP the
+            // new batch (peers re-advertise their best share, so dropped shares
+            // are re-requested later) and warn instead of growing unbounded.
+            if (m_pending_adds.size() >= MAX_PENDING_ADDS) {
+                LOG_WARNING << "[ASYNC-DEFER] BACKPRESSURE: pending_adds at cap ("
+                            << m_pending_adds.size() << "/" << MAX_PENDING_ADDS
+                            << "), dropping batch of " << data.m_items.size()
+                            << " shares from " << addr.to_string()
+                            << " — think() may be wedged";
+                return;
+            }
             LOG_INFO << "[ASYNC-DEFER] processing_shares_phase2: mutex busy, queuing "
                      << data.m_items.size() << " shares from " << addr.to_string()
                      << " (pending=" << m_pending_adds.size() + 1 << ")";
@@ -1313,6 +1328,82 @@ void NodeImpl::prune_shares(const uint256& /*best_share*/)
 
 // (old phases 5-7 removed — replaced by p2pool-style pruning above)
 
+// ── think() watchdog (V36 livelock defense-in-depth) ────────────────────
+// Best-effort recovery if a think() cycle wedges. Runs ENTIRELY on the IO
+// thread and NEVER acquires m_tracker_mutex — so it stays responsive even
+// while the compute thread holds the exclusive lock. Uses an atomic deadline
+// + generation counter rather than a per-dispatch timer object lifetime so a
+// late fire cannot act on a newer cycle.
+void NodeImpl::arm_think_watchdog()
+{
+    if (!m_context)
+        return;
+    auto deadline = std::chrono::steady_clock::now()
+                  + std::chrono::seconds(THINK_WATCHDOG_SECONDS);
+    m_think_deadline_ns.store(
+        std::chrono::duration_cast<std::chrono::nanoseconds>(
+            deadline.time_since_epoch()).count(),
+        std::memory_order_relaxed);
+    uint64_t gen = m_think_generation.fetch_add(1, std::memory_order_relaxed) + 1;
+
+    if (!m_watchdog_timer)
+        m_watchdog_timer = std::make_unique<boost::asio::steady_timer>(*m_context);
+    m_watchdog_timer->expires_after(std::chrono::seconds(THINK_WATCHDOG_SECONDS));
+    m_watchdog_timer->async_wait([this, gen](const boost::system::error_code& ec) {
+        if (ec) return;  // cancelled by disarm (normal completion)
+        // Only act if this is still the cycle we armed for and it has not
+        // already completed (deadline cleared to 0 by disarm).
+        if (m_think_generation.load(std::memory_order_relaxed) != gen)
+            return;
+        int64_t dl = m_think_deadline_ns.load(std::memory_order_relaxed);
+        if (dl == 0)
+            return;  // cycle completed in the gap before this fire
+        auto now_ns = std::chrono::duration_cast<std::chrono::nanoseconds>(
+            std::chrono::steady_clock::now().time_since_epoch()).count();
+        if (now_ns < dl)
+            return;  // not actually overdue (spurious) — let next arm handle it
+
+        // (1) Log + best-effort backtrace dump of the current (IO) thread.
+        // We CANNOT safely unwind the compute thread's stack from here, but
+        // a backtrace of the watchdog firing plus the timing is enough to
+        // confirm the wedge and correlate with the last [ASYNC-THINK] logs.
+        LOG_ERROR << "[THINK-WATCHDOG] think() cycle exceeded "
+                  << THINK_WATCHDOG_SECONDS << "s (gen=" << gen
+                  << ", pending_adds=" << m_pending_adds.size()
+                  << ") — compute thread appears wedged; recovering pipeline";
+        {
+            void* frames[64];
+            int n = ::backtrace(frames, 64);
+            char** syms = ::backtrace_symbols(frames, n);
+            if (syms) {
+                for (int i = 0; i < n; ++i)
+                    LOG_ERROR << "[THINK-WATCHDOG] bt[" << i << "] " << syms[i];
+                ::free(syms);
+            }
+        }
+
+        // (2)+(3) Flag the cycle aborted and reset the running flag so the
+        // pipeline recovers. NOTE: this does NOT forcibly unwind the compute
+        // thread (unsafe); it lets a fresh think() be scheduled. The stuck
+        // think(), if it ever returns, will find m_think_running already false
+        // and its IO-phase still runs harmlessly (idempotent drain + reset).
+        m_think_deadline_ns.store(0, std::memory_order_relaxed);
+        m_think_running.store(false, std::memory_order_relaxed);
+        LOG_WARNING << "[THINK-WATCHDOG] m_think_running reset to false; "
+                       "a new think() cycle may now be scheduled";
+    });
+}
+
+void NodeImpl::disarm_think_watchdog()
+{
+    // Mark cycle complete: clears the deadline so a racing watchdog fire is a
+    // no-op, and cancels the pending timer.
+    m_think_deadline_ns.store(0, std::memory_order_relaxed);
+    if (m_watchdog_timer) {
+        m_watchdog_timer->cancel();
+    }
+}
+
 void NodeImpl::run_think()
 {
     // Skip if a think() is already running on the compute thread.
@@ -1327,6 +1418,10 @@ void NodeImpl::run_think()
              << " pending_adds=" << m_pending_adds.size()
              << " peers=" << m_peers.size();
 
+    // Arm the think() watchdog (IO-thread timer; never touches the tracker
+    // mutex). Disarmed in the IO-phase below on normal completion.
+    arm_think_watchdog();
+
     // Capture block_rel_height fn by value for thread safety
     auto block_rel_height = m_block_rel_height_fn
         ? m_block_rel_height_fn
@@ -1394,7 +1489,12 @@ void NodeImpl::run_think()
         // Publish lock-free snapshot for all IO-thread consumers
         publish_snapshot();
 
-        needs_continue = m_tracker.m_think_needs_continue;
+        // Continue if EITHER the Phase 2 verification budget OR the V36
+        // scoring-walk lock-yield budget was exhausted. Both reuse the same
+        // run_think() re-post: lock released here, drain_pending_adds() runs,
+        // continuation scheduled below.
+        needs_continue = m_tracker.m_think_needs_continue
+                      || m_tracker.m_think_walk_needs_continue;
 
         // Flush verified hashes to LevelDB while lock is held
         flush_verified_to_leveldb();
@@ -1482,6 +1582,10 @@ void NodeImpl::run_think()
                      << " pending batches, peers=" << m_peers.size();
             drain_pending_adds();
 
+            // Cycle completed normally — disarm the watchdog first so it
+            // cannot fire on this (now-finished) generation.
+            disarm_think_watchdog();
+
             // Clear flag AFTER drain — prevents new think() from starting
             // while deferred shares are still being added to the tracker.
             m_think_running.store(false);
@@ -1494,9 +1598,11 @@ void NodeImpl::run_think()
             }
         } catch (const std::exception& e) {
             LOG_ERROR << "run_think() IO phase failed: " << e.what();
+            disarm_think_watchdog();
             m_think_running.store(false);
         } catch (...) {
             LOG_ERROR << "run_think() IO phase failed: unknown error";
+            disarm_think_watchdog();
             m_think_running.store(false);
         }
       });

src/impl/btc/node.hpp

@@ -13,6 +13,8 @@
 #include <sharechain/prepared_list.hpp>
 #include <c2pool/storage/sharechain_storage.hpp>
 
+#include <boost/asio/steady_timer.hpp>
+
 #include <atomic>
 #include <chrono>
 #include <mutex>
@@ -113,6 +115,30 @@ class NodeImpl : public pool::BaseNode<btc::Config, btc::ShareChain, btc::Peer>
     };
     std::vector<PendingShareBatch> m_pending_adds;
 
+    // ── think() watchdog + backpressure (V36 livelock defense-in-depth) ──
+    // If a think() cycle exceeds THINK_WATCHDOG_SECONDS the watchdog logs the
+    // compute-thread state (timing + backtrace dump), flags the cycle, and
+    // resets m_think_running so the pipeline recovers instead of wedging.
+    // Implemented as an atomic deadline checked by an IO-thread steady_timer:
+    // the watchdog NEVER touches m_tracker_mutex (would itself block on the
+    // stuck compute thread). MAX_PENDING_ADDS caps the deferred queue so a
+    // stuck/slow think() cannot grow memory without bound — over the cap new
+    // batches are dropped with a LOG_WARNING backpressure message.
+    static constexpr int THINK_WATCHDOG_SECONDS = 30;
+    static constexpr size_t MAX_PENDING_ADDS = 256;
+    // Steady-clock time-point (ns since epoch) by which the in-flight think()
+    // cycle must complete. 0 == no cycle armed. Set on dispatch, cleared on
+    // completion; read by the watchdog timer on the IO thread.
+    std::atomic<int64_t> m_think_deadline_ns{0};
+    // Generation counter: bumped each dispatch so a fired watchdog only acts
+    // on the cycle it was armed for (guards against late/duplicate fires).
+    std::atomic<uint64_t> m_think_generation{0};
+    // IO-thread timer that polls the deadline. Armed on run_think() dispatch.
+    std::unique_ptr<boost::asio::steady_timer> m_watchdog_timer;
+    // Arm/disarm helpers (defined in node.cpp).
+    void arm_think_watchdog();
+    void disarm_think_watchdog();
+
     // Top-5 scored heads from last think() — used by clean_tracker()
     // to protect the best chains from head pruning (p2pool node.py:363).
     std::vector<uint256> m_last_top5_heads;
@@ -205,8 +231,7 @@ class NodeImpl : public pool::BaseNode<btc::Config, btc::ShareChain, btc::Peer>
             if (on_compute) {
                 ok_ = true;   // exclusive lock already held by caller
             } else {
-                lock_.try_lock();
-                ok_ = lock_.owns_lock();
+                ok_ = lock_.try_lock();
             }
         }
         TrackerReadGuard(TrackerReadGuard&&) = default;

src/impl/btc/share_tracker.hpp

@@ -322,6 +322,16 @@ class ShareTracker
     // Checked by run_think() to schedule a deferred continuation.
     bool m_think_needs_continue{false};
 
+    // V36 livelock mirror (lock-yield): set by think() when the long
+    // decayed-cumulative-weight + PPLNS scoring walk exhausts its
+    // cooperative-yield budget (THINK_WALK_YIELD_BUDGET). It reuses the
+    // SAME run_think() continuation path as m_think_needs_continue so the
+    // compute thread RELEASES m_tracker_mutex, drain_pending_adds() runs,
+    // and a continuation is re-posted — breaking IO-thread starvation when
+    // the scoring window is pathologically large. On healthy windows the
+    // budget never trips (semantics unchanged).
+    bool m_think_walk_needs_continue{false};
+
 private:
     // Retry counter for log throttling only — p2pool retries every think()
     // with no limit.  Counter is cleared on successful verification or
@@ -854,6 +864,23 @@ class ShareTracker
         constexpr int THINK_VERIFY_BUDGET = 100;
         int budget_remaining = bootstrap_mode ? INT_MAX : THINK_VERIFY_BUDGET;
         m_think_needs_continue = false;
+
+        // ── V36 livelock mirror: cooperative-yield budget for the scoring walk ──
+        // The Phase 3/4 scoring step walks the decayed-cumulative-weight + PPLNS
+        // window per scored head while holding m_tracker_mutex.  On a healthy
+        // chain this is cheap, but a pathologically large/forked window can pin
+        // the lock long enough to starve the IO thread (the livelock).  We reuse
+        // the existing needs_continue continuation mechanism: when this budget is
+        // exhausted we stop scoring, release the lock (run_think() does this),
+        // let drain_pending_adds() run, and re-post a continuation.
+        //
+        // The budget is sized FAR above a normal full-window scoring pass so it
+        // NEVER trips on healthy load — behaviour and scoring results are
+        // unchanged for normal-size windows.  It is purely a starvation circuit
+        // breaker for the degenerate case.
+        constexpr int THINK_WALK_YIELD_BUDGET = 1000000;
+        int walk_budget_remaining = bootstrap_mode ? INT_MAX : THINK_WALK_YIELD_BUDGET;
+        m_think_walk_needs_continue = false;
         {
             static int p2_skip_log = 0;
             if (p2_skip_log++ % 20 == 0)
@@ -1063,6 +1090,20 @@ class ShareTracker
         std::vector<DecoratedData<TailScore>> decorated_tails;
         for (auto& [tail_hash, head_hashes] : verified.get_tails())
         {
+            // ── V36 livelock mirror: COARSE walk-yield boundary ──────────────
+            // Conservative boundary: checked once per scored head, NOT inside the
+            // innermost decay iteration (that exact intra-walk yield point is the
+            // marked TODO below, pending ltc-doge PIE symbolization).  When the
+            // scoring walk has consumed its (generous) budget, defer the remaining
+            // tails to a re-posted continuation so the lock is released and
+            // drain_pending_adds() can run.  Trips only on a degenerate window.
+            if (walk_budget_remaining <= 0) {
+                m_think_walk_needs_continue = true;
+                LOG_WARNING << "[think-WALK-YIELD] scoring-walk budget exhausted; "
+                               "deferring remaining tails to continuation";
+                break;
+            }
+
             // p2pool: max(verified.tails[tail_hash], key=verified.get_work)
             uint256 best_head;
             uint288 best_work;
@@ -1085,6 +1126,12 @@ class ShareTracker
                     auto s = score(best_head, block_rel_height_func);
                     s.best_head_work = best_work;  // tiebreak by total work
                     decorated_tails.push_back({s, tail_hash});
+                    // Charge the (coarse) cost of one scored head against the
+                    // walk-yield budget.  score() walks up to CHAIN_LENGTH/16
+                    // window entries; charging the chain_len keeps the accounting
+                    // proportional to real lock-hold time without descending into
+                    // the inner decay loop (see TODO marker there).
+                    walk_budget_remaining -= std::max(s.chain_len, 1);
                 } catch (const std::exception&) {
                     // Chain was concurrently modified (trim removed an
                     // ancestor).  Skip this tail — will be scored next cycle.
@@ -1713,6 +1760,16 @@ class ShareTracker
 
         // Single-pass walk matching p2pool's while loop in
         // get_decayed_cumulative_weights. No pre-collection needed.
+        //
+        // TODO(ltc-doge): pin exact intra-walk yield boundary + optional
+        // zero-divisor guard (degenerate target_ratio==0). This inner decay
+        // iteration is the candidate finest-grained yield point for the V36
+        // livelock lock-yield mechanism; the cooperative budget is currently
+        // enforced at the COARSE per-scored-head boundary in think() Phase 3
+        // (THINK_WALK_YIELD_BUDGET). Once PIE core symbolization pins the true
+        // hot site, move/refine the budget check here. The zero-divisor guard
+        // referenced is the `!this_total.IsNull()` proration guard just below
+        // (remaining / this_total) — confirm it covers the degenerate case.
         auto cur = start;
         while (!cur.IsNull() && chain.contains(cur) && share_count < max_shares)
         {

src/impl/ltc/node.hpp

@@ -216,8 +216,7 @@ class NodeImpl : public pool::BaseNode<ltc::Config, ltc::ShareChain, ltc::Peer>
             if (on_compute) {
                 ok_ = true;   // exclusive lock already held by caller
             } else {
-                lock_.try_lock();
-                ok_ = lock_.owns_lock();
+                ok_ = lock_.try_lock();
             }
         }
         TrackerReadGuard(TrackerReadGuard&&) = default;