validator.cpp

#include <graphene/protocol/block_vote.hpp>
#include <graphene/plugins/validator/validator.hpp>
#include <graphene/plugins/snapshot/plugin.hpp>

#include <graphene/chain/database_exceptions.hpp>
#include <graphene/chain/account_object.hpp>
#include <graphene/chain/chain_objects.hpp>
#include <graphene/chain/chain_object_types.hpp>
#include <graphene/chain/validator_objects.hpp>
#include <graphene/chain/hardfork.hpp>
#include <graphene/time/time.hpp>

#include <graphene/utilities/key_conversion.hpp>

#include <fc/smart_ref_impl.hpp>

// Boost 1.87 dropped the deprecated Asio interface from <boost/asio.hpp>.
// deadline_timer, which production_timer_ below is, still exists but has to be
// included directly now.
#include <boost/asio/deadline_timer.hpp>

#include <cstdio>
#include <fstream>
#include <limits>
#include <map>
#include <memory>
#include <mutex>
#include <thread>

#include <boost/asio.hpp>
#include <boost/date_time/posix_time/posix_time.hpp>

using std::string;
using std::vector;

using namespace graphene::chain;
using namespace graphene::protocol;

namespace bpo = boost::program_options;

void new_chain_banner(const graphene::chain::database &db) {
    std::cerr << "\n"
            "********************************\n"
            "*                              *\n"
            "*   ------- NEW CHAIN ------   *\n"
            "*   -    Welcome to VIZ!   -   *\n"
            "*   ------------------------   *\n"
            "*                              *\n"
            "********************************\n"
            "\n";
    return;
}

template<typename T>
T dejsonify(const string &s) {
    return fc::json::from_string(s).as<T>();
}

#define DEFAULT_VALUE_VECTOR(value) default_value({fc::json::to_string(value)}, fc::json::to_string(value))
#define LOAD_VALUE_SET(options, name, container, type) \
            if( options.count(name) ) { \
                  const std::vector<std::string>& ops = options[name].as<std::vector<std::string>>(); \
                  std::transform(ops.begin(), ops.end(), std::inserter(container, container.end()), &dejsonify<type>); \
            }

namespace graphene {
    namespace plugins {
        namespace validator_plugin {

            namespace asio = boost::asio;
            namespace posix_time = boost::posix_time;
            namespace system = boost::system;

            struct validator_plugin::impl final {
                uint32_t skip_diag_head = 0;
                const char *skip_diag_why = nullptr;
                impl():
                    p2p_(appbase::app().get_plugin<graphene::plugins::p2p::p2p_plugin>()),
                    chain_(appbase::app().get_plugin<graphene::plugins::chain::plugin>()),
                    snapshot_(appbase::app().get_plugin<graphene::plugins::snapshot::snapshot_plugin>()),
                    production_timer_(production_io_service_) {
                    // Keep production_io_service_ alive while the thread runs,
                    // then start the dedicated thread.  This io_service is intentionally
                    // separate from appbase's shared io_service so that P2P network
                    // activity (peer disconnects, send-queue drains) cannot delay the
                    // production timer callback and cause missed-slot lag.
                    production_io_work_ = std::make_unique<asio::executor_work_guard<asio::io_context::executor_type>>(
                        asio::make_work_guard(production_io_service_));
                    production_io_thread_ = std::thread([this]() { production_io_service_.run(); });
                }

                ~impl() {
                    production_io_service_.stop();
                    if (production_io_thread_.joinable())
                        production_io_thread_.join();
                }

                graphene::chain::database& database() {
                    return chain_.db();
                }

                graphene::chain::database& database() const {
                    return chain_.db();
                }

                graphene::plugins::chain::plugin& chain() {
                    return chain_;
                }

                graphene::plugins::chain::plugin& chain() const {
                    return chain_;
                }

                graphene::plugins::p2p::p2p_plugin& p2p(){
                    return p2p_;
                };

                graphene::plugins::p2p::p2p_plugin& p2p() const {
                    return p2p_;
                };

                graphene::plugins::snapshot::snapshot_plugin& snapshot() {
                    return snapshot_;
                }

                graphene::plugins::snapshot::snapshot_plugin& snapshot() const {
                    return snapshot_;
                }

                graphene::plugins::p2p::p2p_plugin& p2p_;

                graphene::plugins::chain::plugin& chain_;

                graphene::plugins::snapshot::snapshot_plugin& snapshot_;

                void schedule_production_loop();

                block_validation_condition::block_validation_condition_enum block_validation_loop();

                /// Print the node's QC state while the head stays stale. A stall that only
                /// says "Waiting for QC of head" per slot cannot be told apart from one
                /// where the QC is held but no longer verifies, or is for another block.
                void report_if_head_stale();

                // Rate limit of the stalled-head report (see report_if_head_stale).
                fc::time_point _last_stall_report;
                uint32_t _last_stall_report_head = 0;

                block_validation_condition::block_validation_condition_enum maybe_validate_block(fc::mutable_variant_object &capture);

                boost::program_options::variables_map _options;
                uint32_t _required_validator_participation = 33 * CHAIN_1_PERCENT;

                std::atomic<uint64_t> head_block_num_;
                block_id_type head_block_id_ = block_id_type();
                std::atomic<uint64_t> total_hashes_;
                fc::time_point hash_start_time_;

                uint32_t _production_skip_flags = graphene::chain::database::skip_nothing;

                // Dedicated io_service for the production timer — must be declared
                // before production_timer_ so it is initialized first.
                asio::io_context production_io_service_;
                std::unique_ptr<asio::executor_work_guard<asio::io_context::executor_type>> production_io_work_;
                std::thread production_io_thread_;
                asio::deadline_timer production_timer_;

                std::map<public_key_type, fc::ecc::private_key> _private_keys;
                std::set<string> _validators;

                // Signing key of each configured validator as committed at the head
                // block. chainbase's session-visible validator object may already
                // contain a pending validator_update that changes the key, but a block
                // header is validated against the pre-block (committed) key
                // (validate_block_header). Production must therefore sign with the
                // committed key, not with the session-visible one.
                //
                // Seeded at startup and refreshed synchronously at the committed
                // chain boundary, before pending transactions are restored. Reading
                // applied block contents misses rotations reverted by a fork switch;
                // reading the DB in deferred applied_block sees pending state.
                std::map<string, public_key_type> _committed_signing_keys;
                uint64_t _committed_keys_block_num = 0;
                block_id_type _committed_keys_block_id;
                std::mutex _committed_keys_mutex;

                /** Key the next block header must be signed with (the key committed
                 *  at head), given the session-visible value from the validator index. */
                public_key_type effective_signing_key(const string &name, const public_key_type &visible) {
                    if (visible == public_key_type())
                        return visible; // blanked / disabled: production is skipped anyway
                    public_key_type committed;
                    uint64_t snapshot_block = 0;
                    block_id_type snapshot_id;
                    {
                        std::lock_guard<std::mutex> lk(_committed_keys_mutex);
                        auto it = _committed_signing_keys.find(name);
                        if (it == _committed_signing_keys.end() || it->second == visible)
                            return visible;
                        committed = it->second;
                        snapshot_block = _committed_keys_block_num;
                        snapshot_id = _committed_keys_block_id;
                    }
                    // DB access outside the lock: a writer must never wait on it.
                    if (snapshot_block != database().head_block_num() || snapshot_id != database().head_block_id())
                        return visible; // snapshot behind head: trust the chain
                    // Snapshot is for head and differs from the session-visible key:
                    // the difference comes from a pending validator_update. Keep
                    // producing with the committed key; the update lands in the next
                    // block and takes effect for the one after it.
                    return committed;
                }

                /** Called under the DB write lock, with no pending session. */
                void update_committed_signing_keys() {
                    auto &d = database();
                    std::map<string, public_key_type> keys;
                    for (const auto &name : _validators) {
                        const auto *validator = d.find_validator(name);
                        if (validator)
                            keys[name] = validator->signing_key;
                    }
                    const auto head_num = d.head_block_num();
                    const auto head_id = d.head_block_id();
                    std::lock_guard<std::mutex> lk(_committed_keys_mutex);
                    _committed_signing_keys.swap(keys);
                    _committed_keys_block_num = head_num;
                    _committed_keys_block_id = head_id;
                }

                fc::time_point last_block_post_validation_time;

                // ── one post-validation signature per height ──────────────────────────────
                // Block-vote safety rule (Jolteon): a validator votes for a block of round r
                // (round = slot of the block) only when r is above the highest round it has
                // voted in. One vote per round follows, and so does "never vote for an older
                // round after a newer one" -- the property that makes two conflicting QCs
                // need an equivocating quorum. Unlike the earlier one-signature-per-height
                // guard it lets a validator vote for a different block at the same height in
                // a LATER round, which is what lets a split vote resolve instead of stalling.
                // Persisted, so a restart does not forget the round. Re-signing the SAME
                // block of the highest round (the tick re-broadcasts every 250 ms) is allowed.
                struct vote_record { uint32_t round = 0; uint32_t num = 0; block_id_type id; };
                std::string block_vote_state_path_;
                std::map<std::string, vote_record> block_vote_state_;
                // Keyed by block id, not height: a livelock rebuilds one height over and over,
                // and a height key logged only the first variant (stand 22.10, #86 x70 silent).
                std::set<std::pair<std::string, block_id_type>> block_vote_refused_logged_;
                // HF16 diagnostics: why our validator casts no vote for head (once per head).
                std::set<std::pair<std::string, block_id_type>> head_vote_diag_logged_;

                void load_block_vote_state() {
                    if (block_vote_state_path_.empty()) return;
                    std::ifstream in(block_vote_state_path_);
                    std::string name, id;
                    uint32_t round = 0, num = 0;
                    size_t n = 0;
                    // Format: "<validator> <round> <block_num> <block_id>" per line.
                    while (in >> name >> round >> num >> id) {
                        try {
                            block_vote_state_[name] = vote_record{round, num, block_id_type(id)};
                            ++n;
                        } catch (...) {
                            wlog("Skipping malformed block-vote record for ${w} at round ${r}", ("w", name)("r", round));
                        }
                    }
                    if (n) ilog("Restored ${n} block-vote round record(s) from ${p}", ("n", n)("p", block_vote_state_path_));
                }

                bool save_block_vote_state() {
                    if (block_vote_state_path_.empty()) return true;
                    const std::string tmp = block_vote_state_path_ + ".tmp";
                    {
                        std::ofstream out(tmp, std::ios::trunc);
                        if (!out) {
                            elog("Cannot write ${p}; not voting until it is writable", ("p", block_vote_state_path_));
                            return false;
                        }
                        for (const auto &v : block_vote_state_)
                            out << v.first << ' ' << v.second.round << ' ' << v.second.num << ' ' << v.second.id.str() << '\n';
                        out.flush();
                        if (!out) {
                            elog("Failed writing ${p}", ("p", block_vote_state_path_));
                            std::remove(tmp.c_str());
                            return false;
                        }
                    }
                    if (std::rename(tmp.c_str(), block_vote_state_path_.c_str()) != 0) {
                        elog("Failed replacing ${p}", ("p", block_vote_state_path_));
                        std::remove(tmp.c_str());
                        return false;
                    }
                    return true;
                }

                // True when `validator` may vote for `id` (height `num`) in `round`. The
                // record is written BEFORE the signature leaves: a vote whose round could
                // not be persisted is not sent, or a restart could vote that round again.
                bool may_vote(const std::string &validator, uint32_t round, uint32_t num,
                              const block_id_type &id, uint32_t lib) {
                    if (block_vote_refused_logged_.size() > 4096) block_vote_refused_logged_.clear();
                    auto it = block_vote_state_.find(validator);
                    if (it != block_vote_state_.end()) {
                        const auto &last = it->second;
                        if (round == last.round && id == last.id) return true;
                        if (round <= last.round) {
                            // Older rounds are routinely skipped (the list carries recent
                            // blocks), so log only blocks above LIB+1, once per block id.
                            // last.id zero = the round was timed out, not voted.
                            if (num > lib + 1 && block_vote_refused_logged_.insert({validator, id}).second)
                                wlog("Not voting for ${id} at #${n} round ${r} for ${w}: last ${what} round ${lr} (${old})",
                                     ("id", id)("n", num)("r", round)("w", validator)
                                     ("what", last.id == block_id_type() ? "timed-out" : "voted")("lr", last.round)("old", last.id));
                            return false;
                        }
                    }
                    const auto prev = it != block_vote_state_.end() ? fc::optional<vote_record>(it->second) : fc::optional<vote_record>();
                    block_vote_state_[validator] = vote_record{round, num, id};
                    if (!save_block_vote_state()) {
                        if (prev) block_vote_state_[validator] = *prev; else block_vote_state_.erase(validator);
                        return false;
                    }
                    return true;
                }

                // HF16 timeout: `validator` gives up on `round` (head without a QC).
                // Jolteon safety: after timing out round r we never vote for a block of
                // round r, so the round is recorded (block id zero) BEFORE the timeout
                // leaves, same as a vote. Having already voted for the block of round r
                // does not forbid the timeout. False = do not send.
                bool may_timeout(const std::string &validator, uint32_t round) {
                    auto sent = timeout_sent_.find(validator);
                    if (sent != timeout_sent_.end() && sent->second >= round) return false;   // once per round
                    auto it = block_vote_state_.find(validator);
                    if (it != block_vote_state_.end() && it->second.round > round) return false; // moved on
                    if (it == block_vote_state_.end() || it->second.round < round) {
                        const auto prev = it != block_vote_state_.end() ? fc::optional<vote_record>(it->second) : fc::optional<vote_record>();
                        block_vote_state_[validator] = vote_record{round, 0, block_id_type()};
                        if (!save_block_vote_state()) {
                            if (prev) block_vote_state_[validator] = *prev; else block_vote_state_.erase(validator);
                            return false;
                        }
                    }
                    timeout_sent_[validator] = round;
                    return true;
                }
                std::map<std::string, uint32_t> timeout_sent_;   // in memory: a restart may resend, which is harmless
                // Own block votes not yet in a QC are re-gossiped with backoff (1, 2, 4,
                // capped at 8 slots; at most VOTE_RESEND_MAX times). A vote lost on the
                // way (peer dropped, queue lost on reconnect) otherwise never came back:
                // with 15 live of 21 every vote is needed, so one loss froze the height.
                struct vote_resend { uint32_t sent = 0; fc::time_point next; };
                static constexpr uint32_t VOTE_RESEND_MAX = 8;
                std::map<std::pair<std::string, uint32_t>, vote_resend> vote_resend_;

                // Fork collision resolution state
                uint32_t fork_collision_defer_count_ = 0;
                uint32_t _fork_collision_timeout_blocks = 21;  // one full validator round (21 slots: 63 s before HF16, 21 s after)

                // Minority fork recovery state: tracks when we rolled back to
                // LIB and are waiting for P2P sync to catch up before
                // re-enabling block production.
                bool _minority_fork_recovering = false;
                fc::time_point _minority_fork_recovery_start;

                // DLT minority-fork ESCALATION state — what resync_from_lib() alone could
                // not do. Measured 31.10.2026: the detector fired on every 250 ms production
                // tick for an hour, each time "resetting to LIB", while the head never moved.
                // A rewind stays inside the same state, so a state that keeps re-deriving the
                // same fork is rewound into itself forever; the way out is to REPLACE the
                // state with a FRESH snapshot from a trusted peer (our own snapshots are
                // always behind our head, so they replay back into the same state). Bounded:
                // the head must stay put for FORK_ESCALATION_SEC, the download is subject to
                // the usual "must be above max(head, LIB)" floor guard, and each stuck head
                // escalates at most once.
                uint32_t _fork_loop_head = 0;
                fc::time_point _fork_loop_since;
                bool _fork_escalated = false;

                // Ask the snapshot plugin to REPLACE the state with a fresh one from trusted
                // peers. One question, two askers: the slave path below (head stuck after a
                // rewind) and the emergency master (still behind peer LIB after a rewind).
                // Guarded here rather than at each call site: an unregistered std::function
                // throws bad_function_call, and that exception would escape into block
                // processing. Returns whether a recovery could actually be started.
                bool request_forced_snapshot_recovery() {
                    bool recovery_started = false;
                    try {
                        if (chain().snapshot_force_recovery_callback)
                            recovery_started = chain().snapshot_force_recovery_callback();
                    } catch (const fc::exception& e) {
                        elog("Forced snapshot recovery could not be requested: ${e}",
                             ("e", e.to_detail_string()));
                    } catch (...) {
                        elog("Forced snapshot recovery could not be requested: unknown exception");
                    }
                    return recovery_started;
                }

                // Emergency-master peer-LIB gate (see maybe_validate_block). The master is
                // exempt from the isolation scan by design, so this is the only place that
                // can notice the network moved to another chain while we kept producing.
                // Two peers must make the claim before it counts: one peer's LIB can be a
                // stale snapshot from before its own reorg, two agreeing ones cannot be
                // produced by a single node with a key.
                static constexpr uint32_t EMERGENCY_PEER_LIB_QUORUM = 2;
                // A head that keeps coming back after a rewind needs the state REPLACED, not
                // rewound (same reasoning as FORK_ESCALATION_SEC below).
                static constexpr int64_t EMERGENCY_PEER_LIB_ESCALATION_SEC = 60;
                uint32_t _master_gate_head = 0;          // head the quorum was first seen on
                fc::time_point _master_gate_since;       // when that head became stuck
                bool _master_gate_escalated = false;     // one forced snapshot per stuck head
                // Startup witness gate. The peer-LIB gate needs peers to say something, so a
                // master that opens its state and reaches its slot before the first seed has
                // answered sails through on "nobody objects". 03.11.2026 03:05Z, api.viz.world:
                // restarted with LIB #84961415 while the network sat thousands of blocks
                // ahead, produced committee #84961416 four seconds later with zero peers, and
                // forked. Until a seed/trusted witness has reported a LIB that is NOT above
                // ours, the master does not sign — at startup one witness is enough to make
                // it wait (waiting is the safe direction; the quorum of two below guards the
                // opposite case, a running master told to stop). After the grace period a
                // genuinely isolated master may produce again, so a dead seed list cannot
                // halt the chain forever.
                static constexpr int64_t EMERGENCY_STARTUP_WITNESS_GRACE_SEC = 300;
                fc::time_point _startup_time;            // set in plugin_startup()
                bool _startup_cleared = false;           // latched: a witness confirmed we are not behind

                // Operator override (testnet / single-operator fork): when set via
                // 'disable-minority-fork-detection', both the standard and the DLT
                // minority-fork checks are skipped entirely.  On a fork where the
                // operator controls ALL validators, "last 21 blocks all ours" is the
                // normal healthy state, not an isolation symptom — the detector would
                // otherwise loop forever resetting to LIB.  Unlike enable-stale-production,
                // this flag is never auto-cleared by the healthy-participation path.
                bool _disable_minority_fork_detection = false;

