Retry and report transient-busy LedgerCloseMeta drops during catchup

crates/app/src/app/persist.rs

@@ -551,18 +551,15 @@ fn handle_persist_error(context: &str, error: &BucketError, shutdown: &Recoverab
 /// reclassified recoverable. Mirrors the `is_query_interrupted` shape in
 /// `henyey_db::queries` (matching the structured `ErrorCode`, NOT the message
 /// string).
+///
+/// The match itself now lives on the error type as
+/// [`henyey_db::DbError::is_transient_busy`] (#3801) so that `crates/history`
+/// — which cannot see this `pub(crate)` helper, and must not depend on
+/// `henyey-app` — shares ONE definition of "transient" instead of a second
+/// copy that could drift. This wrapper is retained so the ~20 existing call
+/// sites (and the `&anyhow::Error` sibling below) are unchanged.
 pub(crate) fn is_transient_db_busy(error: &henyey_db::DbError) -> bool {
-    matches!(
-        error,
-        henyey_db::DbError::Sqlite(rusqlite::Error::SqliteFailure(
-            rusqlite::ffi::Error {
-                code: rusqlite::ffi::ErrorCode::DatabaseBusy
-                    | rusqlite::ffi::ErrorCode::DatabaseLocked,
-                ..
-            },
-            _,
-        ))
-    )
+    error.is_transient_busy()
 }
 
 /// `&anyhow::Error` sibling of [`is_transient_db_busy`] for the ledger-close

crates/app/src/metrics.rs

@@ -293,7 +293,6 @@ macro_rules! metric_catalog {
 //
 // Reserved (NOT yet in the vocabulary — added by the PR that wires the
 // emitter, so no misleading always-zero series exists in between):
-//   - `catchup_meta` — `crates/history/src/catchup/persist.rs` `emit_meta` (#3801)
 //   - `tx_set_gc`    — `crates/herder/src/herder.rs` tx-set GC purge (#3806)
 
 /// Ledger-close persist DB commit (`commit_with_busy_retry`, bounded retry).
@@ -311,6 +310,15 @@ pub const SITE_PUBLISH_QUEUE_EVICT: &str = "publish_queue_evict";
 pub const SITE_PEER_FAILURE_PRUNE: &str = "peer_failure_prune";
 /// Peer-record update from the overlay peer-event loop (log-and-continue).
 pub const SITE_PEER_RECORD_UPDATE: &str = "peer_record_update";
+/// Catchup-replay `LedgerCloseMeta` row write (bounded retry, then
+/// log-and-continue — an RPC-visible hole catchup never revisits, #3801).
+///
+/// Emitted from `crates/history/src/catchup/persist.rs`, which cannot see the
+/// `record_db_busy_*` helpers (`app` → `history`, never the reverse) and so
+/// emits by literal counter name through the global `metrics` facade — the
+/// same split as the `stellar_history_*` counters above. This constant exists
+/// for the catalog/pre-registration side of that split.
+pub const SITE_CATCHUP_META: &str = "catchup_meta";
 
 /// The complete, closed `site` label vocabulary for the `henyey_db_busy_*`
 /// labelled counters. Pinned by `test_db_busy_site_label_vocabulary_pinned`.
@@ -322,6 +330,7 @@ pub const DB_BUSY_SITES: &[&str] = &[
     SITE_PUBLISH_QUEUE_EVICT,
     SITE_PEER_FAILURE_PRUNE,
     SITE_PEER_RECORD_UPDATE,
+    SITE_CATCHUP_META,
 ];
 
 // ── Metric catalog ─────────────────────────────────────────────────────
@@ -1008,7 +1017,7 @@ metric_catalog! {
                 retained unchanged for #3640 continuity — do NOT sum the two",
             "site", [SITE_LEDGER_CLOSE_PERSIST, SITE_CATCHUP_PERSIST, SITE_LEDGER_CLOSE_META,
                      SITE_RETENTION_TRIM, SITE_PUBLISH_QUEUE_EVICT, SITE_PEER_FAILURE_PRUNE,
-                     SITE_PEER_RECORD_UPDATE];
+                     SITE_PEER_RECORD_UPDATE, SITE_CATCHUP_META];
         DB_BUSY_WRITE_DROPPED_TOTAL = "henyey_db_busy_write_dropped_total"
             => "DB writes ultimately NOT applied because of a transient \
                 SQLITE_BUSY/LOCKED — retry-budget exhaustion, or a \
@@ -1023,7 +1032,7 @@ metric_catalog! {
                 is not a missing increment",
             "site", [SITE_LEDGER_CLOSE_PERSIST, SITE_CATCHUP_PERSIST, SITE_LEDGER_CLOSE_META,
                      SITE_RETENTION_TRIM, SITE_PUBLISH_QUEUE_EVICT, SITE_PEER_FAILURE_PRUNE,
-                     SITE_PEER_RECORD_UPDATE];
+                     SITE_PEER_RECORD_UPDATE, SITE_CATCHUP_META];
     }
 
     histograms {
@@ -2567,10 +2576,10 @@ mod tests {
         }
     }
 
