feat: Shutdown RPC endpoint (#2538)

CHANGELOG.md

@@ -4,11 +4,13 @@ Notable updates:
 
 ### Added
 * [database] added ParityDb statistics to the stats endpoint. [#2444](https://github.com/ChainSafe/forest/pull/2444)
+* [api|cli] Add RPC `Filecoin.Shutdown` endpoint and `forest-cli shutdown` subcommand. [#2538](https://github.com/ChainSafe/forest/pull/2538)
 * [cli] A JavaScript console to interact with Filecoin API. [#2492](https://github.com/ChainSafe/forest/pull/2492)
 * [docker] Multi-platform Docker image support. [#2552](https://github.com/ChainSafe/forest/pull/2552)
 * [forest daemon] Added `--exit-after-init` and `--save-token` flags. [#2577](https://github.com/ChainSafe/forest/pull/2577)
 
 ### Changed
+* [cli] Remove Forest ctrl-c hard shutdown behavior on subsequent ctrl-c signals. [#2538](https://github.com/ChainSafe/forest/pull/2538)
 * [libp2p] Use in house bitswap implementation. [#2445](https://github.com/ChainSafe/forest/pull/2445)
 * [libp2p] Ban peers with duration. Banned peers are automatically unbanned after a period of 1h. [#2396](https://github.com/ChainSafe/forest/pull/2396)
 * [libp2p] Support multiple listen addr. [#2570](https://github.com/ChainSafe/forest/pull/2570)

documentation/src/js_console.md

@@ -76,6 +76,10 @@ Forest console comes with a number of helper functions that make interacting wit
  - `showSyncStatus()`
  - `sendFIL(to, attoAmount)`
 
+### Timers
+
+In addition, to support part of the JavaScript language, the console also provides implementation for `sleep(seconds)` timer and a tipset based timer, `sleepTipsets(epochs)`, which sleeps till the number of new tipsets added is equal to or greater than `epochs`.
+
 ### Modules
 
 CommonJS modules is the way to package JavaScript code for Forest console. You can import modules using the `require` function:

forest/cli/src/cli/attach_cmd.rs

@@ -16,15 +16,17 @@ use boa_engine::{
 };
 use convert_case::{Case, Casing};
 use directories::BaseDirs;
+use forest_chain_sync::SyncStage;
 use forest_json::message::json::MessageJson;
 use forest_rpc_api::mpool_api::MpoolPushMessageResult;
 use forest_rpc_client::*;
 use forest_shim::{address::Address, econ::TokenAmount, message::Message_v3};
-use fvm_shared::METHOD_SEND;
+use fvm_shared::{clock::ChainEpoch, METHOD_SEND};
 use num::BigInt;
 use rustyline::{config::Config as RustyLineConfig, EditMode, Editor};
 use serde::Serialize;
 use serde_json::Value as JsonValue;
+use tokio::time;
 
 use super::Config;
 
@@ -113,7 +115,6 @@ fn require(
     let result = if path.exists() {
         read_to_string(path.clone())
     } else if path == PathBuf::from("prelude.js") {
-        //println!("load builtin prelude");
         Ok(PRELUDE_PATH.into())
     } else {
         return context.throw_error("expecting valid module path");
@@ -235,6 +236,42 @@ async fn send_message(
     mpool_push_message((json_message, None), auth_token).await
 }
 
+type SleepParams = (u64,);
+type SleepResult = ();
+
+async fn sleep(
+    params: SleepParams,
+    _auth_token: &Option<String>,
+) -> Result<SleepResult, jsonrpc_v2::Error> {
+    let secs = params.0;
+    time::sleep(time::Duration::from_secs(secs)).await;
+    Ok(())
+}
+
+type SleepTipsetsParams = (ChainEpoch,);
+type SleepTipsetsResult = ();
+
+async fn sleep_tipsets(
+    params: SleepTipsetsParams,
+    auth_token: &Option<String>,
+) -> Result<SleepTipsetsResult, jsonrpc_v2::Error> {
+    let mut epoch = None;
+    loop {
+        let state = sync_status((), auth_token).await?;
+        if state.active_syncs[0].stage() == SyncStage::Complete {
+            if let Some(prev) = epoch {
+                let curr = state.active_syncs[0].epoch();
+                if (curr - prev) >= params.0 {
+                    return Ok(());
+                }
+            } else {
+                epoch = Some(state.active_syncs[0].epoch());
+            }
+        }
+        time::sleep(time::Duration::from_secs(1)).await;
+    }
+}
+
 impl AttachCommand {
     fn setup_context(&self, context: &mut Context, token: &Option<String>) {
         // Disable tracing
@@ -283,8 +320,14 @@ impl AttachCommand {
         // Message Pool API
         bind_func!(context, token, mpool_push_message);
 
-        // Bind send_message
+        // Common API
+        bind_func!(context, token, version);
+        bind_func!(context, token, shutdown);
+
+        // Bind send_message, sleep, sleep_tipsets
         bind_func!(context, token, send_message);
+        bind_func!(context, token, sleep);
+        bind_func!(context, token, sleep_tipsets);
     }
 
     fn import_prelude(&self, context: &mut Context) -> anyhow::Result<()> {
@@ -368,7 +411,6 @@ impl AttachCommand {
                 }
                 match context.parse(buffer.trim_end()) {
                     Ok(_v) => {
-                        // println!("Parse tree:\n{:#?}", v);
                         editor.add_history_entry(&buffer);
                         eval(buffer.trim_end(), &mut context);
                         break;

forest/cli/src/cli/mod.rs

@@ -15,6 +15,7 @@ mod fetch_params_cmd;
 mod mpool_cmd;
 mod net_cmd;
 mod send_cmd;
+mod shutdown_cmd;
 mod snapshot_cmd;
 mod state_cmd;
 mod sync_cmd;
@@ -35,8 +36,8 @@ pub(super) use self::{
     attach_cmd::AttachCommand, auth_cmd::AuthCommands, chain_cmd::ChainCommands,
     config_cmd::ConfigCommands, db_cmd::DBCommands, fetch_params_cmd::FetchCommands,
     mpool_cmd::MpoolCommands, net_cmd::NetCommands, send_cmd::SendCommand,
-    snapshot_cmd::SnapshotCommands, state_cmd::StateCommands, sync_cmd::SyncCommands,
-    wallet_cmd::WalletCommands,
+    shutdown_cmd::ShutdownCommand, snapshot_cmd::SnapshotCommands, state_cmd::StateCommands,
+    sync_cmd::SyncCommands, wallet_cmd::WalletCommands,
 };
 
 /// CLI structure generated when interacting with Forest binary
@@ -102,6 +103,9 @@ pub enum Subcommand {
 
     /// Attach to daemon via a JavaScript console
     Attach(AttachCommand),
+
+    /// Shutdown Forest
+    Shutdown(ShutdownCommand),
 }
 
 /// Pretty-print a JSON-RPC error and exit
@@ -183,7 +187,8 @@ pub(super) fn print_stdout(out: String) {
 }
 
 fn prompt_confirm() -> bool {
-    println!("Do you want to continue? [y/n]");
+    print!("Do you want to continue? [y/n] ");
+    std::io::stdout().flush().unwrap();
     let mut line = String::new();
     std::io::stdin().read_line(&mut line).unwrap();
     let line = line.trim().to_lowercase();

forest/cli/src/cli/shutdown_cmd.rs

@@ -0,0 +1,28 @@
+// Copyright 2019-2023 ChainSafe Systems
+// SPDX-License-Identifier: Apache-2.0, MIT
+
+use forest_rpc_client::shutdown;
+
+use super::{handle_rpc_err, Config};
+use crate::cli::prompt_confirm;
+
+#[derive(Debug, clap::Args)]
+pub struct ShutdownCommand {
+    /// Assume "yes" as answer to shutdown prompt
+    #[arg(long)]
+    force: bool,
+}
+
+impl ShutdownCommand {
+    pub async fn run(&self, config: Config) -> anyhow::Result<()> {
+        println!("Shutting down Forest node");
+        if !self.force && !prompt_confirm() {
+            println!("Aborted.");
+            return Ok(());
+        }
+        shutdown((), &config.client.rpc_token)
+            .await
+            .map_err(handle_rpc_err)?;
+        Ok(())
+    }
+}

forest/cli/src/subcommand.rs

@@ -20,5 +20,6 @@ pub(super) async fn process(command: Subcommand, config: Config) -> anyhow::Resu
         Subcommand::DB(cmd) => cmd.run(&config),
         Subcommand::Snapshot(cmd) => cmd.run(config).await,
         Subcommand::Attach(cmd) => cmd.run(config),
+        Subcommand::Shutdown(cmd) => cmd.run(config).await,
     }
 }

forest/daemon/Cargo.toml

@@ -16,7 +16,6 @@ anes = "0.1.6"
 anyhow.workspace = true
 atty.workspace = true
 clap.workspace = true
-ctrlc = { version = "3.2", features = ["termination"] }
 daemonize-me = "2.0"
 dialoguer.workspace = true
 flume.workspace = true
@@ -51,7 +50,7 @@ serde_json.workspace = true
 shared_memory = "0.12"
 tempfile.workspace = true
 time.workspace = true
-tokio = { workspace = true, features = ["sync", "macros", "rt"] }
+tokio = { workspace = true, features = ["sync", "macros", "rt", "signal"] }
 
 [dev-dependencies]
 assert_cmd.workspace = true

forest/daemon/src/cli/mod.rs

@@ -1,21 +1,12 @@
 // Copyright 2019-2023 ChainSafe Systems
 // SPDX-License-Identifier: Apache-2.0, MIT
 
-use std::{
-    cell::RefCell,
-    io::Write,
-    process,
-    sync::{
-        atomic::{AtomicUsize, Ordering},
-        Arc,
-    },
-};
+use std::io::Write;
 
 use anes::execute;
 use clap::Parser;
 use forest_cli_shared::cli::{CliOpts, FOREST_VERSION_STRING, HELP_MESSAGE};
-use futures::channel::oneshot::Receiver;
-use log::{info, warn};
+use tokio::{signal, task};
 
 /// CLI structure generated when interacting with Forest binary
 #[derive(Parser)]
@@ -27,28 +18,14 @@ pub struct Cli {
     pub cmd: Option<String>,
 }
 
-pub fn set_sigint_handler() -> Receiver<()> {
-    let (ctrlc_send, ctrlc_oneshot) = futures::channel::oneshot::channel();
-    let ctrlc_send_c = RefCell::new(Some(ctrlc_send));
+pub fn set_sigint_handler() {
+    task::spawn(async {
+        let _ = signal::ctrl_c().await;
 
-    let running = Arc::new(AtomicUsize::new(0));
-    ctrlc::set_handler(move || {
-        let prev = running.fetch_add(1, Ordering::SeqCst);
-        if prev == 0 {
-            warn!("Got interrupt, shutting down...");
-            let mut stdout = std::io::stdout();
-            #[allow(clippy::question_mark)]
-            execute!(&mut stdout, anes::ShowCursor).unwrap();
-            // Send sig int in channel to blocking task
-            if let Some(ctrlc_send) = ctrlc_send_c.try_borrow_mut().unwrap().take() {
-                ctrlc_send.send(()).expect("Error sending ctrl-c message");
-            }
-        } else {
-            info!("Exiting process");
-            process::exit(0);
-        }
-    })
-    .expect("Error setting Ctrl-C handler");
-
-    ctrlc_oneshot
+        // the cursor can go missing if we hit ctrl-c during a prompt, so we always
+        // restore it
+        let mut stdout = std::io::stdout();
+        #[allow(clippy::question_mark)]
+        execute!(&mut stdout, anes::ShowCursor).unwrap();
+    });
 }

forest/daemon/src/daemon.rs

@@ -38,7 +38,11 @@ use fvm_ipld_blockstore::Blockstore;
 use log::{debug, error, info, warn};
 use raw_sync::events::{Event, EventInit, EventState};
 use rpassword::read_password;
-use tokio::{sync::RwLock, task::JoinSet};
+use tokio::{
+    signal::unix::{signal, SignalKind},
+    sync::RwLock,
+    task::JoinSet,
+};
 
 use super::cli::set_sigint_handler;
 
@@ -65,7 +69,9 @@ fn unblock_parent_process() -> anyhow::Result<()> {
 
 /// Starts daemon process
 pub(super) async fn start(opts: CliOpts, config: Config) -> anyhow::Result<Db> {
-    let mut ctrlc_oneshot = set_sigint_handler();
+    set_sigint_handler();
+    let (shutdown_send, mut shutdown_recv) = tokio::sync::mpsc::channel(1);
+    let mut terminate = signal(SignalKind::terminate())?;
 
     info!(
         "Starting Forest daemon, version {}",
@@ -291,6 +297,7 @@ pub(super) async fn start(opts: CliOpts, config: Config) -> anyhow::Result<Db> {
                 }),
                 rpc_listen,
                 FOREST_VERSION_STRING.as_str(),
+                shutdown_send,
             )
             .await
             .map_err(|err| anyhow::anyhow!("{:?}", serde_json::to_string(&err)))
@@ -314,10 +321,17 @@ pub(super) async fn start(opts: CliOpts, config: Config) -> anyhow::Result<Db> {
 
     let config = maybe_fetch_snapshot(should_fetch_snapshot, config).await?;
 
-    select! {
+    tokio::select! {
         () = sync_from_snapshot(&config, &state_manager).fuse() => {},
-        _ = ctrlc_oneshot => {
-            // Cancel all async services
+        _ = tokio::signal::ctrl_c() => {
+            services.shutdown().await;
+            return Ok(db);
+        },
+        _ = terminate.recv() => {
+            services.shutdown().await;
+            return Ok(db);
+        },
+        _ = shutdown_recv.recv() => {
             services.shutdown().await;
             return Ok(db);
         },
@@ -334,12 +348,13 @@ pub(super) async fn start(opts: CliOpts, config: Config) -> anyhow::Result<Db> {
 
     // blocking until any of the services returns an error,
     // or CTRL-C is pressed
-    select! {
+    tokio::select! {
         err = propagate_error(&mut services).fuse() => error!("services failure: {}", err),
-        _ = ctrlc_oneshot => {}
+        _ = tokio::signal::ctrl_c() => {},
+        _ = terminate.recv() => {},
+        _ = shutdown_recv.recv() => {},
     }
 
-    // Cancel all async services
     services.shutdown().await;
 
     Ok(db)
@@ -428,7 +443,7 @@ async fn prompt_snapshot_or_die(
     if should_download {
         Ok(true)
     } else {
-        anyhow::bail!("Forest cannot sync without a snapshot. Download a snapshot from a trusted source and import with --import-snapshot=[file] or --download-snapshot to download one automatically");
+        anyhow::bail!("Forest cannot sync without a snapshot. Download a snapshot from a trusted source and import with --import-snapshot=[file] or --auto-download-snapshot to download one automatically");
     }
 }
 

node/rpc-api/src/data_types.rs

@@ -105,7 +105,7 @@ pub struct PeerID {
 }
 
 /// Represents the current version of the API.
-#[derive(Serialize)]
+#[derive(Serialize, Deserialize)]
 #[serde(rename_all = "PascalCase")]
 pub struct APIVersion {
     pub version: String,
@@ -115,7 +115,7 @@ pub struct APIVersion {
 
 /// Integer based value on version information. Highest order bits for Major,
 /// Mid order for Minor and lowest for Patch.
-#[derive(Serialize)]
+#[derive(Serialize, Deserialize)]
 pub struct Version(u32);
 
 impl Version {