Fix two critical stability issues: share verification freeze and header persistence loss

1. Persist verified share status in LevelDB (p2pool known_verified pattern):
   - Add is_verified flag to ShareMetadata with backward-compatible serialization
   - Pre-populate verified tracker on startup, bypassing expensive re-verification
   - Budget think() Phase 2 to 100 shares per call with deferred continuation
   - Flush verified hashes in batches of 50 via mark_shares_verified()

2. Switch header chain to write-back persistence (Litecoin Core FlushStateToDisk pattern):
   - Replace per-header individual put() with dirty set + atomic WriteBatch
   - flush_dirty() writes all dirty headers + tip in single fsync'd batch
   - Prevents header loss on crash (was losing 4.6M headers, keeping only ~8000)
   - Add BatchWriter::commit_sync() for forced fsync

3. Fix on_tip_changed threading: post callback to ioc instead of executing
   on hdr_pool thread, preventing cross-thread access to bcaster/web_server/UTXO

Author: frstrtr

src/c2pool/c2pool_refactored.cpp

@@ -1850,9 +1850,15 @@ int main(int argc, char* argv[]) {
                      chain = embedded_chain.get(),
                      utxo = ltc_utxo_cache.get(),
                      utxo_db = ltc_utxo_db.get(),
-                     &web_server](
+                     &web_server, &ioc](
                         const uint256& old_tip, uint32_t old_height,
                         const uint256& new_tip, uint32_t new_height) {
+                        // This callback fires on the hdr_pool thread.
+                        // Post all work to ioc to avoid cross-thread access to
+                        // bcaster/web_server/UTXO which are ioc-thread objects.
+                        boost::asio::post(ioc,
+                            [tracker, bcaster, chain, utxo, utxo_db, &web_server,
+                             old_tip, old_height, new_tip, new_height]() {
                         bool is_reorg = (new_height <= old_height);
                         LOG_WARNING << "[EMB-LTC] Chain tip changed: "
                                     << old_tip.GetHex().substr(0, 16) << " (h=" << old_height << ") → "
@@ -1919,6 +1925,7 @@ int main(int argc, char* argv[]) {
                         }
                         // Trigger work refresh so stratum miners get the new tip
                         web_server.trigger_work_refresh_debounced();
+                            }); // end post(ioc)
                     });
                 LOG_INFO << "[EMB-LTC] Chain reorg handler registered";
 

src/c2pool/storage/sharechain_storage.cpp

@@ -182,6 +182,26 @@ bool SharechainStorage::remove_share(const uint256& hash)
     return m_leveldb_store->remove_share(hash);
 }
 
