Chain: gate operator-daemon startup on controller::is_synced()

libraries/chain/controller.cpp

@@ -4186,6 +4186,14 @@ size_t controller::fork_db_size() const {
    return my->fork_db_size();
 }
 
+bool controller::is_synced(fc::time_point now, fc::microseconds recency_window) const {
+   return fork_db_has_root() && fork_db_root().block_time() >= now - recency_window;
+}
+
+bool controller::is_synced() const {
+   return is_synced(fc::time_point::now(), fc::milliseconds(default_sync_recency_ms));
+}
+
 const dynamic_global_property_object& controller::get_dynamic_global_properties()const {
   return my->db.get<dynamic_global_property_object>();
 }

libraries/chain/include/sysio/chain/controller.hpp

@@ -358,6 +358,27 @@ namespace sysio::chain {
          // thread-safe
          size_t       fork_db_size() const;
 
+         /// Behind-now gap (ms) under which {@link is_synced} counts the node as synced.
+         /// Must exceed steady-state finality lag (a few blocks) plus scheduling slack, and
+         /// stay well under one epoch so consumers gating work on the predicate never start
+         /// against stale state. Not an operator tunable.
+         static constexpr uint32_t default_sync_recency_ms = 5000;
+
+         /**
+          * @brief True when the LAST IRREVERSIBLE block's time is within `recency_window`
+          *        of `now`; false while no irreversible root exists (fresh boot,
+          *        mid-replay, snapshot catch-up).
+          *
+          * @param now            Instant to measure against (wall-clock in production;
+          *                       explicit for testability).
+          * @param recency_window Maximum behind-`now` gap still considered synced.
+          */
+         // thread-safe
+         bool is_synced(fc::time_point now, fc::microseconds recency_window) const;
+         /// {@link is_synced} against wall-clock now at {@link default_sync_recency_ms}.
+         // thread-safe
+         bool is_synced() const;
+
          // thread-safe, retrieves block according to fork db best branch which can change at any moment
          signed_block_ptr fetch_block_by_number( uint32_t block_num )const;
          // thread-safe

plugins/batch_operator_plugin/src/batch_operator_plugin.cpp

@@ -19,6 +19,7 @@
 #include <sysio/batch_operator_plugin/outpost_opp_job.hpp>
 #include <sysio/depot/opreg_status.hpp>
 #include <sysio/chain/abi_serializer.hpp>
+#include <sysio/chain/plugin_interface.hpp>
 #include <sysio/chain/transaction.hpp>
 #include <sysio/chain_plugin/chain_plugin.hpp>
 #include <sysio/opp/opp.hpp>
@@ -187,6 +188,15 @@ struct batch_operator_plugin::impl {
    std::map<uint64_t, scheduled_opp_job_ids> scheduled_opp_jobs;
    std::atomic<bool>                 shutting_down{false};
 
+   /// Sync gate: `channels::irreversible_block` subscription that arms
+   /// {@link run_deferred_startup_or_quit} once `controller::is_synced()`
+   /// holds — a LIB advance is the only event that can turn the predicate
+   /// true. Unsubscribed after arming; otherwise the handle's destructor
+   /// releases it (no shutdown unsubscribe needed: posted channel deliveries
+   /// are drained without executing after quit, and the slot checks
+   /// `shutting_down` first).
+   chain::plugin_interface::channels::irreversible_block::channel_type::handle sync_gate_subscription;
+
    // -----------------------------------------------------------------------
    //  Chain-agnostic orchestration layer
    // -----------------------------------------------------------------------
@@ -812,6 +822,91 @@ struct batch_operator_plugin::impl {
          elog("batch_operator: push {}::{} timed out", contract, action_name);
       }
    }
+
+   // -----------------------------------------------------------------------
+   //  Sync-gated startup
+   // -----------------------------------------------------------------------
+
+   /// The startup body deferred behind the sync gate: outpost
+   /// discovery → private cron_service creation (sized from the discovered
+   /// outposts) → epoch_tick scheduling → per-outpost relay jobs. Runs on the
+   /// main thread from {@link run_deferred_startup_or_quit} once the node is
+   /// synced. Deferral exists because `refresh_outposts` reads `sysio.chains`
+   /// LOCALLY: on a cold-booting operator node still replaying toward the
+   /// deploy blocks the read throws Account/Contract Query Exceptions
+   /// (3060002/3060003) — spurious boot-window errors the gate removes.
+   void run_deferred_startup() {
+      if (shutting_down) {
+         return;
+      }
+
+      // Discover outposts before the private cron_service starts. Later refresh
+      // ticks add/remove per-outpost cron jobs as the active chain set changes.
+      try {
+         refresh_outposts();
+      } catch (const fc::exception& e) {
+         wlog("batch_operator_plugin: initial outpost discovery failed: {}. "
+              "Starting with 0 per-outpost jobs; refresh ticks will retry after "
+              "the chain has caught up.", e.to_string());
+      }
+
+      // Size the pool for startup outposts. Later dynamic outposts are added to
+      // the same queued cron service; the minimum keeps epoch_tick viable even
+      // when no outposts are known yet.
+      const std::size_t outpost_count   = opp_jobs.size();
+      const std::size_t required_threads = outpost_count * OPP_CRON_JOBS_PER_OUTPOST + EPOCH_TICK_CRON_JOBS;
+      const std::size_t thread_count    = std::max(required_threads, MIN_CRON_THREADS);
+
+      sysio::services::cron_service::options svc_opts;
+      svc_opts.name        = "batch_operator";
+      svc_opts.num_threads = thread_count;
+      svc_opts.autostart   = true;
+
+      cron_svc = sysio::services::cron_service::create(svc_opts);
+      ilog("batch_operator_plugin: cron_service started with {} thread(s) ({} outpost(s) discovered)",
+           thread_count, outpost_count);
+
+      const auto poll_ms = epoch_poll_ms;
+
+      // epoch_tick — refresh epoch state + election. Keeps `current_epoch`
+      // and `within_epoch_window` accurate for every per-outpost job.
+      {
+         sysio::services::cron_service::job_schedule sched;
+         sched.milliseconds = {sysio::services::cron_service::job_schedule::step_value{poll_ms}};
+         sysio::services::cron_service::job_metadata_t meta;
+         meta.label          = "batch_operator_epoch_tick";
+         meta.one_at_a_time  = true;
+         auto id = cron_svc->add(sched,
+                                 [this]() { poll_epoch_state(); },
+                                 meta);
+         cron_job_ids.push_back(id);
+         ilog("batch_operator_plugin: scheduled {} (id={}, every {}ms)", meta.label, id, poll_ms);
+      }
+
+      schedule_opp_jobs();
+   }
+
+   /// {@link run_deferred_startup} plus the uniform fail-fast policy: the
+   /// sync-gate callback is a posted channel delivery, so an escaping
+   /// exception would unwind the application executor mid-task with no
+   /// diagnosable trace of WHAT failed. Contain it long enough to log, then
+   /// shut the node down — an operator daemon whose relay never started must
+   /// be supervisor-visible (it has liveness/slashing consequences), not
+   /// hidden behind a running process. Expected transient failures
+   /// (outpost discovery against a not-yet-deployed registry) are already
+   /// absorbed inside {@link run_deferred_startup} and retried by the refresh
+   /// ticks; what reaches here is structural (cron_service creation or job
+   /// scheduling failed). FC_LOG_AND_DROP deliberately rethrows
+   /// boost::interprocess::bad_alloc — chainbase shared-memory exhaustion
+   /// stays immediately fatal.
+   void run_deferred_startup_or_quit() {
+      try {
+         run_deferred_startup();
+         return;
+      } FC_LOG_AND_DROP("batch_operator_plugin: deferred startup failed unexpectedly:");
+      elog("batch_operator_plugin: deferred startup failed terminally — shutting down node (fail-fast)");
+      app().quit();
+   }
 };
 
 // ---------------------------------------------------------------------------