                // P18: slot=0 stall detection — tracks consecutive
                // not_time_yet returns to detect NTP/clock issues.
                uint32_t _slot_zero_streak = 0;
                fc::time_point _slot_zero_streak_start;

                // Track consecutive not_my_turn results to detect when our validator
                // is in the schedule but slots keep belonging to other validators.
                // Helps diagnose "silent miss" issues where the validator should have
                // produced but didn't (schedule misalignment, timing issues, etc).
                uint32_t _not_my_turn_streak = 0;
                fc::time_point _not_my_turn_streak_start;
                std::string _last_scheduled_validator; // last validator that got the slot

                // Production watchdog: tracks when we last produced a block
                // so the watchdog can fire if the emergency master goes silent.
                bool _ever_produced = false;
                fc::time_point _last_production_time;
                // Last result from a slot > 0 iteration (not_time_yet filtered out so
                // the watchdog shows a meaningful failure code, not between-slot noise).
                int _last_slot_result = -1;

                // Tracks the scheduled_time of the last lag condition.  Used to skip
                // ahead past a missed slot and avoid spinning in a tight lag loop
                // where every 250ms tick rechecks the same missed slot.
                fc::time_point_sec _last_lag_slot_time;

                // Watchdog debug mode: set to true when the watchdog first fires,
                // enabling verbose DEBUG_CRASH logs automatically to help diagnose
                // why production stopped. Reset to false when production resumes
                // so verbose logging does not persist indefinitely.
                bool _watchdog_debug_enabled = false;

                // Remember the user's --debug-block-production config so the
                // watchdog can restore the original value after auto-enabling
                // debug logging for diagnosis.
                bool _debug_block_production_config = false;

                // Slot hijack detection: counts consecutive blocks where committee
                // filled a slot that was assigned to our validator in the shuffled
                // schedule.  In DLT emergency mode the emergency master may blank
                // our signing key and produce committee blocks in our slots; this
                // counter makes the condition visible in watchdog diagnostics.
                uint32_t _slot_hijack_count = 0;
                uint32_t _slot_hijack_height = 0; // last block height where hijack was detected

                // Track last applied block number to detect missed blocks.
                // Updated in the applied_block signal handler.
                uint64_t _last_applied_block_num = 0;

                boost::signals2::connection _applied_block_connection;
                boost::signals2::scoped_connection _committed_chain_state_connection;

                // Protects cross-thread diagnostic fields shared between
                // production_io_thread_ and the P2P thread (on_block_applied /
                // get_production_diagnostics).  Never held during database() calls.
                std::mutex _diag_mutex;

                // applied_block signal handler: detects when incoming blocks
                // reveal missed slots, and if our validator was scheduled for
                // any of them, dumps full plugin state for diagnosis.
                void on_block_applied(const graphene::chain::signed_block &block);
            };

            void validator_plugin::set_program_options(
                    boost::program_options::options_description &command_line_options,
                    boost::program_options::options_description &config_file_options) {
                    string validator_id_example = "initvalidator";

                command_line_options.add_options()
                        ("enable-stale-production", bpo::value<bool>()->implicit_value(true) , "Enable block production, even if the chain is stale.")
                        ("disable-minority-fork-detection", bpo::value<bool>()->implicit_value(true) , "Disable minority-fork detection (testnet / single-operator fork where this node controls all validators). Never auto-cleared by healthy participation.")
                        ("required-participation", bpo::value<uint32_t>()->default_value(33 * CHAIN_1_PERCENT), "Percent of validators (0-99) that must be participating in order to produce blocks")
                        ("validator,v", bpo::value<vector<string>>()->composing()->multitoken(), ("name of validator controlled by this node (e.g. " + validator_id_example + " )").c_str())
                        ("witness,w", bpo::value<vector<string>>()->composing()->multitoken(), "[DEPRECATED] Use --validator. Name of validator controlled by this node (legacy 'witness' option, kept for config.ini backward compatibility).")
                        ("private-key", bpo::value<vector<string>>()->composing()->multitoken(), "WIF PRIVATE KEY to be used by one or more validators")
                        ("emergency-private-key", bpo::value<vector<string>>()->composing()->multitoken(),
                         "WIF PRIVATE KEY for emergency consensus block production. "
                         "Only used when the network enters emergency consensus mode "
                         "(no blocks for >1 hour since last irreversible block). "
                         "Multiple nodes can safely have this key.")
                        ("ntp-server", bpo::value<vector<string>>()->composing()->multitoken(),
                         "NTP server to use for time synchronization (host or host:port). "
                         "Can be specified multiple times. Leave unset to use the built-in defaults "
                         "(pool.ntp.org, time.google.com, time.cloudflare.com).")
                        ("ntp-request-interval", bpo::value<uint32_t>()->default_value(900),
                         "How often to request a time update from NTP servers, in seconds (default: 900 = 15 min).")
                        ("ntp-retry-interval", bpo::value<uint32_t>()->default_value(300),
                         "Retry interval in seconds when NTP has not replied (default: 300 = 5 min).")
                        ("ntp-round-trip-threshold", bpo::value<uint32_t>()->default_value(150),
                         "Round-trip delay threshold in milliseconds; NTP replies slower than this are discarded (default: 150).")
                        ("ntp-history-size", bpo::value<uint32_t>()->default_value(5),
                         "Moving-average history window size for NTP delta smoothing (default: 5).")
                        ("ntp-rejection-threshold-pct", bpo::value<uint32_t>()->default_value(50),
                         "Rejection threshold as a percentage of the absolute moving average; deltas deviating more are rejected (default: 50).")
                        ("ntp-rejection-min-threshold", bpo::value<uint32_t>()->default_value(5),
                         "Minimum rejection threshold in milliseconds, applied regardless of the percentage rule (default: 5).")
                        ("fork-collision-timeout-blocks", bpo::value<uint32_t>()->default_value(21),
                         "Number of consecutive fork-collision deferrals (block slots) before forcing production. "
                         "One full validator schedule round is 21 blocks (63 seconds). Default: 21.")
                        ("debug-block-production", bpo::value<bool>()->default_value(false),
                         "Enable verbose debug logging for block production and chain internals. Default: false.")
                        ;

                config_file_options.add(command_line_options);
            }

            using std::vector;
            using std::pair;
            using std::string;

            void validator_plugin::plugin_initialize(const boost::program_options::variables_map &options) {
                try {
                    ilog("validator plugin:  plugin_initialize() begin");
                    pimpl = std::make_unique<validator_plugin::impl>();

                    pimpl->total_hashes_.store(0, std::memory_order_relaxed);
                    pimpl->_options = &options;
                    pimpl->block_vote_state_path_ =
                        (appbase::app().data_dir() / "block-vote-round.txt").string();
                    pimpl->load_block_vote_state();
                    LOAD_VALUE_SET(options, "validator", pimpl->_validators, string)
                    if (options.count("witness")) {
                        // Deprecated: --witness / config 'witness' is kept for backward compatibility.
                        wlog("Config option 'witness' is deprecated, use 'validator' instead.");
                        LOAD_VALUE_SET(options, "witness", pimpl->_validators, string)
                    }
                    edump((pimpl->_validators));

                    if(options.count("enable-stale-production")){
                        if (options["enable-stale-production"].as<bool>()) {
                            pimpl->_production_skip_flags |= graphene::chain::database::skip_undo_history_check;
                        }
                    }

                    if(options.count("disable-minority-fork-detection")){
                        if (options["disable-minority-fork-detection"].as<bool>()) {
                            pimpl->_disable_minority_fork_detection = true;
                            wlog("Minority-fork detection DISABLED by operator config. "
                                 "Use only on a testnet / single-operator fork where this node "
                                 "controls all validators.");
                        }
                    }

                    if(options.count("required-participation")){
                        pimpl->_required_validator_participation = options["required-participation"].as<uint32_t>();
                    }

                    if (options.count("private-key")) {
                        const std::vector<std::string> keys = options["private-key"].as<std::vector<std::string>>();
                        for (const std::string &wif_key : keys) {
                            fc::optional<fc::ecc::private_key> private_key = graphene::utilities::wif_to_key(wif_key);
                            FC_ASSERT(private_key.valid(), "unable to parse private key");
                            pimpl->_private_keys[private_key->get_public_key()] = *private_key;
                        }
                    }

                    if (options.count("emergency-private-key")) {
                        const std::vector<std::string> keys = options["emergency-private-key"].as<std::vector<std::string>>();
                        for (const std::string &wif_key : keys) {
                            fc::optional<fc::ecc::private_key> private_key = graphene::utilities::wif_to_key(wif_key);
                            FC_ASSERT(private_key.valid(), "unable to parse emergency private key");
                            pimpl->_private_keys[private_key->get_public_key()] = *private_key;
                        }
                        // Add the committee account to our validator set so we produce blocks
                        // when the schedule assigns committee slots during emergency mode
                        pimpl->_validators.insert(CHAIN_EMERGENCY_VALIDATOR_ACCOUNT);
                        ilog("Emergency private key loaded. Will produce blocks during emergency consensus mode.");
                    }

                    // Build and store NTP configuration so it is applied when the NTP service starts.
                    {
                        graphene::time::ntp_config ntp_cfg;
                        if (options.count("ntp-server"))
                            ntp_cfg.servers = options["ntp-server"].as<std::vector<std::string>>();
                        ntp_cfg.request_interval_sec   = options["ntp-request-interval"].as<uint32_t>();
                        ntp_cfg.retry_interval_sec     = options["ntp-retry-interval"].as<uint32_t>();
                        ntp_cfg.round_trip_threshold_ms = options["ntp-round-trip-threshold"].as<uint32_t>();
                        ntp_cfg.history_size            = options["ntp-history-size"].as<uint32_t>();
                        ntp_cfg.rejection_threshold_pct = options["ntp-rejection-threshold-pct"].as<uint32_t>();
                        ntp_cfg.rejection_min_threshold_ms = options["ntp-rejection-min-threshold"].as<uint32_t>();
                        graphene::time::configure_ntp(ntp_cfg);
                    }

                    if (options.count("fork-collision-timeout-blocks")) {
                        pimpl->_fork_collision_timeout_blocks = options["fork-collision-timeout-blocks"].as<uint32_t>();
                    }

                    if (options.count("debug-block-production")) {
                        pimpl->_debug_block_production_config = options["debug-block-production"].as<bool>();
                        pimpl->chain().db()._debug_block_production = pimpl->_debug_block_production_config;
                        if (pimpl->chain().db()._debug_block_production) {
                            ilog("Debug block production logging ENABLED");
                        }
                    }

                    ilog("validator plugin:  plugin_initialize() end");
                } FC_LOG_AND_RETHROW()
            }

            void validator_plugin::plugin_startup() {
                try {
                    ilog("validator plugin:  plugin_startup() begin");
                    auto &d = pimpl->database();
                    pimpl->_startup_time = fc::time_point::now();
                    //Start NTP time client
                    graphene::time::now();

                    // Force NTP sync before first production tick to minimize the
                    // window where get_slot_at_time() returns 0 due to unsynchronized
                    // NTP on restart.
                    graphene::time::update_ntp_time();

                    // Log validator configuration for post-crash diagnostics
                    ilog("Validator config: ${n} validators, ${k} private keys",
                         ("n", pimpl->_validators.size())("k", pimpl->_private_keys.size()));
                    for (const auto& w : pimpl->_validators) {
                        ilog("  configured validator: ${w}", ("w", w));
                    }

                    // Seed the committed signing-key snapshot from head state. At
                    // startup there is no pending transaction session yet, so the
                    // index read is committed state.
                    for (const auto& w : pimpl->_validators) {
                        try {
                            pimpl->_committed_signing_keys[w] = d.get_validator(w).signing_key;
                        } catch (const fc::exception& e) {
                            wlog("Validator ${w} is not on chain yet; its committed signing key "
                                 "will be learned from committed chain state", ("w", w));
                        }
                    }
                    pimpl->_committed_keys_block_num = d.head_block_num();
                    pimpl->_committed_keys_block_id = d.head_block_id();

                    if (!pimpl->_validators.empty()) {
                        ilog("Launching block production for ${n} validators.", ("n", pimpl->_validators.size()));
                        pimpl->p2p().set_block_production(true);

                        pimpl->_committed_chain_state_connection = d.committed_chain_state.connect([this]() {
                            pimpl->update_committed_signing_keys();
                        });

                        // Connect to applied_block signal to detect missed slots
                        // that belong to our validators and log diagnostic state.
                        pimpl->_last_applied_block_num = d.head_block_num();
                        pimpl->_applied_block_connection = d.applied_block.connect([this](const graphene::chain::signed_block &block) {
                            pimpl->on_block_applied(block);
                        });

                        if (pimpl->_production_skip_flags & graphene::chain::database::skip_undo_history_check) {
                            if (d.head_block_num() == 0) {
                                new_chain_banner(d);
                            }
                        }
                        pimpl->schedule_production_loop();
                    } else
                        elog("No validators configured! Please add validator names and private keys to configuration.");
                    ilog("validator plugin:  plugin_startup() end");
                } FC_CAPTURE_AND_RETHROW()
            }

            void validator_plugin::plugin_shutdown() {
                graphene::time::shutdown_ntp_time();
                if (!pimpl->_validators.empty()) {
                    pimpl->_applied_block_connection.disconnect();
                    pimpl->_committed_chain_state_connection.disconnect();
                    ilog("shutting downing production timer");
                    // Stop the dedicated io_service so the production thread exits.
                    // io_service::stop() is thread-safe; it causes run() to return
                    // after the currently executing handler (if any) completes.
                    pimpl->production_io_service_.stop();
                    if (pimpl->production_io_thread_.joinable())
                        pimpl->production_io_thread_.join();
                }
            }

            void validator_plugin::plugin_for_each_dependency(plugin_processor&& l) {
                l(appbase::app().register_plugin<graphene::plugins::chain::plugin>());
                l(appbase::app().register_plugin<graphene::plugins::p2p::p2p_plugin>());
                l(appbase::app().register_plugin<graphene::plugins::snapshot::snapshot_plugin>());
            }

            validator_plugin::validator_plugin() {}

            validator_plugin::~validator_plugin() {}

            bool validator_plugin::is_validator_scheduled_soon() const {
                try {
                    if (!pimpl || pimpl->_validators.empty() || pimpl->_private_keys.empty()) {
                        return false;
                    }

                    auto& db = pimpl->database();
                    auto op_guard = db.make_operation_guard();
                    fc::time_point now_fine = graphene::time::now();
                    fc::time_point_sec now = now_fine + fc::microseconds(250000);

                    uint32_t slot = db.get_slot_at_time(now);
                    if (slot == 0) {
                        slot = 1;
                    }

                    // Check 5 upcoming slots (~15 seconds) to cover snapshot creation time (~10s) + safety margin
                    for (uint32_t s = slot; s <= slot + 4; ++s) {
                        string scheduled_validator = db.get_scheduled_validator(s);
                        if (pimpl->_validators.find(scheduled_validator) == pimpl->_validators.end()) {
                            continue;
                        }

                        const auto& validator_by_name = db.get_index<graphene::chain::validator_index>().indices().get<graphene::chain::by_name>();
                        auto itr = validator_by_name.find(scheduled_validator);
                        if (itr == validator_by_name.end()) {
                            continue;
                        }

                        graphene::protocol::public_key_type scheduled_key =
                                pimpl->effective_signing_key(scheduled_validator, itr->signing_key);
                        if (scheduled_key == graphene::protocol::public_key_type()) {
                            continue; // Disabled validator (zero key)
                        }

                        if (pimpl->_private_keys.find(scheduled_key) != pimpl->_private_keys.end()) {
                            op_guard.release();
                            return true; // We have the private key and are scheduled soon
                        }
                    }
                } catch (const fc::exception& e) {
                    wlog("is_validator_scheduled_soon check failed: ${e}", ("e", e.to_detail_string()));
                } catch (...) {
                    wlog("is_validator_scheduled_soon check failed with unknown exception");
                }
                return false;
            }

            fc::time_point_sec validator_plugin::get_next_validator_slot_time() const {
                try {
                    if (!pimpl || pimpl->_validators.empty() || pimpl->_private_keys.empty()) {
                        return fc::time_point_sec();
                    }

                    auto& db = pimpl->database();
                    auto op_guard = db.make_operation_guard();
                    fc::time_point now_fine = graphene::time::now();
                    fc::time_point_sec now = now_fine + fc::microseconds(250000);

                    uint32_t slot = db.get_slot_at_time(now);
                    if (slot == 0) {
                        slot = 1;
                    }

                    for (uint32_t s = slot; s <= slot + 4; ++s) {
                        string scheduled_validator = db.get_scheduled_validator(s);
                        if (pimpl->_validators.find(scheduled_validator) == pimpl->_validators.end()) {
                            continue;
                        }

                        const auto& validator_by_name = db.get_index<graphene::chain::validator_index>().indices().get<graphene::chain::by_name>();
                        auto itr = validator_by_name.find(scheduled_validator);
                        if (itr == validator_by_name.end()) {
                            continue;
                        }

                        graphene::protocol::public_key_type scheduled_key =
                                pimpl->effective_signing_key(scheduled_validator, itr->signing_key);
                        if (scheduled_key == graphene::protocol::public_key_type()) {
                            continue; // Disabled validator (zero key)
                        }

                        if (pimpl->_private_keys.find(scheduled_key) != pimpl->_private_keys.end()) {
                            fc::time_point_sec slot_time = db.get_slot_time(s);
                            op_guard.release();
                            return slot_time;
                        }
                    }
                } catch (const fc::exception& e) {
                    wlog("get_next_validator_slot_time check failed: ${e}", ("e", e.to_detail_string()));
                } catch (...) {
                    wlog("get_next_validator_slot_time check failed with unknown exception");
                }
                return fc::time_point_sec();
            }

            bool validator_plugin::is_emergency_master() const {
                try {
                    if (!pimpl || pimpl->_validators.empty()) {
                        return false;
                    }

                    // Condition 1: we hold the emergency-private-key.
                    // CHAIN_EMERGENCY_VALIDATOR_ACCOUNT is added to _validators only
                    // when --emergency-private-key is configured (see plugin_initialize).
                    if (pimpl->_validators.find(CHAIN_EMERGENCY_VALIDATOR_ACCOUNT) == pimpl->_validators.end()) {
                        return false;
                    }

                    // Condition 2: the committee account is in the current validator schedule.
                    auto& db = pimpl->database();
                    return db.with_weak_read_lock([&]() -> bool {
                        const validator_schedule_object& wso = db.get_validator_schedule_object();
                        for (int i = 0; i < wso.num_scheduled_validators; i += CHAIN_BLOCK_VALIDATOR_REPEAT) {
                            if (wso.current_shuffled_validators[i] == CHAIN_EMERGENCY_VALIDATOR_ACCOUNT) {
                                return true;
                            }
                        }
                        return false;
                    });
                } catch (const fc::exception& e) {
                    wlog("is_emergency_master check failed: ${e}", ("e", e.to_detail_string()));
                } catch (...) {
                    wlog("is_emergency_master check failed with unknown exception");
                }
                return false;
            }