-    /// #3802: pin the exact `site` vocabulary. #3801 (`catchup_meta`) and
-    /// #3806 (`tx_set_gc`) each add their value in the SAME PR that wires the
-    /// emitter, so this test failing is the intended, reviewed signal — not
-    /// drift. It also pins the label contract #3801's plan was written against.
+    /// #3802: pin the exact `site` vocabulary. #3806 (`tx_set_gc`) adds its
+    /// value in the SAME PR that wires the emitter, so this test failing is the
+    /// intended, reviewed signal — not drift. `catchup_meta` landed with its
+    /// emitter in #3801 (`crates/history/src/catchup/persist.rs`).
     #[test]
     fn test_db_busy_site_label_vocabulary_pinned() {
         assert_eq!(
@@ -2583,16 +2592,13 @@ mod tests {
                 "publish_queue_evict",
                 "peer_failure_prune",
                 "peer_record_update",
+                "catchup_meta",
             ],
             "the db-busy site vocabulary changed; update the catalog entries \
              and confirm the change is deliberate"
         );
 
         // Reserved values must NOT be present until their emitter lands.
-        assert!(
-            !DB_BUSY_SITES.contains(&"catchup_meta"),
-            "catchup_meta is reserved for #3801 and must land with its emitter"
-        );
         assert!(
             !DB_BUSY_SITES.contains(&"tx_set_gc"),
             "tx_set_gc is reserved for #3806 and must land with its emitter"

crates/db/src/error.rs

@@ -79,3 +79,115 @@ pub enum DbError {
     #[error("query exceeded computational budget")]
     QueryBudgetExceeded,
 }
+
+impl DbError {
+    /// Returns `true` iff this is a transient SQLite busy/locked
+    /// (`SQLITE_BUSY` / `SQLITE_LOCKED`, "database is locked"), #3497.
+    ///
+    /// SQLite write transactions are atomic: a `DatabaseBusy`/`DatabaseLocked`
+    /// means the transaction NEVER committed, so the on-disk state is
+    /// consistent and the write can be safely re-issued (or, at a
+    /// log-and-continue site, safely abandoned as a *known* loss rather than
+    /// treated as corruption). This is the recoverable, environmental class.
+    ///
+    /// The match is NARROW by design (the load-bearing consensus-safety
+    /// guard): only the two busy/locked primary `ErrorCode`s are recoverable.
+    /// Every other SQLite code (`DatabaseCorrupt`, `SystemIoFailure`, …) and
+    /// every other [`DbError`] variant (`Integrity`, `Xdr`, …) stay on the
+    /// fatal path — genuine corruption must NEVER be reclassified recoverable.
+    /// Mirrors the `is_query_interrupted` shape in `crate::queries` (matching
+    /// the structured `ErrorCode`, NOT the message string).
+    ///
+    /// Lives here, on the error type, rather than in a single consumer crate
+    /// so that every caller shares ONE definition of "transient": `crates/app`
+    /// (`is_transient_db_busy`, ledger-close/persist/maintenance) and
+    /// `crates/history` (catchup `emit_meta`, #3801) both delegate to it.
+    pub fn is_transient_busy(&self) -> bool {
+        matches!(
+            self,
+            DbError::Sqlite(rusqlite::Error::SqliteFailure(
+                rusqlite::ffi::Error {
+                    code: rusqlite::ffi::ErrorCode::DatabaseBusy
+                        | rusqlite::ffi::ErrorCode::DatabaseLocked,
+                    ..
+                },
+                _,
+            ))
+        )
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    /// Build a `DbError::Sqlite(SqliteFailure(ffi::Error { code, .. }, msg))`
+    /// for the given primary code, mirroring how rusqlite materializes a
+    /// SQLite error on a contended write.
+    fn sqlite_error(code: rusqlite::ffi::ErrorCode, msg: &str) -> DbError {
+        DbError::Sqlite(rusqlite::Error::SqliteFailure(
+            rusqlite::ffi::Error {
+                code,
+                extended_code: 0,
+            },
+            Some(msg.to_string()),
+        ))
+    }
+
+    /// Both busy/locked primary codes are the recoverable, environmental class.
+    #[test]
+    fn test_is_transient_busy_matches_busy_and_locked() {
+        assert!(
+            sqlite_error(rusqlite::ffi::ErrorCode::DatabaseBusy, "database is locked")
+                .is_transient_busy(),
+            "SQLITE_BUSY must classify as transient"
+        );
+        assert!(
+            sqlite_error(
+                rusqlite::ffi::ErrorCode::DatabaseLocked,
+                "database table is locked"
+            )
+            .is_transient_busy(),
+            "SQLITE_LOCKED must classify as transient"
+        );
+    }
+
+    /// The narrow-by-design boundary: genuine corruption and every non-SQLite
+    /// variant must NEVER be reclassified as recoverable.
+    #[test]
+    fn test_is_transient_busy_rejects_corrupt_and_non_sqlite() {
+        assert!(
+            !sqlite_error(
+                rusqlite::ffi::ErrorCode::DatabaseCorrupt,
+                "database disk image is malformed"
+            )
+            .is_transient_busy(),
+            "SQLITE_CORRUPT must stay on the fatal path"
+        );
+        assert!(
+            !sqlite_error(rusqlite::ffi::ErrorCode::SystemIoFailure, "disk I/O error")
+                .is_transient_busy(),
+            "SQLITE_IOERR must stay on the fatal path"
+        );
+        assert!(
+            !DbError::Integrity("bad hash".to_string()).is_transient_busy(),
+            "Integrity must stay on the fatal path"
+        );
+        assert!(
+            !DbError::Xdr(stellar_xdr::Error::Invalid).is_transient_busy(),
+            "Xdr must stay on the fatal path"
+        );
+        assert!(
+            !DbError::NotFound("row".to_string()).is_transient_busy(),
+            "NotFound must stay on the fatal path"
+        );
+        assert!(
+            !DbError::QueryBudgetExceeded.is_transient_busy(),
+            "QueryBudgetExceeded must stay on the fatal path"
+        );
+        assert!(
+            !DbError::Sqlite(rusqlite::Error::QueryReturnedNoRows).is_transient_busy(),
+            "a non-SqliteFailure rusqlite error must stay on the fatal path"
+        );
+    }
+}

crates/history/Cargo.toml

@@ -63,6 +63,10 @@ test-utils = ["dep:axum"]
 [dev-dependencies]
 tempfile.workspace = true
 axum.workspace = true
+# Local (thread-local) Prometheus recorder for asserting counter emission in
+# unit tests — see `metrics::with_local_recorder`. Test-only: the crate itself
+# emits fire-and-forget through the global `metrics` facade.
+metrics-exporter-prometheus.workspace = true
 # Shared deterministic tracing-capture test helper (replaces the deleted local
 # `tracing_test_support` module). See #3559.
 henyey-common = { workspace = true, features = ["test-support"] }

crates/history/src/catchup/mod.rs

@@ -65,6 +65,7 @@ use henyey_common::protocol::LclContext;
 use henyey_common::{Hash256, NetworkId};
 use henyey_db::Database;
 use std::collections::HashMap;
+use std::sync::atomic::{AtomicU32, Ordering};
 use std::sync::Arc;
 
 use henyey_ledger::{LedgerManager, TransactionSetVariant};
@@ -302,6 +303,17 @@ pub struct CatchupManager {
     /// `bucketListHash` is independent of this value (levels are assembled by
     /// index, not completion order), so capping is byte-for-byte state-preserving.
     pub(super) restore_apply_fan_out: Option<usize>,
+
+    /// Count of `ledger_close_meta` rows this manager failed to persist (#3801).
+    ///
+    /// Bumped by `emit_meta` on every abandoned meta write — transient-busy
+    /// exhaustion, non-transient DB error, or serialization failure alike. Each
+    /// increment is an RPC-visible hole catchup will never revisit, so the count
+    /// is surfaced on [`CatchupResult`] as a delta over the run. Atomic (rather
+    /// than `Cell`) because `emit_meta` takes `&self` and `CatchupManager` is
+    /// `Send + Sync`; `Relaxed` is sufficient — this is a pure counter with no
+    /// ordering relationship to any other state.
+    pub(super) meta_rows_dropped: AtomicU32,
 }
 
 impl CatchupManager {
@@ -324,6 +336,7 @@ impl CatchupManager {
             emit_meta_ext_v1: false,
             trusted_scp_anchor: None,
             restore_apply_fan_out: None,
+            meta_rows_dropped: AtomicU32::new(0),
         }
     }
 
@@ -344,6 +357,7 @@ impl CatchupManager {
             emit_meta_ext_v1: false,
             trusted_scp_anchor: None,
             restore_apply_fan_out: None,
+            meta_rows_dropped: AtomicU32::new(0),
         }
     }
 
@@ -352,6 +366,17 @@ impl CatchupManager {
         &self.progress
     }
 
+    /// Number of `ledger_close_meta` rows this manager has failed to persist
+    /// over its whole lifetime (#3801).
+    ///
+    /// Each one is a permanent, RPC-visible gap: catchup does not revisit a
+    /// ledger, so the row is never rewritten. [`CatchupResult::meta_rows_dropped`]
+    /// reports the per-run delta; this accessor exposes the running total for
+    /// callers that reuse one manager across runs.
+    pub fn meta_rows_dropped(&self) -> u32 {
+        self.meta_rows_dropped.load(Ordering::Relaxed)
+    }
+
     /// Set the replay configuration.
     pub fn set_replay_config(&mut self, config: ReplayConfig) {
         self.replay_config = config;
@@ -597,6 +622,10 @@ impl CatchupManager {
         buckets_downloaded: u32,
         ledger_manager: &LedgerManager,
     ) -> Result<CatchupResult> {
+        // #3801: snapshot before, subtract after, so the reported count is this
+        // run's meta loss even when one manager is reused across runs.
+        let meta_dropped_before = self.meta_rows_dropped.load(Ordering::Relaxed);
+
         let (final_seq, final_hash, ledgers_applied) = if checkpoint_seq >= target {
             // No replay needed — target is exactly at checkpoint.
             // §14.5 parity: like `persist_bucket_list_snapshot` above,
@@ -631,11 +660,17 @@ impl CatchupManager {
             final_seq, final_hash, ledgers_applied
         );
 
+        let meta_rows_dropped = self
+            .meta_rows_dropped
+            .load(Ordering::Relaxed)
+            .saturating_sub(meta_dropped_before);
+
         Ok(CatchupResult {
             ledger_seq: final_seq,
             ledger_hash: final_hash,
             ledgers_applied,
             buckets_downloaded,
+            meta_rows_dropped,
         })
     }