fix(node): implement reconciliation sweep as durability backstop (#218)

Author: Gravirei

Cargo.lock

@@ -1,8 +1,9 @@
+use std::time::Duration;
+
 use anyhow::{Context, Result};
 use chrono::{DateTime, Utc};
 use serde::{Deserialize, Serialize};
 use sqlx::{postgres::PgPoolOptions, PgPool, Row};
-use std::time::Duration;
 use tracing::info;
 use uuid::Uuid;
 

crates/gitlawb-node/src/db/mod.rs

@@ -16,6 +16,7 @@ mod operator;
 mod p2p;
 mod pinata;
 mod rate_limit;
+mod reconciliation;
 mod server;
 mod state;
 mod sync;
@@ -498,6 +499,19 @@ async fn main() -> Result<()> {
         info!("auto-sync worker started");
     }
 
+    // Periodic reconciliation sweep: re-derives pin/seal sets and fills gaps
+    // so a dropped replication job never means data loss.
+    {
+        let db = state.db.clone();
+        let config = Arc::clone(&state.config);
+        let http_client = Arc::clone(&state.http_client);
+        let node_keypair = Arc::clone(&state.node_keypair);
+        let node_did = state.node_did.clone();
+        let shutdown_rx = state.subscribe_shutdown();
+        reconciliation::spawn(db, config, http_client, node_keypair, node_did, shutdown_rx);
+        info!("reconciliation sweep worker started");
+    }
+
     // On-chain operator setup: verify stake + spawn heartbeat loop
     if !state.config.contract_node_staking.is_empty()
         && !state.config.operator_private_key.is_empty()

crates/gitlawb-node/src/main.rs

@@ -15,6 +15,9 @@
 //!     `gitlawb_pack_size_bytes`
 //!   * a single `gitlawb_info{version, did}` gauge = 1, for joins/dashboards
 //!   * currently-connected peer count — `gitlawb_peers_connected`
+//!   * reconciliation sweep gaps found and filled —
+//!     `gitlawb_reconciliation_gaps_found_total` /
+//!     `gitlawb_reconciliation_gaps_filled_total`
 //!
 //! All metrics live in a single process-wide registry initialized by
 //! [`init`]. Increment helpers (`record_push`, `record_auth_failure`, ...)
@@ -33,8 +36,8 @@
 use std::sync::OnceLock;
 
 use prometheus::{
-    Encoder, Histogram, HistogramOpts, IntCounterVec, IntGauge, IntGaugeVec, Opts, Registry,
-    TextEncoder,
+    Encoder, Histogram, HistogramOpts, IntCounter, IntCounterVec, IntGauge, IntGaugeVec, Opts,
+    Registry, TextEncoder,
 };
 
 /// The single, process-wide metrics registry. Initialized by [`init`].
@@ -51,6 +54,8 @@ static SYNC_PROCESSED: OnceLock<IntCounterVec> = OnceLock::new();
 static WEBHOOK_DELIVERIES: OnceLock<IntCounterVec> = OnceLock::new();
 static PACK_SIZE: OnceLock<Histogram> = OnceLock::new();
 static PEERS_CONNECTED: OnceLock<IntGauge> = OnceLock::new();
+static RECONCILIATION_GAPS_FOUND: OnceLock<IntCounter> = OnceLock::new();
+static RECONCILIATION_GAPS_FILLED: OnceLock<IntCounter> = OnceLock::new();
 
 /// One-time initializer. Builds the registry, registers every metric,
 /// and sets the constant `gitlawb_info` gauge. Idempotent — calling
@@ -197,6 +202,30 @@ pub fn init(version: &str, node_did: &str) {
         .set(peers_connected)
         .expect("set PEERS_CONNECTED once");
 
+    let gaps_found = IntCounter::with_opts(Opts::new(
+        "gitlawb_reconciliation_gaps_found_total",
+        "Total reconciliation sweep gaps detected (objects that should be pinned but are not)",
+    ))
+    .expect("gitlawb_reconciliation_gaps_found_total definition");
+    registry
+        .register(Box::new(gaps_found.clone()))
+        .expect("register gitlawb_reconciliation_gaps_found_total");
+    RECONCILIATION_GAPS_FOUND
+        .set(gaps_found)
+        .expect("set RECONCILIATION_GAPS_FOUND once");
+
+    let gaps_filled = IntCounter::with_opts(Opts::new(
+        "gitlawb_reconciliation_gaps_filled_total",
+        "Total reconciliation sweep gaps successfully filled (objects pinned by the sweep)",
+    ))
+    .expect("gitlawb_reconciliation_gaps_filled_total definition");
+    registry
+        .register(Box::new(gaps_filled.clone()))
+        .expect("register gitlawb_reconciliation_gaps_filled_total");
+    RECONCILIATION_GAPS_FILLED
+        .set(gaps_filled)
+        .expect("set RECONCILIATION_GAPS_FILLED once");
+
     REGISTRY
         .set(registry)
         .expect("set REGISTRY once (init must be called exactly once)");
@@ -264,6 +293,20 @@ pub fn set_peers_connected(count: i64) {
     }
 }
 
+/// Record reconciliation sweep gaps found (objects that should be pinned but are not).
+pub fn record_reconciliation_gaps_found(count: u64) {
+    if let Some(c) = RECONCILIATION_GAPS_FOUND.get() {
+        c.inc_by(count);
+    }
+}
+
+/// Record reconciliation sweep gaps filled (objects successfully pinned by the sweep).
+pub fn record_reconciliation_gaps_filled(count: u64) {
+    if let Some(c) = RECONCILIATION_GAPS_FILLED.get() {
+        c.inc_by(count);
+    }
+}
+
 /// Encode the registry as the standard Prometheus text exposition format.
 /// Returns an error if `init` was never called.
 pub fn encode() -> Result<String, prometheus::Error> {