            bool validator_plugin::is_emergency_key_configured() const {
                try {
                    if (!pimpl || pimpl->_validators.empty()) {
                        return false;
                    }
                    // CHAIN_EMERGENCY_VALIDATOR_ACCOUNT is added to _validators only
                    // when --emergency-private-key is configured (see plugin_initialize).
                    return pimpl->_validators.find(CHAIN_EMERGENCY_VALIDATOR_ACCOUNT) != pimpl->_validators.end();
                } catch (...) {
                    return false;
                }
            }

            std::string validator_plugin::get_production_diagnostics() const {
                try {
                    if (!pimpl) return "validator=no_pimpl";
                    std::lock_guard<std::mutex> lk(pimpl->_diag_mutex);
                    std::string s = "prod_skip_flags=";
                    s += std::to_string(pimpl->_production_skip_flags);
                    s += " catching_up=";
                    try { s += pimpl->p2p().is_catching_up_after_pause() ? "1" : "0"; } catch (...) { s += "?"; }
                    try { s += " head=#" + std::to_string(pimpl->database().head_block_num()); } catch (...) {}
                    if (pimpl->_ever_produced) {
                        auto ago = (fc::time_point::now() - pimpl->_last_production_time).count() / 1000000;
                        s += " last_prod=" + std::to_string(ago) + "s_ago";
                    } else {
                        s += " last_prod=never";
                    }
                    s += " minority_rcv=";
                    s += pimpl->_minority_fork_recovering ? "1" : "0";
                    s += " slot_hijacks=";
                    s += std::to_string(pimpl->_slot_hijack_count);
                    return "validator[" + s + "]";
                } catch (...) {
                    return "validator=err";
                }
            }

            void validator_plugin::impl::on_block_applied(const graphene::chain::signed_block &block) {
                try {
                    std::lock_guard<std::mutex> _diag_lk(_diag_mutex);
                    uint64_t block_num = block.block_num();
                    uint64_t prev_num = _last_applied_block_num;
                    _last_applied_block_num = block_num;

                    // === SLOT-HIJACK DETECTION (runs for every block) ===
                    // In DLT emergency mode, the emergency master may blank our validator's
                    // signing_key and fill our scheduled slots with committee blocks.  This
                    // detection runs for every incoming block and checks whether the slot
                    // that was just filled was assigned to our validator but produced by someone
                    // else (committee / emergency account).  The counter is included in
                    // watchdog diagnostics so the operator can see the hijack pattern.
                    const auto& dgp_hijack = database().get_dynamic_global_properties();
                    if (database()._dlt_mode && !_validators.empty()
                        && prev_num > 0 && block_num == prev_num + 1
                        && dgp_hijack.emergency_consensus_active) {
                        const auto& wso_sj = database().get_validator_schedule_object();
                        uint32_t nsw_sj = wso_sj.num_scheduled_validators;
                        if (nsw_sj > 0) {
                            // apply_block increments current_aslot to the applied block's slot
                            // before this callback fires, so current_aslot IS the applied slot.
                            uint64_t slot_idx = dgp_hijack.current_aslot % nsw_sj;
                            const std::string& expected_validator =
                                wso_sj.current_shuffled_validators[slot_idx];
                            bool was_our_slot = _validators.count(expected_validator) > 0;
                            // True hijack only if the actual producer is NOT also one of our validators.
                            bool producer_is_ours = _validators.count(block.validator) > 0;

                            if (was_our_slot && !producer_is_ours && block.validator != expected_validator) {
                                // External validator (committee / emergency) produced at our slot.
                                _slot_hijack_count++;
                                _slot_hijack_height = static_cast<uint32_t>(block_num);
                                // Log the first 3 hijacks, then once per minute.
                                static fc::time_point _last_hijack_log;
                                auto _now_sj = fc::time_point::now();
                                if (_slot_hijack_count <= 3 ||
                                    (_now_sj - _last_hijack_log).count() > 60000000) {
                                    _last_hijack_log = _now_sj;
                                    elog("SLOT-HIJACK: block #${bn} by '${wit}' but slot was assigned "
                                         "to our validator '${exp}' (hijack #${cnt}). "
                                         "head=#${head} aslot=${aslot} num_sched=${nsched}",
                                         ("bn", block_num)("wit", block.validator)("exp", expected_validator)
                                         ("cnt", _slot_hijack_count)
                                         ("head", dgp_hijack.head_block_number)
                                         ("aslot", (uint64_t)dgp_hijack.current_aslot)
                                         ("nsched", nsw_sj));
                                }
                            } else if (was_our_slot && (block.validator == expected_validator || producer_is_ours)) {
                                // Our validator (expected or another of ours) produced — reset hijack counter.
                                if (_slot_hijack_count > 0) {
                                    wlog("SLOT-HIJACK-RESOLVED: our validator '${wit}' produced "
                                         "block #${bn} after ${cnt} hijacked slot(s).",
                                         ("wit", block.validator)("bn", block_num)("cnt", _slot_hijack_count));
                                }
                                _slot_hijack_count = 0;
                            }
                        }
                    }

                    // No gap, first block, or emergency mode not active — nothing to check
                    if (prev_num == 0 || block_num <= prev_num + 1) {
                        return;
                    }

                    uint32_t missed_count = static_cast<uint32_t>(block_num - prev_num - 1);

                    // By the time applied_block fires, current_aslot has already been
                    // updated: new_aslot = old_aslot + missed_count + 1.
                    // The missed slots had absolute slot indices:
                    //   old_aslot + 1, old_aslot + 2, ..., old_aslot + missed_count
                    // So: old_aslot = current_aslot - missed_count - 1
                    // Missed slot i (0-based): abs_slot = current_aslot - missed_count + i
                    const auto &dgp = database().get_dynamic_global_properties();
                    const auto &wso = database().get_validator_schedule_object();
                    uint64_t cur_aslot = dgp.current_aslot;
                    uint32_t num_validators = wso.num_scheduled_validators;
                    if (num_validators == 0) return;

                    // Check each missed slot to see if our validator was scheduled
                    bool our_validator_missed = false;
                    std::string missed_validators_list;
                    for (uint32_t i = 0; i < missed_count && i < 100; ++i) {
                        uint64_t abs_slot = cur_aslot - missed_count + i;
                        const std::string &wname = wso.current_shuffled_validators[abs_slot % num_validators];
                        if (!missed_validators_list.empty()) missed_validators_list += ",";
                        missed_validators_list += wname;
                        if (_validators.count(wname) > 0) {
                            our_validator_missed = true;
                        }
                    }

                    if (!our_validator_missed) {
                        // Not our problem — other validators missed their slots
                        return;
                    }

                    // Our validator missed a slot! Dump full diagnostic state.
                    fc::time_point_sec now_ts = graphene::time::now();
                    int64_t ntp_us = 0;
                    try { ntp_us = graphene::time::ntp_error().count(); } catch (...) {}

                    bool catching_up = false;
                    try { catching_up = p2p().is_catching_up_after_pause(); } catch (...) {}
                    bool dlt_syncing = false;
                    try { dlt_syncing = chain().is_syncing(); } catch (...) {}

                    std::string validator_names;
                    for (const auto &w : _validators) {
                        if (!validator_names.empty()) validator_names += ",";
                        validator_names += w;
                    }

                    fc::time_point_sec next_slot = database().get_slot_time(1);
                    std::string next_scheduled = database().get_scheduled_validator(1);

                    // Check on-chain signing key status for our validators
                    std::string key_status;
                    const auto &wit_idx = database().get_index<graphene::chain::validator_index>()
                        .indices().get<graphene::chain::by_name>();
                    for (const auto &wname : _validators) {
                        auto witr = wit_idx.find(wname);
                        if (!key_status.empty()) key_status += " ";
                        if (witr != wit_idx.end()) {
                            bool key_blank = (witr->signing_key == graphene::protocol::public_key_type());
                            key_status += wname + ":key=" + (key_blank ? "BLANK" : "ok")
                                + ":last_conf=" + std::to_string(witr->last_confirmed_block_num);
                        } else {
                            key_status += wname + ":NOT_FOUND";
                        }
                    }

                    elog("MISSED-SLOT-OUR-VALIDATOR: block #${bn} arrived, ${missed} slot(s) missed between #${prev} and #${bn}. "
                         "Missed validators: [${mw}]. OUR validator was scheduled! "
                         "State: ever_produced=${ep} minority_recovering=${mr} "
                         "last_slot_result=${sr} not_my_turn_streak=${nmts} slot0_streak=${sz} "
                         "dlt_syncing=${ds} catching_up=${cu} head=#${h} aslot=${a} num_sched=${ns} "
                         "ntp_offset=${ntp}us now=${now} next_slot_time=${nst} next_scheduled=${nsw} "
                         "validators=[${wn}] keys=[${ks}]",
                         ("bn", block_num)("missed", missed_count)("prev", prev_num)
                         ("mw", missed_validators_list)
                         ("ep", _ever_produced)
                         ("mr", _minority_fork_recovering)
                         ("sr", _last_slot_result)("nmts", _not_my_turn_streak)
                         ("sz", _slot_zero_streak)
                         ("ds", dlt_syncing)("cu", catching_up)
                         ("h", dgp.head_block_number)("a", cur_aslot)
                         ("ns", num_validators)
                         ("ntp", ntp_us)("now", now_ts)
                         ("nst", next_slot)("nsw", next_scheduled)
                         ("wn", validator_names)("ks", key_status));
                } catch (...) {
                    // Non-critical diagnostic — never disrupt block processing
                }
            }

            void validator_plugin::impl::schedule_production_loop() {
                //Schedule for the next 250ms tick regardless of chain state
                // With +250ms look-ahead in maybe_validate_block(), the tick at
                // T_slot - 250ms aligns now exactly to the slot boundary for zero-lag production.
                // If we would wait less than 50ms, wait for the whole 250ms period.
                int64_t ntp_microseconds = graphene::time::now().time_since_epoch().count();
                int64_t next_microseconds = 250000 - ( ntp_microseconds % 250000 );
                if (next_microseconds < 50000) { // we must sleep for at least 50ms
                    next_microseconds += 250000 ;
                }

                // Sanity check: in normal operation next_microseconds is always ≤500ms.
                // A larger value means NTP time jumped backward, which delays the loop
                // and can cause missed slots.
                if (next_microseconds > 500000) {
                    int64_t ntp_us = 0;
                    try { ntp_us = graphene::time::ntp_error().count(); } catch (...) {}
                    wlog("SCHEDULE WARNING: production loop sleeping ${d}ms (expected ≤500ms). "
                         "NTP may have jumped backward. ntp_offset=${n}us",
                         ("d", next_microseconds / 1000)("n", ntp_us));
                }
                production_timer_.expires_from_now( posix_time::microseconds(next_microseconds) );
                production_timer_.async_wait( [this](const system::error_code &) { block_validation_loop(); } );
            }

            void validator_plugin::impl::report_if_head_stale() {
                auto &db = database();
                fc::optional<std::string> report;
                uint32_t head_num = 0;
                uint32_t head_age = 0;
                try {
                    db.with_weak_read_lock([&]() {
                        head_num = db.head_block_num();
                        // Three rounds behind is already a stall: a block is due every round
                        // and the head's QC should have formed inside the next one. Reporting
                        // from there (and at most every 10 s per head) keeps a long stall to
                        // six lines a minute instead of one per slot.
                        const uint32_t now_s = fc::time_point_sec(fc::time_point::now()).sec_since_epoch();
                        const uint32_t head_s = db.head_block_time().sec_since_epoch();
                        // A head stamped in the future is clock skew, not a stall: the
                        // unsigned difference would otherwise wrap into a huge "age".
                        if (head_num == 0 || now_s <= head_s + 3 * db.block_interval()) return;
                        head_age = now_s - head_s;
                        if (head_num == _last_stall_report_head &&
                            (fc::time_point::now() - _last_stall_report).count() < 10 * 1000000) return;
                        _last_stall_report = fc::time_point::now();
                        _last_stall_report_head = head_num;
                        report = db.qc_state_report();
                    });
                } catch (const fc::exception &e) {
                    wlog("Stalled-head report failed: ${e}", ("e", e.to_string()));
                    return;
                } catch (const std::exception &e) {
                    wlog("Stalled-head report failed: ${e}", ("e", std::string(e.what())));
                    return;
                } catch (...) {
                    // The report is called from the production timer callback: an exception
                    // escaping it would take the node down, and a diagnostic must never be
                    // able to do that.
                    wlog("Stalled-head report failed: unknown exception");
                    return;
                }
                if (report)
                    wlog("Chain stalled ${s}s at #${h}: ${r}", ("s", head_age)("h", head_num)("r", *report));
            }

