Implement clean index shutdown to prevent index corruption (with clip…

…py updates for Rust 1.71) (#2275)

* gracefully shutdown index update thread to prevent index corruption

* Use `next_back()` instead of `rev().next()` for rust 1.71

---------

Co-authored-by: victorkirov <victor.kirov@gmail.com>

Cargo.toml

@@ -24,7 +24,7 @@ bitcoin = { version = "0.29.1", features = ["rand"] }
 boilerplate = { version = "0.2.3", features = ["axum"] }
 chrono = "0.4.19"
 clap = { version = "3.2.18", features = ["derive", "deprecated"] }
-ctrlc = "3.2.1"
+ctrlc = { version = "3.2.1", features = ["termination"] }
 derive_more = "0.99.17"
 dirs = "5.0.0"
 env_logger = "0.10.0"

src/index.rs

@@ -347,8 +347,7 @@ impl Index {
         blocks_indexed: wtx
           .open_table(HEIGHT_TO_BLOCK_HASH)?
           .range(0..)?
-          .rev()
-          .next()
+          .next_back()
           .and_then(|result| result.ok())
           .map(|(height, _hash)| height.value() + 1)
           .unwrap_or(0),
@@ -739,8 +738,7 @@ impl Index {
         let current = tx
           .open_table(HEIGHT_TO_BLOCK_HASH)?
           .range(0..)?
-          .rev()
-          .next()
+          .next_back()
           .and_then(|result| result.ok())
           .map(|(height, _hash)| height)
           .map(|x| x.value())
@@ -811,7 +809,7 @@ impl Index {
     let inscription_number_to_inscription_id =
       rtx.open_table(INSCRIPTION_NUMBER_TO_INSCRIPTION_ID)?;
 
-    let latest = match inscription_number_to_inscription_id.iter()?.rev().next() {
+    let latest = match inscription_number_to_inscription_id.iter()?.next_back() {
       Some(Ok((number, _id))) => number.value(),
       Some(Err(_)) => return Ok(Default::default()),
       None => return Ok(Default::default()),

src/index/rtx.rs

@@ -9,8 +9,7 @@ impl Rtx<'_> {
         .0
         .open_table(HEIGHT_TO_BLOCK_HASH)?
         .range(0..)?
-        .rev()
-        .next()
+        .next_back()
         .and_then(|result| result.ok())
         .map(|(height, _hash)| Height(height.value())),
     )
@@ -22,8 +21,7 @@ impl Rtx<'_> {
         .0
         .open_table(HEIGHT_TO_BLOCK_HASH)?
         .range(0..)?
-        .rev()
-        .next()
+        .next_back()
         .and_then(|result| result.ok())
         .map(|(height, _hash)| height.value() + 1)
         .unwrap_or(0),
@@ -44,8 +42,7 @@ impl Rtx<'_> {
           .0
           .open_table(HEIGHT_TO_BLOCK_HASH)?
           .range(0..)?
-          .rev()
-          .next()
+          .next_back()
           .and_then(|result| result.ok())
           .map(|(_height, hash)| BlockHash::load(*hash.value())),
       ),

src/index/updater.rs

@@ -46,8 +46,7 @@ impl Updater {
     let height = wtx
       .open_table(HEIGHT_TO_BLOCK_HASH)?
       .range(0..)?
-      .rev()
-      .next()
+      .next_back()
       .and_then(|result| result.ok())
       .map(|(height, _hash)| height.value() + 1)
       .unwrap_or(0);
@@ -140,8 +139,7 @@ impl Updater {
         let height = wtx
           .open_table(HEIGHT_TO_BLOCK_HASH)?
           .range(0..)?
-          .rev()
-          .next()
+          .next_back()
           .and_then(|result| result.ok())
           .map(|(height, _hash)| height.value() + 1)
           .unwrap_or(0);

src/index/updater/inscription_updater.rs

@@ -70,8 +70,7 @@ impl<'a, 'db, 'tx> InscriptionUpdater<'a, 'db, 'tx> {
 
     let next_number = number_to_id
       .iter()?
-      .rev()
-      .next()
+      .next_back()
       .and_then(|result| result.ok())
       .map(|(number, _id)| number.value() + 1)
       .unwrap_or(0);
@@ -173,8 +172,7 @@ impl<'a, 'db, 'tx> InscriptionUpdater<'a, 'db, 'tx> {
           let seq_num = self
             .reinscription_id_to_seq_num
             .iter()?
-            .rev()
-            .next()
+            .next_back()
             .and_then(|result| result.ok())
             .map(|(_id, number)| number.value() + 1)
             .unwrap_or(0);

src/lib.rs

@@ -131,6 +131,7 @@ const CYCLE_EPOCHS: u64 = 6;
 
 static SHUTTING_DOWN: AtomicBool = AtomicBool::new(false);
 static LISTENERS: Mutex<Vec<axum_server::Handle>> = Mutex::new(Vec::new());
+static INDEXER: Mutex<Option<thread::JoinHandle<()>>> = Mutex::new(Option::None);
 
 fn integration_test() -> bool {
   env::var_os("ORD_INTEGRATION_TEST")
@@ -149,21 +150,32 @@ fn unbound_outpoint() -> OutPoint {
   }
 }
 
+fn gracefully_shutdown_indexer() {
+  if let Some(indexer) = INDEXER.lock().unwrap().take() {
+    // We explicitly set this to true to notify the thread to not take on new work
+    SHUTTING_DOWN.store(true, atomic::Ordering::Relaxed);
+    log::info!("Waiting for index thread to finish...");
+    if indexer.join().is_err() {
+      log::warn!("Index thread panicked; join failed");
+    }
+  }
+}
+
 pub fn main() {
   env_logger::init();
 
   ctrlc::set_handler(move || {
+    if SHUTTING_DOWN.fetch_or(true, atomic::Ordering::Relaxed) {
+      process::exit(1);
+    }
+
+    println!("Shutting down gracefully. Press <CTRL-C> again to shutdown immediately.");
+
     LISTENERS
       .lock()
       .unwrap()
       .iter()
       .for_each(|handle| handle.graceful_shutdown(Some(Duration::from_millis(100))));
-
-    println!("Shutting down gracefully. Press <CTRL-C> again to shutdown immediately.");
-
-    if SHUTTING_DOWN.fetch_or(true, atomic::Ordering::Relaxed) {
-      process::exit(1);
-    }
   })
   .expect("Error setting <CTRL-C> handler");
 
@@ -179,6 +191,11 @@ pub fn main() {
     {
       eprintln!("{}", err.backtrace());
     }
+
+    gracefully_shutdown_indexer();
+
     process::exit(1);
   }
+
+  gracefully_shutdown_indexer();
 }

src/subcommand/server.rs

@@ -126,13 +126,17 @@ pub(crate) struct Server {
 impl Server {
   pub(crate) fn run(self, options: Options, index: Arc<Index>, handle: Handle) -> Result {
     Runtime::new()?.block_on(async {
-      let clone = index.clone();
-      thread::spawn(move || loop {
-        if let Err(error) = clone.update() {
+      let index_clone = index.clone();
+      let index_thread = thread::spawn(move || loop {
+        if SHUTTING_DOWN.load(atomic::Ordering::Relaxed) {
+          break;
+        }
+        if let Err(error) = index_clone.update() {
           log::warn!("{error}");
         }
         thread::sleep(Duration::from_millis(5000));
       });
+      INDEXER.lock().unwrap().replace(index_thread);
 
       let config = options.load_config()?;
       let acme_domains = self.acme_domains()?;