+bool SharechainStorage::load_share(const uint256& hash, std::vector<uint8_t>& serialized_data,
+                core::ShareMetadata& metadata)
+{
+    if (!m_leveldb_store)
+        return false;
+    try {
+        return m_leveldb_store->load_share(hash, serialized_data, metadata);
+    } catch (const std::exception& e) {
+        LOG_ERROR << "Error loading share from LevelDB: " << e.what();
+        return false;
+    }
+}
+
+bool SharechainStorage::mark_shares_verified(const std::vector<uint256>& hashes)
+{
+    if (!m_leveldb_store || hashes.empty())
+        return false;
+    return m_leveldb_store->mark_shares_verified(hashes);
+}
+
 void SharechainStorage::log_storage_stats()
 {
     if (!m_leveldb_store) {

src/c2pool/storage/sharechain_storage.hpp

@@ -106,6 +106,10 @@ class SharechainStorage {
     bool load_share(const uint256& hash, std::vector<uint8_t>& serialized_data,
                     uint256& prev_hash, uint64_t& height, uint64_t& timestamp,
                     uint256& work, uint256& target, bool& is_orphan);
+
+    /// Load a share with full metadata (includes is_verified)
+    bool load_share(const uint256& hash, std::vector<uint8_t>& serialized_data,
+                    core::ShareMetadata& metadata);
 
     /**
      * @brief Check if a share exists in storage
@@ -120,7 +124,10 @@ class SharechainStorage {
      * @return True if successfully removed
      */
     bool remove_share(const uint256& hash);
-    
+
+    /// Batch-mark shares as verified in LevelDB metadata.
+    bool mark_shares_verified(const std::vector<uint256>& hashes);
+
     /**
      * @brief Log storage statistics
      */

src/core/leveldb_store.cpp

@@ -302,6 +302,25 @@ bool LevelDBStore::BatchWriter::commit()
     return true;
 }
 
+bool LevelDBStore::BatchWriter::commit_sync()
+{
+    if (!m_store || !m_store->m_db) {
+        LOG_ERROR << "LevelDB store not available for batch commit_sync";
+        return false;
+    }
+
+    leveldb::WriteOptions write_options;
+    write_options.sync = true;  // force fsync
+
+    leveldb::Status status = m_store->m_db->Write(write_options, m_batch.get());
+    if (!status.ok()) {
+        LOG_ERROR << "Failed to commit_sync batch: " << status.ToString();
+        return false;
+    }
+
+    return true;
+}
+
 // SharechainLevelDBStore implementation
 SharechainLevelDBStore::SharechainLevelDBStore(const std::string& base_path, const std::string& network_name)
     : m_base_path(base_path), m_network_name(network_name)
@@ -459,15 +478,16 @@ bool SharechainLevelDBStore::store_share(const uint256& hash, const std::vector<
         metadata_stream << metadata.work;
         metadata_stream << metadata.target;
         metadata_stream << static_cast<uint8_t>(metadata.is_orphan ? 1 : 0);
-        
+        metadata_stream << static_cast<uint8_t>(metadata.is_verified ? 1 : 0);
+
         // Convert from std::vector<std::byte> to std::vector<uint8_t>
         auto span = metadata_stream.get_span();
         std::vector<uint8_t> metadata_data(
             reinterpret_cast<const uint8_t*>(span.data()),
             reinterpret_cast<const uint8_t*>(span.data()) + span.size()
         );
         batch.put(index_key, metadata_data);
-        
+
         // Store height -> hash mapping for chain traversal
         if (metadata.height > 0) {
             std::string height_key = make_height_key(metadata.height);
@@ -515,6 +535,7 @@ bool SharechainLevelDBStore::store_shares_batch(const std::vector<BatchShareEntr
             metadata_stream << e.metadata.work;
             metadata_stream << e.metadata.target;
             metadata_stream << static_cast<uint8_t>(e.metadata.is_orphan ? 1 : 0);
+            metadata_stream << static_cast<uint8_t>(e.metadata.is_verified ? 1 : 0);
             auto span = metadata_stream.get_span();
             std::vector<uint8_t> metadata_data(
                 reinterpret_cast<const uint8_t*>(span.data()),
@@ -586,7 +607,17 @@ bool SharechainLevelDBStore::load_share(const uint256& hash, std::vector<uint8_t
         uint8_t orphan_flag;
         metadata_stream >> orphan_flag;
         metadata.is_orphan = (orphan_flag != 0);
-        
+
+        // is_verified: backward-compatible — old DBs won't have this byte
+        // PackStream clears m_vch when cursor reaches end, so non-empty means more data
+        try {
+            uint8_t verified_flag;
+            metadata_stream >> verified_flag;
+            metadata.is_verified = (verified_flag != 0);
+        } catch (...) {
+            metadata.is_verified = false;
+        }
+
         return true;
 
     } catch (const std::exception& e) {
@@ -647,6 +678,52 @@ bool SharechainLevelDBStore::remove_share(const uint256& hash)
     }
 }
 
+bool SharechainLevelDBStore::mark_shares_verified(const std::vector<uint256>& hashes)
+{
+    if (!m_store || hashes.empty())
+        return false;
+
+    try {
+        auto batch = m_store->create_batch();
+        int updated = 0;
+
+        for (const auto& hash : hashes) {
+            ShareMetadata metadata;
+            std::vector<uint8_t> dummy;
+            if (!load_share(hash, dummy, metadata))
+                continue;
+            if (metadata.is_verified)
+                continue; // already marked
+
+            metadata.is_verified = true;
+
+            PackStream ms;
+            ms << metadata.prev_hash;
+            ms << metadata.height;
+            ms << metadata.timestamp;
+            ms << metadata.work;
+            ms << metadata.target;
+            ms << static_cast<uint8_t>(metadata.is_orphan ? 1 : 0);
+            ms << static_cast<uint8_t>(1); // is_verified = true
+            auto span = ms.get_span();
+            std::vector<uint8_t> md(
+                reinterpret_cast<const uint8_t*>(span.data()),
+                reinterpret_cast<const uint8_t*>(span.data()) + span.size());
+            batch.put(make_index_key(hash), md);
+            ++updated;
+        }
+
+        if (updated > 0 && !batch.commit()) {
+            LOG_ERROR << "mark_shares_verified: batch commit failed";
+            return false;
+        }
+        return true;
+    } catch (const std::exception& e) {
+        LOG_ERROR << "mark_shares_verified failed: " << e.what();
+        return false;
+    }
+}
+
 std::vector<uint256> SharechainLevelDBStore::get_chain_hashes(const uint256& start_hash, uint64_t max_count, bool forward)
 {
     std::vector<uint256> hashes;

src/core/leveldb_store.hpp

@@ -49,6 +49,7 @@ class LevelDBStore
         void put(const std::string& key, const std::vector<uint8_t>& value);
         void remove(const std::string& key);
         bool commit();
+        bool commit_sync();  // force fsync regardless of store options
     };
 
     LevelDBStore(const std::string& db_path, const LevelDBOptions& options);
@@ -84,6 +85,7 @@ struct ShareMetadata {
     uint256 work = uint256::ZERO;
     uint256 target = uint256::ZERO;
     bool is_orphan = false;
+    bool is_verified = false;
 };
 
 /**
@@ -164,6 +166,9 @@ class SharechainLevelDBStore
     bool has_share(const uint256& hash);
     bool remove_share(const uint256& hash);
 
+    /// Batch-update the is_verified flag for multiple shares without rewriting share data.
+    bool mark_shares_verified(const std::vector<uint256>& hashes);
+
     // Chain traversal
     std::vector<uint256> get_chain_hashes(const uint256& start_hash, uint64_t max_count, bool forward = true);
     std::vector<uint256> get_shares_by_height_range(uint64_t start_height, uint64_t end_height);

src/impl/ltc/coin/header_chain.hpp

@@ -22,6 +22,7 @@
 #include <map>
 #include <mutex>
 #include <optional>
+#include <set>
 #include <thread>
 #include <functional>
 #include <iomanip>
@@ -803,36 +804,60 @@ class HeaderChain {
     //   "height"                   → uint32_t
 
     void persist_header(const IndexEntry& entry) {
-        if (!m_db || !m_db->is_open()) {
-            LOG_DEBUG_COIND << "[EMB-LTC] persist_header: no DB (in-memory mode)";
-            return;
+        // Write-back model (matches Litecoin Core's setDirtyBlockIndex):
+        // Mark header as dirty — actual DB write happens in flush_dirty().
+        m_dirty_headers.insert(entry.block_hash);
+    }
+
+    /// Flush all dirty headers + tip to LevelDB in a single atomic WriteBatch
+    /// with sync=true (fsync).  Matches Litecoin Core's FlushStateToDisk() pattern.
+    /// Caller must hold m_mutex.
+    void flush_dirty() {
+        if (!m_db || !m_db->is_open() || m_dirty_headers.empty()) return;
+
+        auto batch = m_db->create_batch();
+        int count = 0;
+
+        for (const auto& hash : m_dirty_headers) {
+            auto it = m_headers.find(hash);
+            if (it == m_headers.end()) continue;
+
+            auto packed = pack(it->second);
+            std::vector<uint8_t> data(
+                reinterpret_cast<const uint8_t*>(packed.data()),
+                reinterpret_cast<const uint8_t*>(packed.data()) + packed.size());
+
+            std::string key = "h";
+            key.append(reinterpret_cast<const char*>(hash.data()), 32);
+            batch.put(key, data);
+            ++count;
         }
 
-        auto packed = pack(entry);
-        std::vector<uint8_t> data(
-            reinterpret_cast<const uint8_t*>(packed.data()),
-            reinterpret_cast<const uint8_t*>(packed.data()) + packed.size());
+        // Include tip in the same atomic batch
+        {
+            std::vector<uint8_t> tip_data(m_tip.data(), m_tip.data() + 32);
+            batch.put("tip", tip_data);
+
+            uint32_t h = m_tip_height;
+            std::vector<uint8_t> height_data(4);
+            height_data[0] = (h >> 24) & 0xFF;
+            height_data[1] = (h >> 16) & 0xFF;
+            height_data[2] = (h >> 8) & 0xFF;
+            height_data[3] = h & 0xFF;
+            batch.put("height", height_data);
+        }
 
-        std::string key = "h";
-        key.append(reinterpret_cast<const char*>(entry.block_hash.data()), 32);
-        m_db->put(key, data);
+        if (batch.commit_sync()) {
+            m_dirty_headers.clear();
+            LOG_DEBUG_COIND << "[EMB-LTC] flush_dirty: wrote " << count << " headers to LevelDB (synced)";
+        } else {
+            LOG_ERROR << "[EMB-LTC] flush_dirty: WriteBatch FAILED for " << count << " headers";
+        }
     }
 
     void persist_tip() {
-        if (!m_db || !m_db->is_open()) return;
-
-        // Store tip hash
-        std::vector<uint8_t> tip_data(m_tip.data(), m_tip.data() + 32);
-        m_db->put("tip", tip_data);
-
-        // Store height
-        uint32_t h = m_tip_height;
-        std::vector<uint8_t> height_data(4);
-        height_data[0] = (h >> 24) & 0xFF;
-        height_data[1] = (h >> 16) & 0xFF;
-        height_data[2] = (h >> 8) & 0xFF;
-        height_data[3] = h & 0xFF;
-        m_db->put("height", height_data);
+        // Write-back: flush all dirty headers + tip atomically.
+        flush_dirty();
     }
 
     void load_from_db() {
@@ -910,6 +935,9 @@ class HeaderChain {
 
     std::unique_ptr<core::LevelDBStore> m_db;
 
+    /// Dirty set: headers modified since last flush (write-back model).
+    std::set<uint256> m_dirty_headers;
+
     /// Callback fired on tip change (reorg / equal-work switch).
     TipChangedCallback m_on_tip_changed;