            block_validation_condition::block_validation_condition_enum validator_plugin::impl::block_validation_loop() {
                block_validation_condition::block_validation_condition_enum result;
                fc::mutable_variant_object capture;
                if (database()._debug_block_production) ilog("DEBUG_CRASH: block_validation_loop ENTER");
                report_if_head_stale();
                try {
                    result = maybe_validate_block(capture);
                }
                catch (const fc::canceled_exception &) {
                    //We're trying to exit. Go ahead and let this one out.
                    throw;
                }
                catch (const graphene::chain::unknown_hardfork_exception &e) {
                    // Hit a hardfork that the current node know nothing about, stop production and inform user
                    elog("${e}\nNode may be out of date...", ("e", e.to_detail_string()));
                    throw;
                }
                catch (const fc::exception &e) {
                    elog("Got exception while generating block:\n${e}", ("e", e.to_detail_string()));
                    result = block_validation_condition::exception_validating_block;
                }
                catch (const std::exception &e) {
                    // Non-fc exception (e.g. boost::wrapexcept<std::runtime_error>
                    // thrown by chainbase when read-lock acquisition times out
                    // under heavy write-lock contention).  Treat it like any
                    // other transient validation failure: log and reschedule.
                    // Without this handler the exception would escape the asio
                    // timer callback and call std::terminate (Windows DLT-sync
                    // crash near the emergency-mode boundary, where
                    // update_global_dynamic_data emits many per-slot
                    // "Missed block" logs while holding the write lock).
                    elog("Got std::exception while generating block: ${e}",
                         ("e", std::string(e.what())));
                    result = block_validation_condition::exception_validating_block;
                }
                catch (...) {
                    // Final guard: never let an unknown exception type escape
                    // into the asio callback chain.
                    elog("Got unknown exception while generating block");
                    result = block_validation_condition::exception_validating_block;
                }

                if (database()._debug_block_production) ilog("DEBUG_CRASH: maybe_validate_block returned ${r}", ("r", (int)result));
                if (result != block_validation_condition::not_time_yet) {
                    std::lock_guard<std::mutex> lk(_diag_mutex);
                    _last_slot_result = (int)result;
                }
                switch (result) {
                    case block_validation_condition::produced:
                        ilog("\033[92mValidated block #${n} with timestamp ${t} at time ${c} by \033[93m${w}\033[92m with ${tx} transactions\033[0m", (capture));
                        fork_collision_defer_count_ = 0;
                        // _watchdog_debug_enabled is production-only — no lock needed.
                        if (_watchdog_debug_enabled) {
                            _watchdog_debug_enabled = false;
                            database()._debug_block_production = _debug_block_production_config;
                        }
                        {
                            // Snapshot minority-fork state before locking so we can log outside the lock.
                            bool _was_recovering; fc::microseconds _rcv_elapsed;
                            {
                                std::lock_guard<std::mutex> lk(_diag_mutex);
                                _slot_zero_streak = 0;
                                _not_my_turn_streak = 0;
                                _slot_hijack_count = 0;
                                _ever_produced = true;
                                _last_production_time = fc::time_point::now();
                                _was_recovering = _minority_fork_recovering;
                                if (_was_recovering) {
                                    _rcv_elapsed = fc::time_point::now() - _minority_fork_recovery_start;
                                    _minority_fork_recovering = false;
                                }
                            }
                            if (_was_recovering) {
                                ilog("MINORITY FORK RECOVERY COMPLETE: production resumed after ${e}s",
                                     ("e", _rcv_elapsed.count() / 1000000));
                            }
                        }
                        break;
                    case block_validation_condition::not_synced:
                        if (_minority_fork_recovering) {
                            auto elapsed = fc::time_point::now() - _minority_fork_recovery_start;
                            if (elapsed.count() % 5000000 < 300000) { // log every ~5 seconds
                                auto &rdb = database();
                                ilog("MINORITY FORK RECOVERY: waiting for P2P sync (head=#${h}, "
                                     "slot_time=${st}, now=${now}, elapsed=${e}s)",
                                     ("h", rdb.head_block_num())
                                     ("st", rdb.get_slot_time(1))
                                     ("now", graphene::time::now())
                                     ("e", elapsed.count() / 1000000));
                            }
                        } else {
                            static fc::time_point _last_not_synced_log;
                            auto _now_ns = fc::time_point::now();
                            if ((_now_ns - _last_not_synced_log).count() > 10000000) {
                                _last_not_synced_log = _now_ns;
                                wlog("Block production deferred: not_synced (head=#${h}, catching_up=${c})",
                                     ("h", database().head_block_num())
                                     ("c", p2p().is_catching_up_after_pause()));
                            }
                        }
                        fork_collision_defer_count_ = 0;
                        {
                            std::lock_guard<std::mutex> lk(_diag_mutex);
                            _slot_zero_streak = 0;
                            _not_my_turn_streak = 0;
                        }
                        break;
                    case block_validation_condition::not_my_turn:
                        // This log-record is commented, because it outputs very often
                        // ilog("Not producing block because it isn't my turn");
                        fork_collision_defer_count_ = 0;
                        // Emergency master: the EMRG-DIAG log in maybe_validate_block fires
                        // per-slot details; nothing extra needed here.

                        // Track consecutive not_my_turn to detect schedule misalignment.
                        {
                            uint32_t _nmt_snap;
                            {
                                std::lock_guard<std::mutex> lk(_diag_mutex);
                                _slot_zero_streak = 0;
                                _not_my_turn_streak++;
                                if (_not_my_turn_streak == 1)
                                    _not_my_turn_streak_start = fc::time_point::now();
                                _nmt_snap = _not_my_turn_streak;
                            }
                        if (_nmt_snap == 500) {
                            // ~125s with slot>0 but not our validator — investigate.
                            // _not_my_turn_streak_start is production-only; no lock needed.
                            auto elapsed = fc::time_point::now() - _not_my_turn_streak_start;
                            wlog("NOT_MY_TURN STREAK: ${n} consecutive slots (${s}s) went to other validators. "
                                 "Last scheduled: ${sw}. Our validators: ${ow}. "
                                 "Check: is our validator still in shuffled schedule?",
                                 ("n", _nmt_snap)("s", elapsed.count() / 1000000)
                                 ("sw", _last_scheduled_validator)
                                 ("ow", [_validators = _validators]() {
                                     std::string s;
                                     for (const auto& w : _validators) { if (!s.empty()) s += ","; s += w; }
                                     return s;
                                 }()));
                        }
                        } // not_my_turn scope
                        break;
                    case block_validation_condition::not_time_yet:
                        // This log-record is commented, because it outputs very often
                        // ilog("Not producing block because slot has not yet arrived");
                        // P18 fix: Detect slot=0 stall caused by NTP drift.
                        // Only count as a stall when now <= head_block_time (NTP time
                        // has fallen behind chain time).  When now > head_block_time
                        // we are simply between slots — this is normal and should NOT
                        // increment the streak counter, otherwise every 2.5s of normal
                        // waiting triggers a spurious NTP resync.
                        {
                            auto _now = graphene::time::now();
                            auto _hbt = database().head_block_time(); // read before taking lock
                            std::lock_guard<std::mutex> lk(_diag_mutex);
                            if (_now <= fc::time_point(_hbt)) {
                                // Real stall: NTP time is behind chain time
                                _slot_zero_streak++;
                            } else {
                                // Normal: just waiting for next slot
                                _slot_zero_streak = 0;
                            }
                        }
                        if (_slot_zero_streak == 1) {
                            _slot_zero_streak_start = fc::time_point::now();
                        }
                        // Threshold 3 (~750ms): first logged warning. Threshold 1 is too
                        // noisy — blocks naturally arrive with timestamps slightly ahead
                        // of our NTP clock, triggering a false streak-start every ~3s.
                        // At threshold 3, the gap is large enough to be worth investigating.
                        if (_slot_zero_streak == 3) {
                            auto _now_init = graphene::time::now();
                            auto _hbt_init = database().head_block_time();
                            auto _nst_init = database().get_slot_time(1);
                            int64_t _drift_us = 0;
                            try { _drift_us = graphene::time::ntp_error().count(); } catch (...) {}
                            int64_t _gap_ms = (_nst_init - _now_init).count() / 1000;
                            std::string _next_w3 = database().get_scheduled_validator(1);
                            bool _ours3 = _validators.count(_next_w3) > 0;
                            wlog("SLOT=0 STREAK: ${n} consecutive not_time_yet. "
                                 "now=${now} head_block_time=${hbt} (drift=${d}us) "
                                 "next_slot=${nst} (gap=${g}ms) next_validator=${nw} is_ours=${o} head=#${h}.",
                                 ("n", _slot_zero_streak)("now", _now_init)("hbt", _hbt_init)
                                 ("d", _drift_us)("nst", _nst_init)("g", _gap_ms)
                                 ("nw", _next_w3)("o", _ours3)
                                 ("h", database().head_block_num()));
                        }
                        if (_slot_zero_streak == 10) {
                            // ~2.5s at 250ms schedule interval — NTP drift detected
                            auto _now10 = graphene::time::now();
                            auto _hbt10 = database().head_block_time();
                            auto _nst10 = database().get_slot_time(1);
                            int64_t _drift10 = 0;
                            try { _drift10 = graphene::time::ntp_error().count(); } catch (...) {}
                            int64_t _gap_ms = (_nst10 - _now10).count() / 1000;
                            std::string _next_w10 = database().get_scheduled_validator(1);
                            bool _ours10 = _validators.count(_next_w10) > 0;
                            wlog("slot=0 streak: ${n} consecutive not_time_yet (${s}s elapsed). "
                                 "now=${now}, head_block_time=${hbt}, next_slot_time=${nst}. "
                                 "Time to next slot: ${g}ms. NTP drift: ${d}us. "
                                 "head_age=${ha}ms. next_validator=${nw} is_ours=${o}. Forcing NTP resync.",
                                 ("n", _slot_zero_streak)("s", (_now10 - _slot_zero_streak_start).count() / 1000000)
                                 ("now", _now10)("hbt", _hbt10)("nst", _nst10)("g", _gap_ms)
                                 ("d", _drift10)("ha", (_now10 - fc::time_point(_hbt10)).count() / 1000)
                                 ("nw", _next_w10)("o", _ours10));
                            graphene::time::update_ntp_time();
                        }
                        if (_slot_zero_streak == 60) {
                            // ~15s — prolonged stall, investigate root cause
                            auto _now60 = graphene::time::now();
                            auto _hbt60 = database().head_block_time();
                            auto _nst60 = database().get_slot_time(1);
                            int64_t _drift60 = 0;
                            try { _drift60 = graphene::time::ntp_error().count(); } catch (...) {}
                            int64_t _future_ms = (fc::time_point(_hbt60) - _now60).count() / 1000;
                            bool catching_up = false;
                            try { catching_up = p2p().is_catching_up_after_pause(); } catch (...) {}
                            bool dlt_syncing = false;
                            try { dlt_syncing = chain().is_syncing(); } catch (...) {}
                            std::string _next_w60 = database().get_scheduled_validator(1);
                            bool _ours60 = _validators.count(_next_w60) > 0;
                            elog("SLOT=0 PROLONGED STALL: ${n} consecutive not_time_yet (${s}s). "
                                 "head_block_time=${hbt} is ${f}ms AHEAD of now=${now}! "
                                 "next_slot_time=${nst} next_validator=${nw} is_ours=${o}, NTP drift=${d}us. "
                                 "catching_up=${c}, dlt_syncing=${ds}, head=#${h}. "
                                 "Check: was a block with future timestamp applied?",
                                 ("n", _slot_zero_streak)("s", (_now60 - _slot_zero_streak_start).count() / 1000000)
                                 ("hbt", _hbt60)("f", _future_ms)("now", _now60)("nst", _nst60)
                                 ("nw", _next_w60)("o", _ours60)
                                 ("d", _drift60)("c", catching_up)("ds", dlt_syncing)
                                 ("h", database().head_block_num()));
                        }
                        if (_slot_zero_streak == 120) {
                            // ~30s — serious stall, head_block_time may be in the future
                            auto _now120 = graphene::time::now();
                            auto _hbt120 = database().head_block_time();
                            auto _nst120 = database().get_slot_time(1);
                            int64_t _drift120 = 0;
                            try { _drift120 = graphene::time::ntp_error().count(); } catch (...) {}
                            int64_t _future120_ms = (fc::time_point(_hbt120) - _now120).count() / 1000;
                            bool catching_up120 = false;
                            try { catching_up120 = p2p().is_catching_up_after_pause(); } catch (...) {}
                            bool dlt_syncing120 = false;
                            try { dlt_syncing120 = chain().is_syncing(); } catch (...) {}
                            const auto &wso120 = database().get_validator_schedule_object();
                            std::string shuffled_top3;
                            for (int i = 0; i < std::min<int>(3, wso120.num_scheduled_validators); i++) {
                                if (!shuffled_top3.empty()) shuffled_top3 += ",";
                                shuffled_top3 += wso120.current_shuffled_validators[i];
                            }
                            std::string _next_w120 = database().get_scheduled_validator(1);
                            bool _ours120 = _validators.count(_next_w120) > 0;
                            elog("CRITICAL: slot=0 stall for ${s}s! head_block_time=${hbt} is ${f}ms in the future "
                                 "relative to NTP time (now=${now}). next_slot_time=${nst} next_validator=${nw} is_ours=${o}, NTP drift=${d}us. "
                                 "Network is stalled. catching_up=${c}, dlt_syncing=${ds}, head=#${h}. "
                                 "shuffled_validators[0..2]=[${sw}]. "
                                 "ACTION REQUIRED: Check NTP sync, system clock, or restart node.",
                                 ("s", (_now120 - _slot_zero_streak_start).count() / 1000000)
                                 ("hbt", _hbt120)("f", _future120_ms)("now", _now120)("nst", _nst120)
                                 ("nw", _next_w120)("o", _ours120)
                                 ("d", _drift120)("c", catching_up120)("ds", dlt_syncing120)
                                 ("h", database().head_block_num())("sw", shuffled_top3));
                        }
                        break;
                    case block_validation_condition::no_private_key:
                        ilog("Not producing block for ${scheduled_validator} because I don't have the private key for ${scheduled_key}",
                             (capture));
                        break;
                    case block_validation_condition::low_participation:
                        elog("Not producing block because node appears to be on a minority fork with only ${pct}% validator participation",
                             (capture));
                        break;
                    case block_validation_condition::lag:
                        elog("Not producing block because node didn't wake up within 500ms of the slot time.");
                        graphene::time::update_ntp_time();  // Force NTP sync on timing issues
                        // After a lag, the current slot is already missed.  Skip ahead past this
                        // slot to avoid spinning in a tight loop where every 250ms tick rechecks
                        // the same missed slot and returns lag again.  By advancing _last_slot_time_
                        // to the missed slot's time, the next production loop will compute a slot
                        // that is actually in the future (or get slot=0 and wait normally).
                        _last_lag_slot_time = capture["scheduled_time"].as<fc::time_point_sec>();
                        break;
                    case block_validation_condition::consecutive:
                        elog("Not producing block because the last block was generated by the same validator.\nThis node is probably disconnected from the network so block production has been disabled.\nDisable this check with --allow-consecutive option.");
                        break;
                    case block_validation_condition::exception_validating_block:
                        elog("Failure when producing block with no transactions");
                        break;
                    case block_validation_condition::fork_collision:
                        wlog("Deferred block production due to fork collision; will retry next slot");
                        graphene::time::update_ntp_time();  // Force NTP sync on fork issues
                        break;
                    case block_validation_condition::minority_fork:
                        elog("Not producing block: minority fork detected, resyncing from P2P network");
                        break;
                }

                // === NO-NEW-BLOCKS STATUS: chain-wide stall visibility ===
                // Independent of whether WE are scheduled or producing: report when
                // NO block from ANYONE has arrived for a while, so operators can watch
                // the gap climb toward CHAIN_EMERGENCY_CONSENSUS_TIMEOUT_SEC (the point
                // at which emergency consensus activates). Logged once per minute of
                // head age. Suppressed while syncing/catching-up, where an old head
                // just reflects historical replay rather than a real network stall.
                {
                    auto &db_age = database();
                    int64_t head_age_s =
                        (fc::time_point::now() - fc::time_point(db_age.head_block_time())).count() / 1000000;
                    bool catching_up_age = false;
                    try { catching_up_age = p2p().is_catching_up_after_pause(); } catch (...) {}
                    bool dlt_syncing_age = false;
                    try { dlt_syncing_age = chain().is_syncing(); } catch (...) {}

                    if (head_age_s >= 60 && !catching_up_age && !dlt_syncing_age) {
                        // Log at most once per whole minute of head age, and re-arm
                        // when head advances (age drops back below the last logged minute).
                        static int64_t _last_headage_logged_min = 0;
                        int64_t cur_min = head_age_s / 60;
                        if (cur_min != _last_headage_logged_min) {
                            _last_headage_logged_min = cur_min;
                            std::string next_w = "?";
                            fc::time_point_sec next_slot_time;
                            try {
                                next_w = db_age.get_scheduled_validator(1);
                                next_slot_time = db_age.get_slot_time(1);
                            } catch (...) {}
                            int64_t to_emergency_s =
                                (int64_t)CHAIN_EMERGENCY_CONSENSUS_TIMEOUT_SEC - head_age_s;
                            wlog("NO NEW BLOCKS: head=#${h} produced at ${hbt}, now=${now}, "
                                 "gap=${g}s (${m} min). Next expected: block #${nb} by ${nw} at ${nst}. "
                                 "Emergency consensus activates in ~${e}s.",
                                 ("h", db_age.head_block_num())
                                 ("hbt", db_age.head_block_time())
                                 ("now", graphene::time::now())
                                 ("g", head_age_s)("m", cur_min)
                                 ("nb", db_age.head_block_num() + 1)
                                 ("nw", next_w)("nst", next_slot_time)
                                 ("e", to_emergency_s > 0 ? to_emergency_s : 0));
                        }
                    }
                }

                // Production watchdog: elog if we've produced before but have gone
                // silent for too long while production is still enabled.
                // Emergency master threshold: 60s (before 315s blanking at 105 missed blocks).
                // Regular validator threshold: 180s (before 600s blanking at 200 missed blocks).
                // Fires every 30s once triggered so the operator has multiple chances to react.
                //
                // When returning from a lag condition, skip ahead past the missed slot
                // so the production loop doesn't immediately recheck the same slot and
                // spin in a tight lag loop.
                if (_last_lag_slot_time != fc::time_point_sec()) {
                    // Compute elapsed time in milliseconds (count() is microseconds).
                    int64_t time_since_lag_ms = (fc::time_point::now() - fc::time_point(_last_lag_slot_time)).count() / 1000;
                    // If less than one full slot interval has passed since the lag, skip ahead
                    // to avoid rechecking the same missed slot.
                    if (time_since_lag_ms < (int64_t)database().block_interval() * 1000) {
                        int64_t skip_ms = (database().block_interval() * 1000) - time_since_lag_ms;
                        if (skip_ms > 0) {
                            if (database()._debug_block_production) {
                                ilog("DEBUG_CRASH: skipping ${ms}ms to avoid rechecking lagged slot", ("ms", skip_ms));
                            }
                            production_timer_.expires_from_now(posix_time::milliseconds(skip_ms));
                            production_timer_.async_wait([this](const system::error_code &) { block_validation_loop(); });
                            _last_lag_slot_time = fc::time_point_sec();
                            return result;
                        }
                    }
                    _last_lag_slot_time = fc::time_point_sec();
                }

                if (_ever_produced) {
                    // Check if production should be active by querying actual state
                    bool should_be_producing = false;
                    // Whether emergency consensus is actually active on-chain right now.
                    // Holding the committee (emergency) key alone does NOT make us an
                    // emergency master — committee only produces while emergency mode is on.
                    bool emergency_active = false;
                    try {
                        const auto& dgp_watch = database().get_dynamic_global_properties();
                        emergency_active = dgp_watch.emergency_consensus_active;
                        // Production should be active if:
                        // - Not in minority fork recovery
                        // - Validators are configured
                        // - Either emergency master OR network is healthy (participation >= 33%)
                        if (!_minority_fork_recovering && !_validators.empty()) {
                            if (emergency_active) {
                                // Emergency mode: should produce if we have emergency key
                                should_be_producing = (_validators.count(CHAIN_EMERGENCY_VALIDATOR_ACCOUNT) > 0);
                            } else {
                                // Normal mode: should produce if participation is healthy
                                uint32_t prate_watch = database().validator_participation_rate();
                                should_be_producing = (prate_watch >= 33 * CHAIN_1_PERCENT);
                            }
                        }
                    } catch (...) {}

                    if (should_be_producing) {
                        auto silent_for = fc::time_point::now() - _last_production_time;
                        // Only treat ourselves as the emergency master (tight 60s threshold)
                        // when emergency consensus is actually active. In normal mode a
                        // single-slot validator's natural production gap (~round length)
                        // can exceed 60s and would otherwise trip the watchdog falsely.
                        bool is_emrg_master = emergency_active && _validators.count(CHAIN_EMERGENCY_VALIDATOR_ACCOUNT) > 0;
                        int64_t threshold_us = is_emrg_master ? 60000000 : 180000000;
                        if (silent_for.count() > threshold_us) {
                            // === AUTO-ENABLE DEBUG LOGGING ON FIRST WATCHDOG FIRE ===
                            // When the watchdog detects that production has stopped unexpectedly,
                            // automatically enable verbose DEBUG_CRASH logging to capture
                            // detailed execution flow on every subsequent production tick.
                            // This helps diagnose why blocks are not being produced.
                            if (!_watchdog_debug_enabled) {
                                _watchdog_debug_enabled = true;
                                database()._debug_block_production = true;
                                elog("WATCHDOG: Auto-enabled _debug_block_production for detailed diagnostic logging");
                            }

                            static fc::time_point _last_watchdog_log;
                            auto _now_wdog = fc::time_point::now();
                            if ((_now_wdog - _last_watchdog_log).count() > 30000000) {
                                _last_watchdog_log = _now_wdog;
                                auto& db_wd = database();
                                bool catching_up = false;
                                try { catching_up = p2p().is_catching_up_after_pause(); } catch (...) {}
                                bool dlt_syncing = false;
                                try { dlt_syncing = chain().is_syncing(); } catch (...) {}
                                std::string validator_names;
                                for (const auto& w : _validators) { if (!validator_names.empty()) validator_names += ","; validator_names += w; }
                                int64_t ntp_us = 0;
                                try { ntp_us = graphene::time::ntp_error().count(); } catch (...) {}

                                // Who does the chain expect to produce right now?
                                std::string scheduled_now = "?";
                                bool we_are_scheduled = false;
                                // How many of our validators appear anywhere in the full shuffled schedule?
                                uint32_t our_slots_in_schedule = 0;
                                // Which of our validators have zero on-chain signing key (blanked by emergency consensus)?
                                std::string blanked_keys;
                                try {
                                    fc::time_point_sec now_sec = graphene::time::now() + fc::microseconds(250000);
                                    uint32_t cur_slot = db_wd.get_slot_at_time(now_sec);
                                    if (cur_slot > 0) {
                                        scheduled_now = db_wd.get_scheduled_validator(cur_slot);
                                        we_are_scheduled = _validators.count(scheduled_now) > 0;
                                    } else {
                                        // Between slots: show who gets the NEXT slot
                                        scheduled_now = "between_slots/" + db_wd.get_scheduled_validator(1);
                                    }

                                    // Scan full shuffled schedule for our validators
                                    const auto &wso_wd = db_wd.get_validator_schedule_object();
                                    for (int i = 0; i < wso_wd.num_scheduled_validators; i++) {
                                        if (_validators.count(wso_wd.current_shuffled_validators[i]) > 0)
                                            our_slots_in_schedule++;
                                    }

                                    // Check on-chain signing keys for our validators
                                    const auto &wit_idx = db_wd.get_index<graphene::chain::validator_index>().indices().get<graphene::chain::by_name>();
                                    for (const auto& w_name : _validators) {
                                        auto w_itr = wit_idx.find(w_name);
                                        if (w_itr != wit_idx.end() &&
                                            w_itr->signing_key == graphene::protocol::public_key_type()) {
                                            if (!blanked_keys.empty()) blanked_keys += ",";
                                            blanked_keys += w_name;
                                        }
                                    }
                                } catch (...) {}

                                int64_t head_age_s = (fc::time_point::now() - fc::time_point(db_wd.head_block_time())).count() / 1000000;

                                // _slot_hijack_count is written by on_block_applied (P2P thread):
                                // snapshot under lock before the elog.
                                uint32_t _shj_snap;
                                { std::lock_guard<std::mutex> lk(_diag_mutex); _shj_snap = _slot_hijack_count; }

                                elog("VALIDATOR-WATCHDOG: ${t} silent for ${s}s! "
                                    "validators=${w} keys=${k} skip_flags=${sf} minority_recovering=${mr} "
                                    "slot_result=${sr} dlt_syncing=${ds} catching_up=${c} "
                                    "head=#${h} head_age=${ha}s scheduled_now=${sw} we_are_scheduled=${ws} "
                                    "in_schedule=${is}/${total} blanked_keys=[${bk}] "
                                    "slot0_streak=${sz} not_my_turn_streak=${nmt} last_scheduled=${nmtw} "
                                    "ntp_offset=${n}us slot_hijacks=${shj} debug_logging=${dl}",
                                    ("t", is_emrg_master ? "emergency master" : "validator")
                                    ("s", silent_for.count() / 1000000)
                                    ("w", validator_names)
                                    ("k", _private_keys.size())
                                    ("sf", _production_skip_flags)
                                    ("mr", _minority_fork_recovering)
                                    ("sr", _last_slot_result)
                                    ("ds", dlt_syncing)
                                    ("c", catching_up)
                                    ("h", db_wd.head_block_num())
                                    ("ha", head_age_s)
                                    ("sw", scheduled_now)
                                    ("ws", we_are_scheduled)
                                    ("is", our_slots_in_schedule)
                                    ("total", _validators.size())
                                    ("bk", blanked_keys)
                                    ("sz", _slot_zero_streak)
                                    ("nmt", _not_my_turn_streak)
                                    ("nmtw", _last_scheduled_validator)
                                    ("n", ntp_us)
                                    ("shj", _shj_snap)
                                    ("dl", _watchdog_debug_enabled));

                                // === WATCHDOG PRODUCTION RECOVERY ===
                                // Brute-force recovery: if production is silent but
                                // the node is clearly operational (head advancing,
                                // FORWARD mode, peers connected), force-reset every
                                // flag that could silently block production.  This
                                // covers any safety gate that may have gotten stuck
                                // due to race conditions, stale state, or edge
                                // cases we haven't diagnosed.
                                //
                                // Conditions for recovery:
                                //   - Head is recent (external blocks arriving)
                                //   - Not in active P2P sync
                                //   - At least some peers connected
                                //   - We have validators with valid keys in schedule
                                bool head_advancing = (head_age_s >= 0 && head_age_s < 30);
                                bool has_peers = false;
                                try { has_peers = p2p().get_connections_count() > 0; } catch (...) {}
                                bool has_active_keys = (our_slots_in_schedule > 0 && blanked_keys.size() < validator_names.size());

                                if (head_advancing && !dlt_syncing && !catching_up && has_peers && has_active_keys) {
                                    bool did_recover = false;

                                    // Clear minority fork recovery state
                                    if (_minority_fork_recovering) {
                                        _minority_fork_recovering = false;
                                        did_recover = true;
                                        elog("WATCHDOG-RECOVERY: cleared _minority_fork_recovering");
                                    }

                                    // Force-clear P2P catchup flag
                                    try {
                                        if (p2p().is_catching_up_after_pause()) {
                                            p2p().clear_catchup_flag();
                                            did_recover = true;
                                            elog("WATCHDOG-RECOVERY: force-cleared P2P catching_up_after_pause flag");
                                        }
                                    } catch (...) {}

                                    // Force-clear chain syncing flag
                                    try {
                                        if (chain().is_syncing()) {
                                            chain().clear_syncing();
                                            did_recover = true;
                                            elog("WATCHDOG-RECOVERY: force-cleared chain syncing flag");
                                        }
                                    } catch (...) {}

                                    // Reset streak counters that may affect logic
                                    _not_my_turn_streak = 0;
                                    _slot_zero_streak = 0;

                                    if (did_recover) {
                                        elog("WATCHDOG-RECOVERY: production forcibly restored after ${s}s silence "
                                            "(head=#${h}, head_age=${ha}s, peers=${p}, in_schedule=${is})",
                                            ("s", silent_for.count() / 1000000)
                                            ("h", db_wd.head_block_num())
                                            ("ha", head_age_s)
                                            ("p", has_peers)
                                            ("is", our_slots_in_schedule));
                                    }
                                } else {
                                    elog("WATCHDOG-RECOVERY: skipped — conditions not met "
                                        "(head_advancing=${ha} dlt_syncing=${ds} catching_up=${cu} "
                                        "has_peers=${hp} has_active_keys=${hk})",
                                        ("ha", head_advancing)("ds", dlt_syncing)("cu", catching_up)
                                        ("hp", has_peers)("hk", has_active_keys));
                                }
                            }
                        }
                    }
                }

                if (database()._debug_block_production) ilog("DEBUG_CRASH: scheduling next production loop");
                schedule_production_loop();
                if (database()._debug_block_production) ilog("DEBUG_CRASH: block_validation_loop EXIT");
                return result;
            }