@@ -872,6 +967,17 @@ void batch_operator_plugin::plugin_initialize(const variables_map& options) {
    _impl->cron_plug  = &app().get_plugin<cron_plugin>();
    _impl->eth_plug   = &app().get_plugin<outpost_ethereum_client_plugin>();
    _impl->sol_plug   = &app().get_plugin<outpost_solana_client_plugin>();
+
+   // Operator daemons are designed for read-mode = irreversible: the sync gate
+   // (controller::is_synced) measures LIB recency and every local table read the
+   // relay performs serves the irreversible view. Any other read mode would relay
+   // envelopes derived from state that can still fork out. Together with
+   // producer_plugin's inverse assert (no producer-name under irreversible
+   // read-mode), this also makes co-hosting a producer with an operator daemon
+   // impossible by configuration.
+   FC_ASSERT(!_impl->enabled ||
+                _impl->chain_plug->chain().get_read_mode() == chain::db_read_mode::IRREVERSIBLE,
+             "batch_operator_plugin requires read-mode = irreversible");
 }
 
 void batch_operator_plugin::plugin_startup() {
@@ -882,50 +988,49 @@ void batch_operator_plugin::plugin_startup() {
 
    ilog("batch_operator_plugin: starting for account {}", _impl->operator_account.to_string());
 
-   // Discover outposts before the private cron_service starts. Later refresh
-   // ticks add/remove per-outpost cron jobs as the active chain set changes.
-   try {
-      _impl->refresh_outposts();
-   } catch (const fc::exception& e) {
-      wlog("batch_operator_plugin: initial outpost discovery failed: {}. "
-           "Starting with 0 per-outpost jobs; refresh ticks will retry after "
-           "the chain has caught up.", e.to_string());
-   }
-
-   // Size the pool for startup outposts. Later dynamic outposts are added to
-   // the same queued cron service; the minimum keeps epoch_tick viable even
-   // when no outposts are known yet.
-   const std::size_t outpost_count   = _impl->opp_jobs.size();
-   const std::size_t required_threads = outpost_count * OPP_CRON_JOBS_PER_OUTPOST + EPOCH_TICK_CRON_JOBS;
-   const std::size_t thread_count    = std::max(required_threads, MIN_CRON_THREADS);
-
-   sysio::services::cron_service::options svc_opts;
-   svc_opts.name        = "batch_operator";
-   svc_opts.num_threads = thread_count;
-   svc_opts.autostart   = true;
-
-   _impl->cron_svc = sysio::services::cron_service::create(svc_opts);
-   ilog("batch_operator_plugin: cron_service started with {} thread(s) ({} outpost(s) discovered)",
-        thread_count, outpost_count);
-
-   const auto poll_ms = _impl->epoch_poll_ms;
-
-   // epoch_tick — refresh epoch state + election. Keeps `current_epoch`
-   // and `within_epoch_window` accurate for every per-outpost job.
-   {
-      sysio::services::cron_service::job_schedule sched;
-      sched.milliseconds = {sysio::services::cron_service::job_schedule::step_value{poll_ms}};
-      sysio::services::cron_service::job_metadata_t meta;
-      meta.label          = "batch_operator_epoch_tick";
-      meta.one_at_a_time  = true;
-      auto id = _impl->cron_svc->add(sched,
-                                     [impl = _impl.get()]() { impl->poll_epoch_state(); },
-                                     meta);
-      _impl->cron_job_ids.push_back(id);
-      ilog("batch_operator_plugin: scheduled {} (id={}, every {}ms)", meta.label, id, poll_ms);
-   }
-
-   _impl->schedule_opp_jobs();
+   // The startup body's outpost discovery reads `sysio.chains` LOCALLY. On a
+   // cold-booting operator node those reads see mid-sync (possibly genesis)
+   // state and throw spuriously, so the whole body (discovery → cron_service →
+   // epoch_tick → relay jobs) is DEFERRED until the node is synced —
+   // `controller::is_synced()`: the LAST IRREVERSIBLE block's time within
+   // `controller::default_sync_recency_ms` of now (the state the reads
+   // actually serve under read-mode = irreversible). The wake-up is the
+   // existing `irreversible_block` channel: a LIB advance is the only event
+   // that can turn the predicate true, and channel deliveries are posted to
+   // the application executor — main thread, AFTER the triggering block fully
+   // commits — so the callback may run the startup body directly. There is
+   // deliberately no already-synced fast path: operator daemons boot with
+   // genesis-stale LIB in every deployment topology (producer co-hosting is
+   // impossible by configuration — see the read-mode requirement in
+   // plugin_initialize), and a node that somehow is synced at startup is
+   // released by the next LIB advance. That advance is also the relay's WORK
+   // SUPPLY — everything here acts on newly-finalized state — so a finality
+   // stall that spans startup (the classic lost-wakeup case) is one with
+   // nothing to relay: the first finalized block re-arms the gate and
+   // creates the first actionable state at the same instant.
+   auto& chain = _impl->chain_plug->chain();
+   ilog("batch_operator_plugin: waiting for chain sync before outpost discovery "
+        "(head {} is {}s behind now; irreversible state is {}s behind)",
+        chain.head().block_num(),
+        (fc::time_point::now() - chain.head().block_time()).to_seconds(),
+        chain.fork_db_has_root()
+           ? std::to_string((fc::time_point::now() - chain.fork_db_root().block_time()).to_seconds())
+           : "n/a");
+   _impl->sync_gate_subscription =
+      app().get_channel<chain::plugin_interface::channels::irreversible_block>().subscribe(
+         [impl = _impl.get()](const chain::block_signal_params&) {
+            if (impl->shutting_down || !impl->chain_plug->chain().is_synced()) {
+               return;
+            }
+            // One-shot consumption: unsubscribe (safe from within the slot) and
+            // run the startup body directly — channel deliveries are posted to
+            // the application executor, so this already runs on the main thread
+            // AFTER the triggering block committed (mid block-application,
+            // table reads would observe an incomplete view).
+            impl->sync_gate_subscription.unsubscribe();
+            ilog("batch_operator_plugin: chain synced — starting deferred startup");
+            impl->run_deferred_startup_or_quit();
+         });
 }
 
 void batch_operator_plugin::plugin_shutdown() {

plugins/underwriter_plugin/include/sysio/underwriter_plugin/sync_detail.hpp

@@ -1,17 +1,19 @@
 #pragma once
 /**
  * @file sync_detail.hpp
- * @brief Pure sync-gate predicate for the underwriter plugin, lifted out of the
- *        `.cpp`-private impl so it is unit-testable without standing up a chain.
+ * @brief Pure gate-lifecycle machinery for the underwriter plugin — the deferred-
+ *        startup states and the `/v1/underwriter/*` gate payloads — lifted out of
+ *        the `.cpp`-private impl so they are unit-testable without standing up a
+ *        chain.
  *
  * The underwriter's startup preflight validates depot-side state (opreg
  * registration, chain registry, authex links) via LOCAL table reads. On a
  * cold-booting operator node those reads see mid-sync (possibly genesis)
  * state and fail spuriously, so the plugin must not begin underwriting until
- * the node is synced. "Synced" here means the LAST IRREVERSIBLE block's time
- * is within a small window of wall-clock now — the irreversible state is what
- * the plugin's table reads actually serve (operator daemons run read-mode =
- * irreversible).
+ * the node is synced. The sync predicate itself is `controller::is_synced()`
+ * (LIB recency, woken by the `irreversible_block` channel), shared by every
+ * operator-daemon plugin; this header carries only the underwriter's OWN gate
+ * lifecycle surface.
  */
 
 #include <fc/time.hpp>
@@ -23,26 +25,6 @@
 
 namespace sysio::underwriter_detail {
 
-/**
- * @brief True when a block time is recent enough to treat the node as synced.
- *
- * While a cold-booting node syncs blocks the tested time trails `now` by the
- * catch-up gap; once caught up, it tracks `now` to within finality-lag
- * jitter. The window must exceed that jitter (plus scheduling slack) but stay
- * well under one epoch so underwriting never starts against stale state.
- *
- * @param block_time     The block timestamp to test — the caller's sync
- *                       criterion (the underwriter feeds the LAST IRREVERSIBLE
- *                       block's time, since that is the state its reads serve).
- * @param now            Wall-clock now.
- * @param recency_window Maximum behind-now gap still considered synced.
- * @return True when `block_time >= now - recency_window`.
- */
-inline bool block_time_is_recent(fc::time_point block_time, fc::time_point now,
-                                 fc::microseconds recency_window) {
-   return block_time >= now - recency_window;
-}
-
 /// JSON field keys shared by the gate payloads below, the post-startup
 /// response builders in `underwriter_plugin.cpp`, and their tests — one
 /// spelling authority instead of per-translation-unit string literals.
@@ -73,13 +55,30 @@ enum class startup_state : uint8_t {
    wiring_failed,      ///< Preflight passed but outpost client wiring threw (terminal).
    startup_failed,     ///< The deferred startup body threw past the specific
                        ///< failure paths above (terminal). Exists so an escaping
-                       ///< exception is contained as a diagnosable gate state
-                       ///< instead of unwinding the app executor — which would
-                       ///< tear down the whole node — with the gate still
-                       ///< claiming `waiting_for_sync`.
+                       ///< exception is contained as a diagnosable gate state —
+                       ///< stored and logged BEFORE the fail-fast shutdown — instead
+                       ///< of unwinding the app executor mid-task with the gate
+                       ///< still claiming `waiting_for_sync`.
    active              ///< Deferred startup completed — the scan cron is scheduled.
 };
 
+/**
+ * @brief True for the terminal deferred-startup FAILURE states — the states the
+ *        uniform fail-fast policy shuts the node down on (see
+ *        `underwriter_plugin.cpp`'s `quit_if_startup_failed_terminally`).
+ *
+ * `active` is terminal success, not failure; `preflight_retrying` is transient by
+ * definition (bounded by the retry grace). Kept beside the enum so a new state
+ * cannot be added without deciding its fail-fast classification here.
+ *
+ * @param s The deferred-startup lifecycle state to classify.
+ * @return True when `s` is a terminal failure state.
+ */
+inline constexpr bool is_terminal_failure(startup_state s) {
+   return s == startup_state::preflight_failed || s == startup_state::wiring_failed ||
+          s == startup_state::startup_failed;
+}
+
 /// The wire `status` spelling of {@link startup_state::active} — shared by the
 /// post-startup response builders so the value has one authority beside the
 /// enum whose member spellings define the wire format.
@@ -91,15 +90,15 @@ static_assert(active_status == "active");
 /// `detail` carried by the {@link startup_state::preflight_failed} gate payload.
 inline constexpr std::string_view preflight_failed_detail =
    "startup preflight failed after sync — depot-side state for this underwriter "
-   "is incomplete; underwriting is disabled (see node log)";
+   "is incomplete; the node is shutting down (fail-fast; see node log)";
 /// `detail` carried by the {@link startup_state::wiring_failed} gate payload.
 inline constexpr std::string_view wiring_failed_detail =
-   "outpost client wiring failed after preflight; underwriting is disabled "
-   "(see node log)";
+   "outpost client wiring failed after preflight; the node is shutting down "
+   "(fail-fast; see node log)";
 /// `detail` carried by the {@link startup_state::startup_failed} gate payload.
 inline constexpr std::string_view startup_failed_detail =
-   "deferred startup failed unexpectedly; underwriting is disabled "
-   "(see node log)";
+   "deferred startup failed unexpectedly; the node is shutting down "
+   "(fail-fast; see node log)";
 
 /**
  * @brief The JSON body a `/v1/underwriter/...` endpoint answers with for `state`.