            block_validation_condition::block_validation_condition_enum validator_plugin::impl::maybe_validate_block(fc::mutable_variant_object &capture) {
                auto &db = database();
                if (db._debug_block_production) ilog("DEBUG_CRASH: maybe_validate_block ENTER");
                fc::time_point now_fine = graphene::time::now();
                fc::time_point_sec now = now_fine + fc::microseconds( 250000 );

                // Read DGP early so the DLT sync guard can check emergency consensus state.
                // In emergency mode the master MUST produce blocks regardless of sync state;
                // blocking production here creates a permanent deadlock because:
                //   - The master is the sole block producer
                //   - No blocks arrive to clear the syncing flag
                //   - The production loop is the only path to advance the chain
                if (db._debug_block_production) ilog("DEBUG_CRASH: getting dgp");
                // Copy emergency_consensus_active under a read lock so the local bool
                // stays valid even if a concurrent P2P resize remaps the segment between
                // here and the op_guard created below at line ~1629.  Keeping a raw
                // const& into shared memory without an op_guard is a dangling-reference
                // risk: begin_resize_barrier() can complete while _active_operations==0
                // and unmap the old segment before we dereference the field.
                bool emergency_active = false;
                db.with_weak_read_lock([&]() {
                    emergency_active = db.get_dynamic_global_properties().emergency_consensus_active;
                });
                if (db._debug_block_production) ilog("DEBUG_CRASH: dgp ok, head=${h} emergency=${e}", ("h", db.head_block_num())("e", emergency_active));

                // === DLT MODE: DEFER PRODUCTION DURING ACTIVE SYNC ===
                // In DLT mode, the validator must not produce blocks while the
                // chain is actively receiving sync blocks from P2P.  Producing
                // during sync creates blocks on a stale head that conflict
                // with incoming blocks, causing "failed to link" errors and
                // re-triggering sync — the oscillation bug described in
                // problem6.log.
                //
                // EMERGENCY MASTER EXCEPTION: When emergency consensus is active
                // AND this node holds the emergency-private-key (i.e. it IS the
                // master), production MUST proceed regardless of sync state.
                // The master is the sole block producer — waiting for sync to
                // complete would deadlock because no blocks arrive to clear
                // the syncing flag (p18.log).
                //
                // EMERGENCY SLAVE: Must still respect the syncing flag.  A slave
                // node that produces on a stale head creates double-production
                // collisions and minority forks (p32.log).
                //
                // Outside DLT mode this check is NOT applied because normal
                // validators must produce on the canonical chain head even
                // while the network is catching up.
                if (db._dlt_mode && chain().is_syncing()) {
                    bool we_are_emergency_master =
                        emergency_active &&
                        _validators.find(CHAIN_EMERGENCY_VALIDATOR_ACCOUNT) != _validators.end();
                    if (!we_are_emergency_master) {
                        return block_validation_condition::not_synced;
                    }
                    // Emergency master: bypass sync check to avoid deadlock.
                }

                // === SNAPSHOT PAUSE / POST-PAUSE CATCHUP GATE ===
                // Defer block production when P2P block processing is paused
                // (snapshot creation holding DB read lock) or while catching
                // up after the pause (draining queued blocks).  Producing
                // during pause deadlocks on the write lock; producing after
                // pause but before drain creates a fork on a stale head.
                //
                // This gate applies to ALL validator types (emergency and normal).
                // The flag is cleared when: pause ends + drain completes +
                // no peer is ahead (see drain_paused_block_queue / periodic_task).
                //
                // Check snapshot plugin directly for snapshot_in_progress flag.
                try {
                    if (snapshot().is_snapshot_in_progress()) {
                        wlog("Deferring block production: snapshot creation in progress "
                             "(head=#${h}). Waiting for snapshot to complete.",
                             ("h", db.head_block_num()));
                        return block_validation_condition::not_synced;
                    }
                } catch (...) {
                    // snapshot plugin may not be available
                }

                try {
                    if (p2p().is_catching_up_after_pause()) {
                        wlog("Deferring block production: P2P is catching up after "
                             "snapshot pause (head=#${h}). Waiting for gap fill.",
                             ("h", db.head_block_num()));
                        return block_validation_condition::not_synced;
                    }
                } catch (...) {
                    // p2p plugin may not be available during startup
                }

                // === HARDFORK 12: THREE-STATE SAFETY ENFORCEMENT ===
                if (db._debug_block_production) ilog("DEBUG_CRASH: checking hardfork12 and emergency path");

                if (db.has_hardfork(CHAIN_HARDFORK_12)) {
                    if (emergency_active) {
                        // EMERGENCY MODE: auto-bypass both stale and participation checks
                        // for the emergency master only.  The master holds the
                        // emergency-private-key and MUST produce to avoid deadlock.
                        //
                        // Slave nodes (no emergency key) must still sync first —
                        // producing on a stale head creates double-production
                        // collisions and minority forks (p32.log).
                        bool we_are_emergency_master =
                            _validators.find(CHAIN_EMERGENCY_VALIDATOR_ACCOUNT) != _validators.end();
                        if (!we_are_emergency_master && _validators.empty()) {
                            elog("EMERGENCY MODE ACTIVE but no validators configured! "
                                 "Block production impossible. Add --emergency-private-key to config.");
                        }
                    } else {
                        uint32_t prate = db.validator_participation_rate();
                        if (prate >= 33 * CHAIN_1_PERCENT) {
                            // HEALTHY NETWORK: enforce safe defaults automatically.
                            // Even if operator has enable-stale-production=true in config,
                            // it's overridden because the network doesn't need it.
                            // Clear the stale-production skip flag so that minority fork
                            // detection is re-enabled now that the network is healthy.
                            _production_skip_flags &= ~graphene::chain::database::skip_undo_history_check;
                            // Participation is already >= 33%, no need to check again
                        } else {
                            // DISTRESSED NETWORK (participation < 33%, not yet emergency):
                            // Honor manual config overrides -- operator may be trying to
                            // accelerate recovery before the 1-hour timeout.
                            //
                            // HF16: the 128-slot window is the pre-BFT proxy for "is a
                            // supermajority alive on my branch". A QC we hold for head or its
                            // parent IS that proof, and the unstick rule (60 s of chain silence)
                            // was written to produce exactly when the window says no. After any
                            // stall the window stays below 33% for ~2 min, so without this the
                            // chain the unstick block just restarted stops on the next slot:
                            // mainnet 02.11.2026 21:47 -- one unstick block, then 31 min of
                            // silence until an operator forced stale production, then emergency.
                            // A minority partition cannot form a QC, so it stays bounded to one
                            // uncertified block per unstick window.
                            bool quorum_proven = false;
                            if (db.has_hardfork(CHAIN_HARDFORK_16)) {
                                db.with_weak_read_lock([&]() {
                                    const uint32_t hn = db.head_block_num();
                                    auto qc_on_branch = [&](uint32_t n) {
                                        const auto qc = db.get_block_qc(n);
                                        return qc && qc->block_id == db.find_block_id_for_num(n);
                                    };
                                    quorum_proven = qc_on_branch(hn) || (hn > 1 && qc_on_branch(hn - 1)) ||
                                                    db.emergency_unstick_block(now);
                                });
                            }
                            if (prate < _required_validator_participation && !quorum_proven) {
                                if (_production_skip_flags & graphene::chain::database::skip_undo_history_check) {
                                    // enable-stale-production=true: operator override, produce anyway
                                    // to bootstrap/recover a fully stalled network where all nodes
                                    // see low participation and would otherwise deadlock.
                                    dlog("Validator participation is ${p}% but stale-production is enabled, "
                                         "producing anyway to recover stalled network",
                                         ("p", uint32_t(prate / CHAIN_1_PERCENT)));
                                } else {
                                    // Network partition guard: if participation is below 33%
                                    // this node is likely in a minority network segment.
                                    // Producing here risks two partitions simultaneously
                                    // building chains — each seeing only the other segment's
                                    // validators as absent, neither triggering minority_fork
                                    // detection below (which requires ALL recent fork_db
                                    // blocks to be ours).  Stopping production is the safe
                                    // choice; use enable-stale-production=true to override
                                    // when you know the low participation is caused by
                                    // offline validators rather than a partition.
                                    capture("pct", uint32_t(prate / CHAIN_1_PERCENT));
                                    return block_validation_condition::low_participation;
                                }
                            }
                        }
                    }
                } else {
                    // Pre-hardfork 12: no participation check here (done later)
                }

                //try get block post validation list for each validator
                //if validator can validate it, sign chain_id and block_id for message
                //broadcast validation message by p2p plugin
                if (db._debug_block_production) ilog("DEBUG_CRASH: emergency/participation check done, entering block_post_validation");
                if(last_block_post_validation_time < now_fine ){
                    last_block_post_validation_time = now;
                    if (db._debug_block_production) ilog("DEBUG_CRASH: block_post_validation tick, iterating ${n} validators", ("n", _validators.size()));

                    // Pre-compute the current scheduled validators set so we can skip
                    // configured validators that are not actually scheduled.  A validator
                    // that is not in the current schedule cannot contribute to LIB
                    // advancement and broadcasting their post-validation is wasted
                    // bandwidth and CPU.
                    const validator_schedule_object &wso = db.get_validator_schedule_object();
                    std::set<string> scheduled_validators_set;
                    for (int i = 0; i < wso.num_scheduled_validators; i += CHAIN_BLOCK_VALIDATOR_REPEAT) {
                        if (wso.current_shuffled_validators[i] != account_name_type()) {
                            scheduled_validators_set.insert(wso.current_shuffled_validators[i]);
                        }
                    }

                    //get block post validation for each validator we have
                    for (auto &validator_account : _validators) {
                        // Skip validators not in the current schedule — they cannot
                        // contribute to block post validation and broadcasting their
                        // signatures is pointless network spam.
                        if (scheduled_validators_set.find(validator_account) == scheduled_validators_set.end()) {
                            continue;
                        }

                        bool ignore_validator = false;
                        if (db._debug_block_production) ilog("DEBUG_CRASH: get_validator_confirmations for ${w}", ("w", validator_account));
                        auto block_post_validations = db.get_validator_confirmations(validator_account);
                        if (db._debug_block_production) ilog("DEBUG_CRASH: got ${n} post_validations for ${w}", ("n", block_post_validations.size())("w", validator_account));
                        if (block_post_validations.size() > 0) {
                            const auto &validator_by_name = db.get_index<graphene::chain::validator_index>().indices().get<graphene::chain::by_name>();
                            auto w_itr = validator_by_name.find(validator_account);
                            if (w_itr == validator_by_name.end()) {
                                wlog("Validator ${w} not found in validator index, skipping block post validation", ("w", validator_account));
                                continue;
                            }
                            graphene::protocol::public_key_type validator_pub_key = w_itr->signing_key;

                            // Skip validators with zero/null signing key (intentionally disabled)
                            if (validator_pub_key == graphene::protocol::public_key_type()) {
                                ignore_validator = true;
                            }

                            auto private_key_itr = _private_keys.find(validator_pub_key);

                            if (!ignore_validator && private_key_itr == _private_keys.end()) {
                                ilog("No private key to public ${p} for ${w}", ("p", validator_pub_key)("w", validator_account));
                                ignore_validator = true;
                            }
                            if(!ignore_validator){
                                if (db._debug_block_production) ilog("DEBUG_CRASH: signing post_validations for ${w}", ("w", validator_account));
                                graphene::protocol::private_key_type validator_priv_key = private_key_itr->second;
                                //we have block post validations for this validator
                                //check if we have a block
                                // Round = slot of the block. Vote in ascending round order:
                                // the safety rule refuses any round at or below the last one.
                                struct candidate { uint32_t round; uint32_t num; block_id_type id; };
                                std::vector<candidate> candidates;
                                for (const auto &bpv : block_post_validations) {
                                    if (bpv.block_num == 0 || bpv.block_id == block_id_type()) continue;
                                    const auto blk = db.fetch_block_by_id(bpv.block_id);
                                    if (!blk) continue;          // not applied here: nothing to vote for yet
                                    candidates.push_back({db.round_of(blk->timestamp),
                                                          bpv.block_num, bpv.block_id});
                                }
                                if (db.has_hardfork(CHAIN_HARDFORK_16)) {
                                    const uint32_t hn = db.head_block_num();
                                    bool listed = false, cand = false;
                                    for (const auto &bpv : block_post_validations) if (bpv.block_num == hn) listed = true;
                                    for (const auto &c : candidates) if (c.num == hn) cand = true;
                                    // The producer never votes its own block: not a finding.
                                    const bool own = db.get_dynamic_global_properties().current_validator == validator_account;
                                    if (!cand && !own && head_vote_diag_logged_.insert({std::string(validator_account), db.head_block_id()}).second) {
                                        if (head_vote_diag_logged_.size() > 4096) head_vote_diag_logged_.clear();
                                        wlog("HF16 no vote candidate for head #${n} by ${w}: in confirmation list ${l} (${k} entries)",
                                             ("n", hn)("w", validator_account)("l", listed)("k", block_post_validations.size()));
                                    }
                                }
                                if (!candidates.empty()) {
                                    const uint32_t min_round = std::min_element(candidates.begin(), candidates.end(),
                                            [](const candidate &a, const candidate &b) { return a.round < b.round; })->round;
                                    for (auto it = vote_resend_.begin(); it != vote_resend_.end(); )
                                        if (it->first.first == std::string(validator_account) && it->first.second < min_round)
                                            it = vote_resend_.erase(it);
                                        else ++it;
                                }
                                std::sort(candidates.begin(), candidates.end(),
                                          [](const candidate &a, const candidate &b) { return a.round < b.round; });
                                const uint32_t lib = db.get_dynamic_global_properties().last_irreversible_block_num;
                                for (const auto &c : candidates) {
                                    // Jolteon: vote for round r only on QC(r-1) or TC(r-1), so a
                                    // block that jumped over rounds without a TC gets no QC. A block
                                    // without a parent QC (HF16 start, after emergency) is exempt.
                                    if (db.has_hardfork(CHAIN_HARDFORK_16)) {
                                        const auto blk = db.fetch_block_by_id(c.id);
                                        uint32_t qc_round = 0, tc_round = 0;
                                        bool has_qc = false;
                                        for (const auto &e : blk->extensions) {
                                            if (e.which() == 3) { has_qc = true; qc_round = e.get<graphene::protocol::block_qc_certificate>().round; }
                                            else if (e.which() == 4) tc_round = e.get<graphene::protocol::block_timeout_certificate>().round;
                                            // HF18 carries the same certificate in its own extension; the
                                            // round is what covers the gap either way.
                                            else if (e.which() == 6) tc_round = e.get<graphene::protocol::block_timeout_certificate_v2>().round;
                                        }
                                        // Late-QC fallback (grandparent QC, consecutive rounds) is
                                        // valid too, unless we hold a QC for a SIBLING of the parent
                                        // (another block at its height): voting past a certified
                                        // block we know of is what Jolteon's rule forbids.
                                        bool late_ok = false;
                                        if (has_qc && qc_round + 2 == c.round) {
                                            const auto held = db.get_block_qc(c.num - 1);
                                            late_ok = !held || held->block_id == blk->previous;
                                        }
                                        // Same window validate_block_qc accepts: a TC on ANY round
                                        // strictly between the parent's QC round and ours covers the
                                        // gap, so the block is chain-valid and may be voted for.
                                        // Requiring exactly TC(round-1) refused every block a stalled
                                        // chain can build (mainnet stall 31.10.2026: TCs formed for
                                        // 245..372 while the producer's slot round was 600+, so not
                                        // one rebuilt block ever collected a vote).
                                        const bool tc_covers_gap = tc_round > qc_round && tc_round < c.round;
                                        // The exemption below is for a chain with no certificate to
                                        // carry; it is not a licence to vote past one we hold (#236
                                        // unstick blocks compete with a certified head otherwise).
                                        if (!has_qc && db.qc_conflicts_with(*blk)) {
                                            wlog("Not voting for QC-less #${n} round ${r}: a held QC certifies another block at or above its parent",
                                                 ("n", c.num)("r", c.round));
                                            continue;
                                        }
                                        if (has_qc && qc_round + 1 != c.round && !tc_covers_gap && !late_ok) {
                                            wlog("Not voting for #${n} round ${r}: parent QC round ${q}, TC round ${t}",
                                                 ("n", c.num)("r", c.round)("q", qc_round)("t", tc_round));
                                            continue;
                                        }
                                    }
                                    if (!may_vote(validator_account, c.round, c.num, c.id, lib)) continue;
                                    auto &rs = vote_resend_[std::make_pair(std::string(validator_account), c.round)];
                                    if (rs.sent > 0) {
                                        if (rs.sent > VOTE_RESEND_MAX || now_fine < rs.next) continue;
                                        const auto held = db.get_block_qc(c.num);
                                        if (held && held->block_id == c.id && held->round >= c.round) continue;
                                        ilog("Re-sending vote of ${w} for #${n} round ${r} (no QC yet, attempt ${k})",
                                             ("w", validator_account)("n", c.num)("r", c.round)("k", rs.sent));
                                    }
                                    graphene::protocol::signature_type vote_signature = validator_priv_key.sign_compact(
                                            graphene::protocol::block_vote_digest(db.get_chain_id(), c.num, c.id, c.round));
                                    p2p().post_broadcast_block_post_validation(c.id, validator_account, vote_signature, c.round);
                                    const uint32_t backoff = std::min<uint32_t>(1u << std::min<uint32_t>(rs.sent, 3), 8);
                                    rs.next = now_fine + fc::seconds(int64_t(db.block_interval()) * backoff);
                                    ++rs.sent;
                                }
                            }
                        }
                    }
                }

                // HF16 timeout votes. The head's round is r; the producer of r+1 needs
                // QC(head). If the whole slot r+1 has passed and that QC still is not
                // here, each of our scheduled validators times out round r, naming the
                // round of the highest QC it holds (the one head carries). A quorum of
                // timeouts forms TC(r), which lets the next producer skip the QC-less head.
                // Sent once per round; a lost timeout is not resent (gossip relays it).
                if (!emergency_active && db.has_hardfork(CHAIN_HARDFORK_16) && db.head_block_num() > 0) {
                    const uint32_t head_num = db.head_block_num();
                    const uint32_t head_round = db.round_of(db.head_block_time());
                    const uint32_t now_round = db.round_of(now);
                    const auto head_qc = db.get_block_qc(head_num);
                    const bool certified = head_qc && head_qc->block_id == db.head_block_id();
                    // Certificate gossip. Votes always travelled; the QC they add up to never
                    // did, so a head certified by a quorum stayed a private fact of the few
                    // nodes that collected the votes and the next producer had nothing to
                    // extend it with (mainnet sat 5 h 20 min on #84865875, 01.11.2026).
                    // While the chain is not moving, re-send the QC we hold for the head: that
                    // is what hands it to a node that restarted and forgot, and to whoever
                    // produces next. Silent while the chain advances, so normal operation and
                    // its bandwidth are untouched.
                    if (certified && now_round >= head_round + 2) {
                        graphene::protocol::block_qc_certificate qc;
                        qc.block_num = head_num;
                        qc.block_id = head_qc->block_id;
                        qc.round = head_qc->round;
                        for (const auto &sv : head_qc->signatures) qc.signatures.emplace(sv.first, sv.second);
                        p2p().post_broadcast_block_qc(qc);
                    }
                    // Time out the last fully passed round t = now-1 once no block of a
                    // later round than head has come: head certified -> the slot after it
                    // stayed empty; head QC-less -> head is being given up on too.
                    const auto head_block = db.fetch_block_by_number(head_num);
                    // Highest QC we hold: head's own if certified, else the QC head carries.
                    // Not block_cert_round(): it also counts a TC, and a TC round reported
                    // as a QC round made every timeout vote claim a QC that never existed,
                    // so the TC skip of that head was refused forever (21-validator stand,
                    // netem 300 ms: head #183 with TC 691, all timeouts "high QC 691").
                    uint32_t carried_qc_round = 0;
                    graphene::protocol::block_id_type carried_qc_block_id;
                    if (head_block)
                        for (const auto &e : head_block->extensions)
                            if (e.which() == 3) {
                                const auto &carried = e.get<graphene::protocol::block_qc_certificate>();
                                carried_qc_round = carried.round;
                                carried_qc_block_id = carried.block_id;
                            }
                    // Pacemaker backoff (Q#3123=A). A QC-less head whose QC lags it by g
                    // rounds has been rebuilt after TCs in a row: wait w = 2^floor(log2 g)
                    // rounds (1, 2, 4, capped at 8) before giving up on it, so votes slower
                    // than a slot still reach a quorum. g comes from the head itself, so
                    // every validator on it picks the same w and the timeouts meet in one
                    // round. A certified head, or a QC-less one right after its QC, keeps
                    // w = 1. Without it, at 300 ms per hop each rebuilt head was given up
                    // on before its votes arrived (21-validator stand: #179 rebuilt 444
                    // times, TC every round, height frozen).
                    uint32_t wait_rounds = 1;
                    if (!certified && carried_qc_round > 0 && head_round > carried_qc_round + 1) {
                        const uint32_t g = head_round - carried_qc_round;
                        while (wait_rounds * 2 <= g && wait_rounds < 8) wait_rounds *= 2;
                    }
                    if (now_round >= head_round + 1 + wait_rounds) {
                        const uint32_t t = now_round - 1;
                        // Emergency vote honesty (owner 01.11.2026). A QC in our memory is not
                        // provable to anyone else, and a timeout vote naming a round above the
                        // parent QC poisons every TC built with it (validate_block_qc wants
                        // every vote <= parent QC round). On 01.11.2026 three nodes voting the
                        // head's own round left 12 honest votes against a quorum of 13, and
                        // the head could not be skipped. After CHAIN_EMERGENCY_UNSTICK_TIMEOUT_SEC
                        // of a still chain, claim only the QC head itself proves -- the one it
                        // carries -- so the three become honest and the TC reaches a quorum.
                        const bool stalled =
                                (now - db.head_block_time()).to_seconds() >= CHAIN_EMERGENCY_UNSTICK_TIMEOUT_SEC;
                        const bool claim_head_qc = certified && !stalled;
                        const uint32_t high_qc_round = claim_head_qc ? head_qc->round : carried_qc_round;
                        // HF18: the claim names the BLOCK, not only the round. Both sources
                        // below carry a block id already (the head QC certifies the head; a
                        // carried QC names the block it certifies), so the pair is always
                        // one that exists on chain. Round 0 means "no QC at all" and goes
                        // with the zero id.
                        const graphene::protocol::block_id_type high_qc_block_id =
                                claim_head_qc ? head_qc->block_id : carried_qc_block_id;
                        const bool hf18 = db.has_hardfork(CHAIN_HARDFORK_18);
                        const auto &wso = db.get_validator_schedule_object();
                        const auto &by_name = db.get_index<graphene::chain::validator_index>().indices().get<graphene::chain::by_name>();
                        for (int i = 0; i < wso.num_scheduled_validators; i += CHAIN_BLOCK_VALIDATOR_REPEAT) {
                            const std::string v = wso.current_shuffled_validators[i];
                            if (!_validators.count(v)) continue;
                            auto w_itr = by_name.find(v);
                            if (w_itr == by_name.end() || w_itr->signing_key == graphene::protocol::public_key_type()) continue;
                            auto key_itr = _private_keys.find(w_itr->signing_key);
                            if (key_itr == _private_keys.end()) continue;
                            if (high_qc_round >= t) continue;   // malformed head; push would refuse it
                            if (!may_timeout(v, t)) continue;
                            // A v2 vote is "round 0 with the zero id" (no QC) or "a round and
                            // the block certified there", never half of each: every node
                            // refuses the mixed form. Normalise instead of skipping, so a head
                            // that names a round without a block still gets a vote out.
                            uint32_t claim_round = high_qc_round;
                            graphene::protocol::block_id_type claim_id = high_qc_block_id;
                            if (claim_round == 0 || claim_id == graphene::protocol::block_id_type()) {
                                claim_round = 0;
                                claim_id = graphene::protocol::block_id_type();
                            }
                            if (hf18) {
                                const auto sig = key_itr->second.sign_compact(
                                        graphene::protocol::block_timeout_digest_v2(db.get_chain_id(), t,
                                                claim_round, claim_id));
                                ilog("Timeout: ${w} gives up on round ${r} (head #${h} round ${hr}, ${c}; high QC round ${q}${b})",
                                     ("w", v)("r", t)("h", head_num)("hr", head_round)
                                     ("c", certified ? "certified" : "no QC")("q", claim_round)
                                     ("b", claim_round ? " on block " + std::string(claim_id) : std::string()));
                                p2p().post_broadcast_timeout_vote_v2(t, claim_round, claim_id, v, sig);
                            } else {
                                const auto sig = key_itr->second.sign_compact(
                                        graphene::protocol::block_timeout_digest(db.get_chain_id(), t, high_qc_round));
                                ilog("Timeout: ${w} gives up on round ${r} (head #${h} round ${hr}, ${c}; high QC round ${q})",
                                     ("w", v)("r", t)("h", head_num)("hr", head_round)
                                     ("c", certified ? "certified" : "no QC")("q", high_qc_round));
                                p2p().post_broadcast_timeout_vote(t, high_qc_round, v, sig);
                            }
                        }
                    }
                }

                if (db._debug_block_production) ilog("DEBUG_CRASH: block_post_validation done, entering minority fork detection");
                // === MINORITY FORK DETECTION ===
                // If the last CHAIN_MAX_VALIDATORS (21) blocks in fork_db were ALL
                // produced by our own configured validators, we are likely stuck on
                // a minority fork where no external validators are participating.
                //
                // SKIP during emergency consensus: in emergency mode all blocks are
                // produced by the committee account (which is in _validators), so the
                // check would always falsely trigger and kill recovery.
                //
                // EXCEPTION: In DLT mode, even during emergency consensus, we apply
                // a higher-threshold minority fork check.  See the DLT-specific block
                // below.
                //
                // With enable-stale-production=true: operator knows what they're doing,
                //   continue producing (bootstrap / testnet / recovery scenario).
                // With enable-stale-production=false (default): we're on the wrong fork,
                //   pop back to LIB and resync from the P2P network.
                if (!emergency_active && !_disable_minority_fork_detection) {
                    auto fork_head = db.get_fork_db().head();
                    if (fork_head) {
                        bool all_ours = true;
                        uint32_t blocks_checked = 0;
                        auto current = fork_head;

                        while (current && blocks_checked < CHAIN_MAX_VALIDATORS) {
                            if (_validators.find(current->data.validator) == _validators.end()) {
                                all_ours = false;
                                break;
                            }
                            blocks_checked++;
                            current = current->prev.lock();
                        }

                        if (all_ours && blocks_checked >= CHAIN_MAX_VALIDATORS) {
                            if (_production_skip_flags & graphene::chain::database::skip_undo_history_check) {
                                // enable-stale-production=true: operator override, continue
                                dlog("Minority fork detected (last ${n} blocks from our validators) "
                                     "but stale production enabled, continuing",
                                     ("n", blocks_checked));
                            } else {
                                // Wrong fork: trigger recovery
                                elog("MINORITY FORK DETECTED: last ${n} blocks all from our validators. "
                                     "Resetting to LIB and resyncing from P2P network.",
                                     ("n", blocks_checked));
                                p2p().resync_from_lib();
                                _minority_fork_recovering = true;
                                _minority_fork_recovery_start = fc::time_point::now();
                                return block_validation_condition::minority_fork;
                            }
                        }
                    }
                }

                // === DLT-SPECIFIC MINORITY FORK DETECTION IN EMERGENCY MODE ===
                // In emergency + DLT mode, the standard minority fork check above is
                // skipped because committee blocks are produced by an account that
                // may be in _validators.  However, a DLT emergency validator that has
                // lost its P2P connection to the master will produce blocks for its
                // own validator slots AND the committee slots (because the emergency
                // key covers committee).  After a few rounds with NO external blocks
                // at all, the node is on a minority fork.
                //
                // Detect this by checking whether the last full round (21 blocks)
                // in fork_db contain ONLY blocks from our validators.  In a healthy
                // emergency hybrid schedule, committee slots are filled by the master
                // node's blocks — so we should see non-our-validator blocks regularly.
                // If we don't, we're isolated.
                //
                // We use 1 round (21 blocks) because in a healthy emergency hybrid
                // schedule the committee (master) produces at least 1 block per
                // round, so we should never see 21 consecutive blocks from only
                // our validators unless we're isolated from the master.  This matches
                // the standard non-emergency minority fork threshold.
                //
                // IMPORTANT: If committee (CHAIN_EMERGENCY_VALIDATOR_ACCOUNT) is in the
                // current validator schedule AND we have its key (emergency-private-key
                // configured), this node IS the emergency master.  All blocks being
                // "ours" is expected — other nodes sync from us.  Skip minority fork
                // detection entirely to avoid false positives and the production
                // deadlock that would otherwise occur.
                if (emergency_active && db._dlt_mode && !_disable_minority_fork_detection) {
                    // If committee is in the schedule and we have its key, WE are the
                    // emergency master.  All blocks being "ours" is expected -- other
                    // nodes sync from us.  Skip minority fork detection to prevent
                    // false positives and the production deadlock.
                    // Check both conditions: (a) committee is in the schedule, AND
                    // (b) we have its key (committee is in _validators only when
                    // emergency-private-key was configured — see plugin_initialize).
                    bool we_are_master = false;
                    if (_validators.find(CHAIN_EMERGENCY_VALIDATOR_ACCOUNT) != _validators.end()) {
                        const validator_schedule_object &wso = db.get_validator_schedule_object();
                        for (int i = 0; i < wso.num_scheduled_validators; i += CHAIN_BLOCK_VALIDATOR_REPEAT) {
                            if (wso.current_shuffled_validators[i] == CHAIN_EMERGENCY_VALIDATOR_ACCOUNT) {
                                we_are_master = true;
                                break;
                            }
                        }
                    }

                    if (!we_are_master) {
                        // Slave DLT node: committee not in schedule or we don't have
                        // the key.  Run the existing fork_db isolation scan.
                        auto fork_head = db.get_fork_db().head();
                        if (fork_head) {
                            const uint32_t dlt_minority_threshold = CHAIN_MAX_VALIDATORS; // 21 blocks = 1 full round
                            bool all_ours = true;
                            uint32_t blocks_checked = 0;
                            auto current = fork_head;

                            while (current && blocks_checked < dlt_minority_threshold) {
                                if (_validators.find(current->data.validator) == _validators.end()) {
                                    all_ours = false;
                                    break;
                                }
                                blocks_checked++;
                                current = current->prev.lock();
                            }

                            if (all_ours && blocks_checked >= dlt_minority_threshold) {
                                elog("DLT EMERGENCY MINORITY FORK DETECTED: last ${n} blocks all from our "
                                     "validators (1+ full rounds). Node is isolated from master. "
                                     "Resetting to LIB and resyncing from P2P network.",
                                     ("n", blocks_checked));

                                // Escalation: a rewind stays inside our state, so a state that keeps
                                // re-deriving the same fork is rewound into itself forever — measured
                                // 31.10.2026, once per 250 ms tick for an hour with the head never
                                // moving. After FORK_ESCALATION_SEC of that, replace the state with a
                                // FRESH snapshot downloaded from trusted peers (our own snapshots are
                                // behind our head and would replay back into the same state), then
                                // wipe + import + dlt replay with P2P paused and resumed. The snapshot
                                // plugin owns that sequence and answers whether it could start; it is
                                // not attempt_auto_recovery(), which imports a local file. Once per
                                // stuck head.
                                const int64_t FORK_ESCALATION_SEC = 60;
                                const uint32_t stuck_head = db.head_block_num();
                                if (stuck_head != _fork_loop_head) {
                                    _fork_loop_head = stuck_head;
                                    _fork_loop_since = fc::time_point::now();
                                    _fork_escalated = false;
                                } else if (!_fork_escalated &&
                                           (fc::time_point::now() - _fork_loop_since).to_seconds() >= FORK_ESCALATION_SEC) {
                                    _fork_escalated = true;
                                    const bool recovery_started = request_forced_snapshot_recovery();
                                    if (recovery_started) {
                                        elog("DLT MINORITY FORK UNRESOLVED for ${s}s (head stuck at #${h}): P2P resync "
                                             "keeps returning to the same head. Escalating to a forced re-import of a "
                                             "fresh snapshot from trusted peers — the state is replaced, not rewound.",
                                             ("s", FORK_ESCALATION_SEC)("h", stuck_head));
                                        _minority_fork_recovering = true;
                                        _minority_fork_recovery_start = fc::time_point::now();
                                        return block_validation_condition::minority_fork;
                                    }
                                    wlog("DLT MINORITY FORK UNRESOLVED for ${s}s (head stuck at #${h}), but no forced "
                                         "recovery could be started (see the snapshot plugin's log: no trusted peers, "
                                         "no snapshot ahead, or a snapshot operation already running) — staying on P2P "
                                         "resync, retrying in ${s}s.",
                                         ("s", FORK_ESCALATION_SEC)("h", stuck_head));
                                    _fork_loop_since = fc::time_point::now();
                                    _fork_escalated = false;
                                }

                                p2p().resync_from_lib(true /*force_emergency*/);
                                _minority_fork_recovering = true;
                                _minority_fork_recovery_start = fc::time_point::now();
                                return block_validation_condition::minority_fork;
                            }
                        }
                    } else {
                        // ── Emergency master: producing is not a licence to ignore the peers ──
                        // 02.11.2026, api.viz.world: after a 6-minute stall this node produced
                        // committee blocks on its own tip while the network resumed on another
                        // branch, and the public API served the minority fork for ~18 minutes.
                        // The exemption above is exactly why nothing caught it: "all blocks are
                        // ours" is expected FROM the master, and the master's head keeps moving,
                        // so the stalled-sync escalation cannot fire either. Peers are the only
                        // witnesses left, and their LIB is the one part of the claim that cannot
                        // be invented without two thirds of the validators: several peers
                        // finalising PAST OUR LIB means the network is building elsewhere.
                        const auto peer_libs = p2p().get_peer_lib_view();
                        if (!_startup_cleared && peer_libs.witnesses > 0 &&
                            peer_libs.best_peer_lib <= peer_libs.our_lib) {
                            _startup_cleared = true;
                        }
                        if (!_startup_cleared) {
                            const int64_t since_start = (fc::time_point::now() - _startup_time).to_seconds();
                            // The grace frees only a master nobody has answered: a witness that
                            // reports a LIB above ours keeps it waiting, because with a single
                            // matching seed the quorum gate below never fires.
                            if (since_start < EMERGENCY_STARTUP_WITNESS_GRACE_SEC || peer_libs.witnesses > 0) {
                                wlog("EMERGENCY MASTER: startup ${s}s ago, ${w} witness(es), best peer LIB "
                                     "#${pl} vs ours #${ol} — not signing a committee block on head #${h} "
                                     "until a seed/trusted peer confirms we are not behind (grace ${g}s)",
                                     ("s", since_start)("w", peer_libs.witnesses)("pl", peer_libs.best_peer_lib)
                                     ("ol", peer_libs.our_lib)("h", db.head_block_num())
                                     ("g", EMERGENCY_STARTUP_WITNESS_GRACE_SEC));
                                return block_validation_condition::not_synced;
                            }
                            _startup_cleared = true;  // isolated past the grace: fall through
                        }
                        if (peer_libs.peers_above_our_lib >= EMERGENCY_PEER_LIB_QUORUM &&
                            peer_libs.best_peer_lib > peer_libs.our_lib) {
                            const uint32_t stuck_head = db.head_block_num();
                            elog("EMERGENCY MASTER: ${n} peer(s) report LIB #${pl} above ours #${ol} — "
                                 "refusing to extend our tip (head=#${h}) with a committee block: the "
                                 "network is finalising a chain we are not on",
                                 ("n", peer_libs.peers_above_our_lib)("pl", peer_libs.best_peer_lib)
                                 ("ol", peer_libs.our_lib)("h", stuck_head));

                            // Rewind first (the fork point is usually still below our LIB); if the
                            // same head keeps coming back, the state itself is the problem and only
                            // a replacement fixes it — same escalation as the slave path above.
                            if (stuck_head != _master_gate_head) {
                                _master_gate_head = stuck_head;
                                _master_gate_since = fc::time_point::now();
                                _master_gate_escalated = false;
                            } else if (!_master_gate_escalated &&
                                       (fc::time_point::now() - _master_gate_since).to_seconds() >=
                                           EMERGENCY_PEER_LIB_ESCALATION_SEC) {
                                _master_gate_escalated = true;
                                const bool recovery_started = request_forced_snapshot_recovery();
                                if (recovery_started) {
                                    elog("EMERGENCY MASTER: still behind peer LIB after ${s}s at head #${h}; "
                                         "replacing the state with a fresh snapshot from trusted peers",
                                         ("s", EMERGENCY_PEER_LIB_ESCALATION_SEC)("h", stuck_head));
                                    _minority_fork_recovering = true;
                                    _minority_fork_recovery_start = fc::time_point::now();
                                    return block_validation_condition::minority_fork;
                                }
                                wlog("EMERGENCY MASTER: still behind peer LIB after ${s}s at head #${h}, "
                                     "but no forced recovery could be started (see the snapshot plugin) — "
                                     "staying on P2P resync, retrying in ${s}s",
                                     ("s", EMERGENCY_PEER_LIB_ESCALATION_SEC)("h", stuck_head));
                                _master_gate_since = fc::time_point::now();
                                _master_gate_escalated = false;
                            }

                            p2p().resync_from_lib(true /*force_emergency*/);
                            _minority_fork_recovering = true;
                            _minority_fork_recovery_start = fc::time_point::now();
                            return block_validation_condition::minority_fork;
                        }

                        if (db._debug_block_production) {
                            ilog("DEBUG_CRASH: DLT minority fork check SKIPPED - we are emergency master "
                                 "(no peer reports a LIB above ours)");
                        }
                    }
                }

                // Guard lockless reads into shared memory with the resize barrier.
                // This prevents a concurrent shared memory resize from invalidating
                // pointers while we read validator schedule, slot time, etc.
                // The guard is released before generate_block() which has its own.
                if (db._debug_block_production) ilog("DEBUG_CRASH: creating op_guard");
                fc::time_point _guard_enter = fc::time_point::now();
                auto op_guard = db.make_operation_guard();
                if (db._debug_block_production) ilog("DEBUG_CRASH: op_guard ok");

                // Re-capture 'now' after acquiring op_guard: if make_operation_guard()
                // blocked on a DB resize, the original 'now' (captured at function entry)
                // is stale and get_slot_at_time() would return 0, causing the production
                // loop to silently miss all blocks until the watchdog fires.
                now_fine = graphene::time::now();
                now = now_fine + fc::microseconds(250000);

                // Detect op_guard stall crossing a slot boundary.
                // A stall of 3+ seconds shifts 'now' into the next validator's slot,
                // causing not_my_turn even when our slot just passed — silent miss.
                {
                    int64_t _guard_ms = (fc::time_point::now() - _guard_enter).count() / 1000;
                    if (_guard_ms > 100) {
                        uint32_t _slot_before = db.get_slot_at_time(now_fine + fc::microseconds(250000) - fc::microseconds(_guard_ms * 1000));
                        std::string _wit_before = _slot_before > 0 ? db.get_scheduled_validator(_slot_before) : "none";
                        bool _our_slot_lost = _slot_before > 0 && _validators.count(_wit_before) > 0;
                        if (_our_slot_lost) {
                            elog("VALIDATOR-SLOT-LOST: op_guard stall ${d}ms crossed slot boundary! "
                                 "missed slot for ${w} — now points to next slot after refresh. head=#${h}",
                                 ("d", _guard_ms)("w", _wit_before)("h", db.head_block_num()));
                        } else {
                            wlog("VALIDATOR-GUARD-STALL: op_guard blocked ${d}ms (slot before=${sb} validator=${w}). head=#${h}",
                                 ("d", _guard_ms)("sb", _slot_before)("w", _wit_before)("h", db.head_block_num()));
                        }
                    }
                }

                // is anyone scheduled to produce now or one second in the future?
                if (db._debug_block_production) ilog("DEBUG_CRASH: get_slot_at_time");
                uint32_t slot = db.get_slot_at_time(now);
                if (db._debug_block_production) ilog("DEBUG_CRASH: slot=${s}", ("s", slot));
                if (slot == 0) {
                    capture("next_time", db.get_slot_time(1));
                    // Emergency master diagnostic: log when we are stuck at slot=0 and
                    // real time is well past the expected next slot (i.e. we should have
                    // a slot available but get_slot_at_time says 0 — NTP or head-time anomaly)
                    if (_validators.find(CHAIN_EMERGENCY_VALIDATOR_ACCOUNT) != _validators.end()) {
                        const auto &_dgp2 = db.get_dynamic_global_properties();
                        if (_dgp2.emergency_consensus_active) {
                            static fc::time_point _last_slot0_log;
                            auto _now2 = fc::time_point::now();
                            if ((_now2 - _last_slot0_log).count() > 10000000) { // log every 10s max
                                _last_slot0_log = _now2;
                                dlog("EMRG-DIAG slot=0: head=#${h} head_time=${ht} now=${now} next_slot=${ns} aslot=${a} num_sched=${ns2}",
                                    ("h", _dgp2.head_block_number)
                                    ("ht", db.head_block_time())
                                    ("now", now_fine)
                                    ("ns", db.get_slot_time(1))
                                    ("a", _dgp2.current_aslot)
                                    ("ns2", db.get_validator_schedule_object().num_scheduled_validators));
                            }
                        }
                    }
                    // NTP drift check: informational only. The measured offset is the
                    // software correction graphene::time::now() ALREADY applies, and slot
                    // math above runs on that corrected time — so this drift does not by
                    // itself cause missed slots. It merely signals that the host system
                    // clock is poorly disciplined (common on Windows / W32Time); the
                    // operator should fix it at the OS level for robustness.
                    {
                        int64_t ntp_us = 0;
                        try { ntp_us = graphene::time::ntp_error().count(); } catch (...) {}

                        #if defined(_WIN32)
                        // Windows (W32Time) syncs rarely and loosely, so a larger standing
                        // offset is normal there — raise the notice threshold accordingly.
                        constexpr int64_t NTP_WARN_THRESHOLD_US = 500000; // 500ms Windows
                        #else
                        constexpr int64_t NTP_WARN_THRESHOLD_US = 250000;  // 250ms Linux/macOS
                        #endif

                        if (ntp_us > NTP_WARN_THRESHOLD_US) {
                            static fc::time_point _last_ntp_drift_log;
                            auto _now_nd = fc::time_point::now();
                            if ((_now_nd - _last_ntp_drift_log).count() > 600000000) { // at most once per 10 min
                                _last_ntp_drift_log = _now_nd;
                                auto next_slot_time = db.get_slot_time(1);
                                ilog("NTP offset: host clock is ${n}ms off NTP (corrected in software; "
                                     "consider disciplining the system clock). "
                                     "(now=${now} next_slot=${ns} head=#${h})",
                                     ("n", ntp_us / 1000)("now", now_fine)
                                     ("ns", next_slot_time)("h", db.head_block_num()));
                            }
                        }
                    }
                    return block_validation_condition::not_time_yet;
                }

                //
                // this assert should not fail, because now <= db.head_block_time()
                // should have resulted in slot == 0.
                //
                // if this assert triggers, there is a serious bug in get_slot_at_time()
                // which would result in allowing a later block to have a timestamp
                // less than or equal to the previous block
                //
                assert(now > db.head_block_time());

                if (db._debug_block_production) ilog("DEBUG_CRASH: get_scheduled_validator(${s})", ("s", slot));
                string scheduled_validator = db.get_scheduled_validator(slot);
                if (db._debug_block_production) ilog("DEBUG_CRASH: scheduled_validator=${w}", ("w", scheduled_validator));
                // we must control the validator scheduled to produce the next block.
                if (_validators.find(scheduled_validator) == _validators.end()) {
                    capture("scheduled_validator", scheduled_validator);
                    _last_scheduled_validator = scheduled_validator; // track for diagnostic
                    // Emergency master diagnostic: log when committee is configured but
                    // get_scheduled_validator returned a different name — reveals schedule misalignment
                    if (_validators.find(CHAIN_EMERGENCY_VALIDATOR_ACCOUNT) != _validators.end()) {
                        const auto &_dgp3 = db.get_dynamic_global_properties();
                        if (_dgp3.emergency_consensus_active) {
                            static fc::time_point _last_nmt_log;
                            auto _now3 = fc::time_point::now();
                            if ((_now3 - _last_nmt_log).count() > 3000000) { // log every 3s max (once per slot)
                                _last_nmt_log = _now3;
                                const auto &_wso3 = db.get_validator_schedule_object();
                                dlog("EMRG-DIAG not_my_turn: slot=${s} scheduled=${sw} head=#${h} aslot=${a} num_sched=${ns} aslot_mod=${am}",
                                    ("s", slot)
                                    ("sw", scheduled_validator)
                                    ("h", _dgp3.head_block_number)
                                    ("a", _dgp3.current_aslot)
                                    ("ns", _wso3.num_scheduled_validators)
                                    ("am", _dgp3.current_aslot % _wso3.num_scheduled_validators));
                            }
                        }
                    }
                    return block_validation_condition::not_my_turn;
                }

                if (db._debug_block_production) ilog("DEBUG_CRASH: looking up validator in index");
                const auto &validator_by_name = db.get_index<graphene::chain::validator_index>().indices().get<graphene::chain::by_name>();
                auto itr = validator_by_name.find(scheduled_validator);
                if (db._debug_block_production) ilog("DEBUG_CRASH: validator found=${f}", ("f", itr != validator_by_name.end()));

                fc::time_point_sec scheduled_time = db.get_slot_time(slot);
                // The header of the next block is validated against the key committed
                // at head; a pending validator_update may already be visible in this
                // index, so do not take the session-visible key at face value.
                graphene::protocol::public_key_type scheduled_key =
                        effective_signing_key(scheduled_validator, itr->signing_key);
                if (db._debug_block_production) ilog("DEBUG_CRASH: scheduled_key=${k}", ("k", scheduled_key));

                // Skip production if the scheduled slot time is at or before
                // the current head block time. This means the slot was already filled
                // by another block (e.g. received from P2P during/after a snapshot pause).
                // Without this guard, the validator produces a competing block at the same
                // height, creating a micro-fork that propagates to all peers.
                //
                // This can happen when:
                //   1. Snapshot pauses P2P processing for several seconds
                //   2. A block from another validator fills the slot during/after pause
                //   3. Our validator production loop fires for a slot that's now occupied
                if (scheduled_time <= db.head_block_time()) {
                    wlog("Skipping block production: scheduled slot ${st} is at or before "
                         "head_block_time ${hbt} (head=#${hn}). Slot was already filled.",
                         ("st", scheduled_time)("hbt", db.head_block_time())
                         ("hn", db.head_block_num()));
                    return block_validation_condition::not_time_yet;
                }

                // Check if validator has zero/null signing key (intentionally disabled for block production)
                if (scheduled_key == graphene::protocol::public_key_type()) {
                    if (scheduled_validator == CHAIN_EMERGENCY_VALIDATOR_ACCOUNT &&
                        _validators.find(CHAIN_EMERGENCY_VALIDATOR_ACCOUNT) != _validators.end()) {
                        static fc::time_point _last_zerokey_log;
                        auto _now_zk = fc::time_point::now();
                        if ((_now_zk - _last_zerokey_log).count() > 3000000) {
                            _last_zerokey_log = _now_zk;
                            dlog("EMRG-DIAG zero-key: committee scheduled at slot=${s} but signing_key is ZERO on chain! "
                                 "head=#${h} aslot=${a}",
                                 ("s", slot)("h", db.head_block_num())
                                 ("a", db.get_dynamic_global_properties().current_aslot));
                        }
                    } else if (_validators.count(scheduled_validator)) {
                        // Our configured validator is scheduled but its on-chain signing_key is zero.
                        // This means the chain blanked the key due to too many missed blocks
                        // (database.cpp update_global_dynamic_data).  Production is permanently
                        // blocked until the operator sends an update_validator transaction.
                        static fc::time_point _last_zerokey_regular_log;
                        auto _now_zkr = fc::time_point::now();
                        if ((_now_zkr - _last_zerokey_regular_log).count() > 60000000) {
                            _last_zerokey_regular_log = _now_zkr;
                            elog("Validator ${w} scheduled at slot=${s} but signing_key is ZERO on chain! "
                                 "Key was blanked due to too many missed blocks. "
                                 "Send update_validator transaction to re-enable. head=#${h}",
                                 ("w", scheduled_validator)("s", slot)("h", db.head_block_num()));
                        }
                    }
                    return block_validation_condition::not_my_turn;
                }

                auto private_key_itr = _private_keys.find(scheduled_key);

                if (private_key_itr == _private_keys.end()) {
                    capture("scheduled_validator", scheduled_validator);
                    capture("scheduled_key", scheduled_key);
                    return block_validation_condition::no_private_key;
                }

                // Pre-HF12 participation check (legacy behavior)
                if (!db.has_hardfork(CHAIN_HARDFORK_12)) {
                    uint32_t prate = db.validator_participation_rate();
                    if (prate < _required_validator_participation) {
                        if (_production_skip_flags & graphene::chain::database::skip_undo_history_check) {
                            dlog("Validator participation is ${p}% but stale-production is enabled, "
                                 "producing anyway to recover stalled network",
                                 ("p", uint32_t(prate / CHAIN_1_PERCENT)));
                        } else {
                            capture("pct", uint32_t(prate / CHAIN_1_PERCENT));
                            return block_validation_condition::low_participation;
                        }
                    }
                }

                if (llabs((scheduled_time - now).count()) > fc::milliseconds(500).count()) {
                    capture("scheduled_time", scheduled_time)("now", now);
                    {
                        static fc::time_point _last_lag_log;
                        auto _now_lag = fc::time_point::now();
                        if ((_now_lag - _last_lag_log).count() > 60000000) {
                            _last_lag_log = _now_lag;
                            wlog("Block production LAG: our slot for ${w} at ${st} but now=${now} "
                                 "(delta=${d}ms). Production loop fired too late for this slot. "
                                 "head=#${h}",
                                 ("w", scheduled_validator)("st", scheduled_time)("now", now)
                                 ("d", (scheduled_time - now).count() / 1000)("h", db.head_block_num()));
                        }
                    }
                    return block_validation_condition::lag;
                }

                // Check if a competing block already exists in the fork database for this block height.
                // Two-level fork collision resolution:
                //   Level 1: Vote-weighted comparison when both forks are in fork_db
                //   Level 2: Stuck-head timeout after one full validator round (21 blocks = 63s)
                // Not after HF16 (outside emergency): competing blocks are settled by QC/TC fork
                // choice and a double producer is punished by equivocation evidence. Deferring
                // here let ONE double-signing validator halt the chain: its twin at head+1 made
                // every producer defer, and the counter only grows on a node's own slot, so the
                // timeout needed 21 own slots (~7 min at 1 s), after which gap fill brought the
                // twin back (21-validator stand, qc-4 doubling viz2+viz3: head stuck at #81).
                if (emergency_active || !db.has_hardfork(CHAIN_HARDFORK_16)) {
                    auto existing_blocks = db.get_fork_db().fetch_block_by_number(db.head_block_num() + 1);
                    if (existing_blocks.size() > 0) {
                        bool has_competing_block = false;
                        graphene::chain::item_ptr competing_block;

                        if (emergency_active) {
                            // During emergency mode: ANY block at this height is competing.
                            // Multiple nodes with the emergency key may have produced.
                            // Defer to the deterministic hash-based resolution in fork_db.
                            has_competing_block = true;
                            competing_block = existing_blocks[0];
                        } else {
                            // Normal mode: only count blocks from different validators
                            // on a different parent as competing
                            for (const auto &eb : existing_blocks) {
                                if (eb->data.validator != scheduled_validator &&
                                    eb->data.previous != db.head_block_id()) {
                                    has_competing_block = true;
                                    competing_block = eb;
                                    break;
                                }
                            }
                        }

                        if (has_competing_block && competing_block) {
                            fork_collision_defer_count_++;

                            // LEVEL 2: Stuck-head timeout
                            // If we've been deferring and the head hasn't advanced, the competing
                            // block is from a dead fork. The network has moved on without it.
                            // After 21 consecutive deferrals (one full validator round = 63s),
                            // we can be sure the longer chain had all scheduled validators
                            // produce on it — confirming it's the canonical chain.
                            // This applies regardless of hardfork version — even pre-HF12
                            // nodes must not defer forever.
                            if (fork_collision_defer_count_ > _fork_collision_timeout_blocks) {
                                wlog("Fork collision timeout exceeded (${n} deferrals, head stuck at ${h}). "
                                     "Removing dead-fork competing block and producing on our chain.",
                                     ("n", fork_collision_defer_count_)("h", db.head_block_num()));
                                db.get_fork_db().remove_blocks_by_number(db.head_block_num() + 1);
                                fork_collision_defer_count_ = 0;
                                // Fall through to produce block
                            }
                            // LEVEL 1: Vote-weighted comparison (when both forks are in fork_db)
                            else if (db.has_hardfork(CHAIN_HARDFORK_12)) {
                                int weight_cmp = db.compare_fork_branches(
                                    competing_block->id, db.head_block_id());

                                if (weight_cmp < 0) {
                                    // Our fork has MORE vote weight -> produce on our fork
                                    wlog("Our fork has more vote weight at height ${h}. "
                                         "Producing despite competing block from weaker fork.",
                                         ("h", db.head_block_num() + 1));
                                    // Remove the losing competing block
                                    db.get_fork_db().remove(competing_block->id);
                                    fork_collision_defer_count_ = 0;
                                    // Fall through to produce block
                                } else if (weight_cmp > 0) {
                                    // Competing fork has MORE vote weight
                                    // Defer to let the fork switch happen naturally via _push_block.
                                    capture("height", db.head_block_num() + 1)("scheduled_validator", scheduled_validator);
                                    wlog("Competing fork at height ${h} has more vote weight. "
                                         "Deferring to allow fork switch to stronger chain.",
                                         ("h", db.head_block_num() + 1));
                                    return block_validation_condition::fork_collision;
                                } else {
                                    // Tied or comparison impossible (one tip not in fork_db)
                                    // Defer briefly, timeout will kick in
                                    capture("height", db.head_block_num() + 1)("scheduled_validator", scheduled_validator);
                                    wlog("Fork collision at height ${h} with tied/unknown vote weight. "
                                         "Deferring (attempt ${n}/${max}).",
                                         ("h", db.head_block_num() + 1)
                                         ("n", fork_collision_defer_count_)
                                         ("max", _fork_collision_timeout_blocks));
                                    return block_validation_condition::fork_collision;
                                }
                            }
                            // Pre-HF12: defer, but timeout still applies on next iteration
                            else {
                                capture("height", db.head_block_num() + 1)("scheduled_validator", scheduled_validator);
                                wlog("Fork collision at height ${h} (pre-HF12). "
                                     "Deferring (attempt ${n}/${max}).",
                                     ("h", db.head_block_num() + 1)
                                     ("n", fork_collision_defer_count_)
                                     ("max", _fork_collision_timeout_blocks));
                                return block_validation_condition::fork_collision;
                            }
                        }
                    }
                }

                // Release the operation guard before generate_block(), which
                // acquires its own guard internally via apply_pending_resize()
                // and with_strong_write_lock().
                op_guard.release();

                // Re-check snapshot pause: the gate at ~line 1133 passed before the
                // snapshot could have started (race window ~1 block interval).
                // If the snapshot began since then, _block_processing_paused is now
                // true and generate_block would immediately contend on the read lock
                // held by the snapshot thread, causing 2-11s write-lock starvation
                // (p67 incident).  Returning not_time_yet here costs one missed slot
                // (3 s) — far cheaper than the full snapshot read hold time.
                //
                // Check snapshot plugin directly for snapshot_in_progress flag.
                try {
                    if (snapshot().is_snapshot_in_progress()) {
                        dlog("Snapshot started between production checks for slot ${s}, skipping produce",
                             ("s", slot));
                        return block_validation_condition::not_time_yet;
                    }
                } catch (...) {}

                try {
                    if (p2p().is_catching_up_after_pause()) {
                        dlog("Snapshot started between production checks for slot ${s}, skipping produce",
                             ("s", slot));
                        return block_validation_condition::not_time_yet;
                    }
                } catch (...) {}

                // HF16: a handoff block needs the parent's QC. Without it we skip
                // the slot rather than fork: votes keep arriving, and the next
                // producer (or we, next slot) builds once the quorum is there.
                // If a TC for a round after head's QC is here instead, head was given up
                // on: drop it and build on its (certified) parent with the TC (Jolteon).
                // The chain accepts a TC on ANY round strictly between our parent's QC round
                // and ours (validate_block_qc), so we look for the newest such TC rather
                // than only for TC(own_round-1): a stalled chain forms TCs on whichever
                // rounds time out, and requiring exactly own_round-1 never matches once
                // timeouts have piled up (mainnet stall 31.10.2026: TC(245..372) formed,
                // own_round-1 was 600, and the chain stood still).
                fc::optional<graphene::protocol::block_timeout_certificate> skip_tc;
                // HF18 form of the same certificate: its votes name the block they certified.
                fc::optional<graphene::protocol::block_timeout_certificate_v2> skip_tc_v2;
                bool waiting_for_qc = false;
                bool rebuild_on_anchor = false;
                // Late-QC head: the block to rebuild on is the one its QC certifies.
                fc::optional<graphene::protocol::block_id_type> rebuild_on;
                const uint32_t own_round = db.round_of(scheduled_time);
                // A certified sibling of a QC-less head is taken before any TC is looked for:
                // a TC rebuild would add one more uncertified sibling nobody votes for.
                try {
                    // Read-lock pre-check: the write lock is taken only when a QC at head's
                    // height names another block, not on every production tick.
                    const bool sibling_qc = db.with_weak_read_lock([&]() {
                        if (!db.block_needs_parent_qc(scheduled_validator)) return false;
                        const auto qc = db.get_block_qc(db.head_block_num());
                        return qc && qc->block_id != db.head_block_id();
                    });
                    if (sibling_qc) db.with_strong_write_lock([&]() { db.switch_head_to_certified_sibling(); });
                } catch (const fc::exception &e) {
                    wlog("Switch to certified sibling failed: ${e}", ("e", e.to_string()));
                    return block_validation_condition::not_time_yet;
                }
                db.with_weak_read_lock([&]() {
                    if (!db.block_needs_parent_qc(scheduled_validator)) return;
                    const bool hf18 = db.has_hardfork(CHAIN_HARDFORK_18);
                    const auto parent_cert = db.parent_qc_for_next_block(scheduled_validator);
                    if (parent_cert.valid()) {
                        // Head certified. A gap after it needs a TC between its round and
                        // ours, or our block gets no votes. The certificate is checked
                        // against exactly this parent QC, block id included from HF18.
                        if (parent_cert->round + 1 >= own_round) return;
                        if (hf18)
                            skip_tc_v2 = db.get_timeout_cert_between_v2(parent_cert->round, own_round,
                                                                        parent_cert->round, parent_cert->block_id);
                        else
                            skip_tc = db.get_timeout_cert_between(parent_cert->round, own_round, parent_cert->round);
                        return;
                    }
                    waiting_for_qc = true;
                    // Diagnostics (stand, netem 300 ms: TCs formed for 63 rounds, head never
                    // dropped): say once per head why the TC skip is not taken.
                    // Keyed by head AND reason: keyed by head alone, the first refusal
                    // ("no TC yet") hid the real one for the rest of the stall.
                    const auto skip_refused = [&](const char *why) {
                        if (skip_diag_head == db.head_block_num() && skip_diag_why == why) return;
                        skip_diag_head = db.head_block_num();
                        skip_diag_why = why;
                        wlog("TC skip of QC-less head #${h} refused for slot round ${r}: ${why}",
                             ("h", db.head_block_num())("r", own_round)("why", why));
                    };
                    const auto head = db.fetch_block_by_number(db.head_block_num());
                    if (!head) return;
                    const graphene::protocol::block_qc_certificate *parent_qc = nullptr;
                    for (const auto &e : head->extensions)
                        if (e.which() == 3) parent_qc = &e.get<graphene::protocol::block_qc_certificate>();
                    // Head's parent is the HF16 activation block (the round anchor): head is
                    // exempt from a parent QC and so is its replacement. If head's round timed
                    // out (validators sent timeout votes before head arrived, so they will
                    // never vote for it), drop head and rebuild on the anchor with no TC.
                    // Without this the chain stalls for good right after activation
                    // (21-validator stand, 22.10: TC 15/15 every slot, head #67 forever).
                    if (!parent_qc) {
                        const auto &dgp = db.get_dynamic_global_properties();
                        const auto parent = db.fetch_block_by_id(head->previous);
                        if (parent && dgp.round_anchor_time != fc::time_point_sec() &&
                            parent->timestamp <= dgp.round_anchor_time) {
                            // Any TC proves some round after the anchor timed out; the TC
                            // itself is not carried (the rebuild block has no parent QC), so
                            // the block id does not matter here — the max round makes every
                            // vote "below the cut".
                            if (hf18)
                                skip_tc_v2 = db.get_timeout_cert_between_v2(0, own_round,
                                        std::numeric_limits<uint32_t>::max(), graphene::protocol::block_id_type());
                            else
                                skip_tc = db.get_timeout_cert_between(0, own_round, std::numeric_limits<uint32_t>::max());
                            rebuild_on_anchor = true;
                        }
                        return;
                    }
                    // The parent must be certified, and nobody who timed out may
                    // have seen a QC above it (else our block would be invalid); the
                    // window search already keeps only such certificates.
                    if (hf18) {
                        skip_tc_v2 = db.get_timeout_cert_between_v2(parent_qc->round, own_round,
                                                                    parent_qc->round, parent_qc->block_id);
                        if (!skip_tc_v2) { skip_refused("no usable TC between head's QC round and our round"); return; }
                    } else {
                        skip_tc = db.get_timeout_cert_between(parent_qc->round, own_round, parent_qc->round);
                        if (!skip_tc) { skip_refused("no usable TC between head's QC round and our round"); return; }
                    }
                    // Late-QC head (it carries its grandparent's QC because its parent's
                    // never formed): nothing between the certified block and head can get
                    // a QC any more, so after the TC the proposal extends the certified
                    // block itself, as Jolteon prescribes (extend high QC). Without this a
                    // late-QC head that also misses its own QC stalls the chain for good
                    // (21-validator stand, netem 300 ms: head #184 carried QC of #182).
                    if (parent_qc->block_id != head->previous) {
                        const auto certified = db.fetch_block_by_id(parent_qc->block_id);
                        if (!certified || certified->block_num() < db.get_dynamic_global_properties().last_irreversible_block_num ||
                            db.head_block_num() - certified->block_num() > 2) {
                            skip_refused("head's QC certifies a block that cannot be rebuilt on");
                            return;
                        }
                        rebuild_on = parent_qc->block_id;
                    }
                    // Head activated HF18: the block we would rebuild on is pre-fork, so the
                    // v2 TC cannot ride on it, and no v1 TC exists (everyone signed v2). Popping
                    // anyway leaves us on the pre-fork parent, where the next tick builds a
                    // TC-less block that gets no votes and activates HF18 again -- every slot,
                    // with a fresh head time, so the 60 s unstick never fires. Keep head and
                    // wait: the unstick builds on it.
                    // ponytail: timestamp test is exact only when the version quorum held before
                    // the HF time (mainnet 17/17); a late quorum falls back to the post-pop guard.
                    if (skip_tc_v2) {
                        const auto &hfp = db.get_hardfork_property_object();
                        const auto base = db.fetch_block_by_id(rebuild_on ? *rebuild_on : head->previous);
                        if (base && hfp.next_hardfork == CHAIN_HARDFORK_18_VERSION &&
                            base->timestamp < hfp.next_hardfork_time) {
                            skip_tc_v2.reset();
                            skip_refused("head activated HF18; its TC cannot ride on the pre-fork parent, waiting for the unstick");
                        }
                    }
                });
                if (waiting_for_qc && (skip_tc || skip_tc_v2)) {
                    const auto skipped = db.head_block_id();
                    const uint32_t skip_tc_round = skip_tc ? skip_tc->round : skip_tc_v2->round;
                    ilog("TC for round ${r}: dropping QC-less head #${h} ${id}, building slot ${s} as ${w} on its parent",
                         ("r", skip_tc_round)("h", db.head_block_num())("id", skipped)("s", slot)("w", scheduled_validator));
                    try {
                        db.with_strong_write_lock([&]() {
                            db.pop_block();
                            // Late-QC head: also drop its uncertified parent (at most one more).
                            if (rebuild_on && db.head_block_id() != *rebuild_on) db.pop_block();
                            FC_ASSERT(!rebuild_on || db.head_block_id() == *rebuild_on,
                                      "TC rebuild: head is not the certified block");
                        });
                    } catch (const fc::exception &e) {
                        wlog("Could not drop head ${id} for the TC: ${e}", ("id", skipped)("e", e.to_string()));
                        return block_validation_condition::not_time_yet;
                    }
                    // The skipped block may itself have activated HF18 (only reachable when the
                    // check above could not see it). A certificate selected on its state cannot
                    // be carried on the pre-fork parent.
                    if (skip_tc_v2 && db.with_weak_read_lock([&]() { return !db.has_hardfork(CHAIN_HARDFORK_18); }))
                        return block_validation_condition::not_time_yet;
                    // Our block now sits on the anchor: it needs no parent QC and may
                    // carry no TC ("A TC goes only next to a parent QC").
                    if (rebuild_on_anchor) { skip_tc.reset(); skip_tc_v2.reset(); }
                    waiting_for_qc = false;
                }
                // Late-QC fallback. `now` is whole seconds with a +250 ms look-ahead, so the
                // lag check above lets us produce until ~750 ms after the slot, on 250 ms
                // ticks; the last one lands ~500 ms in. There, rather than lose the slot,
                // carry the grandparent QC head already carries; finality slips one slot.
                fc::optional<graphene::protocol::block_qc_certificate> late_qc;
                if (waiting_for_qc && !skip_tc && !skip_tc_v2 &&
                    now_fine - fc::time_point(scheduled_time) >= fc::milliseconds(400)) {
                    late_qc = db.with_weak_read_lock([&]() {
                        return db.grandparent_qc_for_next_block(scheduled_validator, own_round);
                    });
                    if (late_qc)
                        ilog("Late QC: head #${h} still uncertified at the end of slot ${s}; ${w} carries QC of #${g} instead",
                             ("h", db.head_block_num())("s", slot)("w", scheduled_validator)("g", late_qc->block_num));
                }
                // HF16 unstick (emergency, owner 01.11.2026): no QC and no usable TC after a
                // minute of silence -- produce the block without a parent certificate rather
                // than wait for one the network evidently cannot deliver. The rule reads chain
                // time only (database::emergency_unstick_block), so the validators of this block
                // reach the same verdict.
                if (waiting_for_qc && !skip_tc && !skip_tc_v2 && !late_qc &&
                    db.with_weak_read_lock([&]() { return db.emergency_unstick_block(scheduled_time); })) {
                    wlog("Unstick: head #${h} uncertified for ${g}s, producing slot ${s} as ${w} without a parent QC",
                         ("h", db.head_block_num())("s", slot)("w", scheduled_validator)
                         ("g", (scheduled_time - db.head_block_time()).to_seconds()));
                    waiting_for_qc = false;
                }
                if (waiting_for_qc && !skip_tc && !skip_tc_v2 && !late_qc) {
                    ilog("Waiting for QC of head #${h} before producing slot ${s} as ${w}",
                         ("h", db.head_block_num())("s", slot)("w", scheduled_validator));
                    return block_validation_condition::not_time_yet;
                }

                if (db._debug_block_production) ilog("DEBUG_CRASH: calling generate_block for ${w}", ("w", scheduled_validator));
                if (scheduled_validator == CHAIN_EMERGENCY_VALIDATOR_ACCOUNT) {
                    dlog("EMRG-DIAG producing: slot=${s} scheduled_time=${st} head=#${h} aslot=${a}",
                         ("s", slot)("st", scheduled_time)("h", db.head_block_num())
                         ("a", db.get_dynamic_global_properties().current_aslot));
                }
                int retry = 0;
                do {
                    try {
                        // TODO: the same thread as used in chain-plugin,
                        //       but in the future it should refactored to calling of a chain-plugin function
                        auto block = db.generate_block(
                                scheduled_time,
                                scheduled_validator,
                                private_key_itr->second,
                                _production_skip_flags,
                                skip_tc,
                                late_qc,
                                skip_tc_v2
                        );
                        capture("n", block.block_num())("t", block.timestamp)("c", now)("w", scheduled_validator)("tx", block.transactions.size());
                        p2p().broadcast_block(block);

                        // If we produced a block but have few/no peers,
                        // force-reconnect seeds so the block can propagate.
                        // Skip in isolated-peers mode — seed reconnects are intentionally suppressed there.
                        auto peer_count = p2p().get_connections_count();
                        if (peer_count < 2 && !p2p().is_isolated_peers()) {
                            wlog("Produced block #${n} but only ${p} peer(s) connected — force-reconnecting seeds",
                                 ("n", block.block_num())("p", peer_count));
                            p2p().reconnect_seeds();
                        }

                        return block_validation_condition::produced;
                    }
                    catch (const graphene::chain::shared_memory_corruption_exception& e) {
                        elog("Shared memory corruption detected during block generation: ${e}", ("e", e.to_detail_string()));
                        chain().attempt_auto_recovery();
                        return block_validation_condition::exception_validating_block;
                    }
                    catch (const graphene::chain::unlinkable_block_exception& e) {
                        // Fork DB broken prev chain — retrying won't help.
                        // Roll back to LIB and resync from P2P network.
                        elog("unlinkable_block_exception during block generation: fork_db broken. "
                             "Rolling back to LIB and resyncing from P2P network.");
                        p2p().resync_from_lib(emergency_active /*force_emergency*/);
                        _minority_fork_recovering = true;
                        _minority_fork_recovery_start = fc::time_point::now();
                        return block_validation_condition::minority_fork;
                    }
                    catch (fc::exception &e) {
                        elog("${e}", ("e", e.to_detail_string()));
                        elog("Clearing pending transactions and attempting again");
                        db.clear_pending();
                        retry++;
                    }
                } while (retry < 2);

                return block_validation_condition::exception_validating_block;
            }
        }
    }
}