Add database migration

Cargo.toml

@@ -30,7 +30,7 @@ ckb-jsonrpc-types = "0.118.0"
 ckb-chain-spec    = "0.118.0"
 ckb-resource      = "0.118.0"
 rocksdb = { package = "ckb-rocksdb", version = "=0.21.1", features = [
-    "snappy",
+    "lz4"
 ], default-features = false }
 serde_with = { version = "3.7.0", features = ["macros", "base64"] }
 hex = "0.4.3"
@@ -60,6 +60,9 @@ tokio = { version = "1", features = [
     "time",
     "signal",
 ] }
+indicatif = "0.16"
+console = "0.15.8"
+bincode = "1.3.3"
 
 [profile.release]
 panic = "abort"

src/config.rs

@@ -85,6 +85,14 @@ struct Args {
     /// config for ckb
     #[command(flatten)]
     pub ckb: <CkbConfig as ClapSerde>::Opt,
+
+    /// option to run database migration
+    #[arg(
+        short = 'm',
+        long = "migrate",
+        help = "run database migration, default: false"
+    )]
+    pub migrate: bool,
 }
 
 #[derive(Deserialize)]
@@ -119,7 +127,7 @@ pub(crate) fn print_help_and_exit(code: i32) {
 }
 
 impl Config {
-    pub fn parse() -> Self {
+    pub fn parse() -> (Self, bool) {
         // Parse whole args with clap
         let mut args = Args::parse();
 
@@ -193,12 +201,16 @@ impl Config {
         let cch = services.contains(&Service::CCH).then_some(cch);
         let rpc = services.contains(&Service::RPC).then_some(rpc);
         let ckb = services.contains(&Service::CkbChain).then_some(ckb);
-        Self {
-            fiber,
-            cch,
-            rpc,
-            ckb,
-            base_dir,
-        }
+
+        (
+            Self {
+                fiber,
+                cch,
+                rpc,
+                ckb,
+                base_dir,
+            },
+            args.migrate,
+        )
     }
 }

src/errors.rs

@@ -48,6 +48,8 @@ pub enum Error {
     InvalidPeerMessage(String),
     #[error("Onion packet error: {0}")]
     InvalidOnionPacket(crate::fiber::types::Error),
+    #[error("Database error: {0}")]
+    DBInternalError(String),
 }
 
 pub type Result<T> = std::result::Result<T, Error>;

src/fiber/channel.rs

@@ -7,6 +7,7 @@ use crate::{
     fiber::{
         fee::calculate_tlc_forward_fee,
         network::{get_chain_hash, SendOnionPacketCommand},
+        serde_utils::PubNonceAsBytes,
         types::{ChannelUpdate, TlcErr, TlcErrPacket, TlcErrorCode},
     },
     invoice::{CkbInvoice, CkbInvoiceStatus, InvoiceStore},
@@ -2084,7 +2085,9 @@ pub struct ChannelActorState {
     #[serde_as(as = "Option<EntityHex>")]
     pub local_shutdown_script: Option<Script>,
 
+    #[serde_as(as = "Option<PubNonceAsBytes>")]
     pub previous_remote_nonce: Option<PubNonce>,
+    #[serde_as(as = "Option<PubNonceAsBytes>")]
     pub remote_nonce: Option<PubNonce>,
 
     // The latest commitment transaction we're holding
@@ -2107,7 +2110,7 @@ pub struct ChannelActorState {
 }
 
 #[serde_as]
-#[derive(Clone, Serialize, Deserialize)]
+#[derive(Clone, Serialize, Deserialize, Eq, PartialEq, Debug)]
 pub struct ShutdownInfo {
     #[serde_as(as = "EntityHex")]
     pub close_script: Script,
@@ -2121,6 +2124,7 @@ pub struct ShutdownInfo {
 // For ChannelUpdate config, only information on our side are saved here because we have no
 // control to the config on the counterparty side. And they will publish
 // the config to the network via another ChannelUpdate message.
+#[serde_as]
 #[derive(Default, Clone, Debug, Serialize, Deserialize)]
 pub struct PublicChannelInfo {
     pub enabled: bool,
@@ -2140,6 +2144,8 @@ pub struct PublicChannelInfo {
     // Channel announcement signatures, may be empty for private channel.
     pub local_channel_announcement_signature: Option<(EcdsaSignature, PartialSignature)>,
     pub remote_channel_announcement_signature: Option<(EcdsaSignature, PartialSignature)>,
+
+    #[serde_as(as = "Option<PubNonceAsBytes>")]
     pub remote_channel_announcement_nonce: Option<PubNonce>,
 
     pub channel_announcement: Option<ChannelAnnouncement>,
@@ -2297,11 +2303,6 @@ enum CommitmentSignedFlags {
 }
 
 #[derive(Copy, Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
-#[serde(
-    rename_all = "SCREAMING_SNAKE_CASE",
-    tag = "state_name",
-    content = "state_flags"
-)]
 pub enum ChannelState {
     /// We are negotiating the parameters required for the channel prior to funding it.
     NegotiatingFunding(NegotiatingFundingFlags),

src/fiber/graph.rs

@@ -847,7 +847,7 @@ pub trait NetworkGraphStateStore {
     fn get_payment_session(&self, payment_hash: Hash256) -> Option<PaymentSession>;
     fn insert_payment_session(&self, session: PaymentSession);
     fn insert_payment_history_result(&mut self, from: Pubkey, target: Pubkey, result: TimedResult);
-    fn get_payment_history_result(&self) -> Vec<(Pubkey, Pubkey, TimedResult)>;
+    fn get_payment_history_results(&self) -> Vec<(Pubkey, Pubkey, TimedResult)>;
 }
 
 #[derive(Clone, Debug, PartialEq, Serialize, Deserialize)]

src/fiber/history.rs

@@ -281,7 +281,7 @@ where
     }
 
     pub(crate) fn load_from_store(&mut self) {
-        let results = self.store.get_payment_history_result();
+        let results = self.store.get_payment_history_results();
         for (from, target, result) in results.into_iter() {
             self.inner.entry(from).or_default().insert(target, result);
         }

src/fiber/serde_utils.rs

@@ -1,4 +1,5 @@
 use molecule::prelude::Entity;
+use musig2::{BinaryEncoding, CompactSignature, PubNonce, SCHNORR_SIGNATURE_SIZE};
 use serde::{de::Error, Deserialize, Deserializer, Serializer};
 use serde_with::{serde_conv, DeserializeAs, SerializeAs};
 
@@ -110,3 +111,55 @@ uint_as_hex!(U128Hex, u128);
 uint_as_hex!(U64Hex, u64);
 uint_as_hex!(U32Hex, u32);
 uint_as_hex!(U16Hex, u16);
+
+pub struct CompactSignatureAsBytes;
+
+impl SerializeAs<CompactSignature> for CompactSignatureAsBytes {
+    fn serialize_as<S>(signature: &CompactSignature, serializer: S) -> Result<S::Ok, S::Error>
+    where
+        S: Serializer,
+    {
+        serializer.serialize_bytes(&signature.to_bytes())
+    }
+}
+
+impl<'de> DeserializeAs<'de, CompactSignature> for CompactSignatureAsBytes {
+    fn deserialize_as<D>(deserializer: D) -> Result<CompactSignature, D::Error>
+    where
+        D: Deserializer<'de>,
+    {
+        let bytes: &[u8] = Deserialize::deserialize(deserializer)?;
+        if bytes.len() != SCHNORR_SIGNATURE_SIZE {
+            return Err(serde::de::Error::custom("expected 64 bytes"));
+        }
+        let mut array = [0u8; SCHNORR_SIGNATURE_SIZE];
+        array.copy_from_slice(bytes);
+        CompactSignature::from_bytes(&array).map_err(serde::de::Error::custom)
+    }
+}
+
+pub struct PubNonceAsBytes;
+
+impl SerializeAs<PubNonce> for PubNonceAsBytes {
+    fn serialize_as<S>(nonce: &PubNonce, serializer: S) -> Result<S::Ok, S::Error>
+    where
+        S: Serializer,
+    {
+        serializer.serialize_bytes(&nonce.to_bytes())
+    }
+}
+
+impl<'de> DeserializeAs<'de, PubNonce> for PubNonceAsBytes {
+    fn deserialize_as<D>(deserializer: D) -> Result<PubNonce, D::Error>
+    where
+        D: Deserializer<'de>,
+    {
+        let bytes: &[u8] = Deserialize::deserialize(deserializer)?;
+        if bytes.len() != 66 {
+            return Err(serde::de::Error::custom("expected 66 bytes"));
+        }
+        let mut array = [0u8; 66];
+        array.copy_from_slice(bytes);
+        PubNonce::from_bytes(&array).map_err(serde::de::Error::custom)
+    }
+}

src/fiber/tests/graph.rs

@@ -33,7 +33,7 @@ struct MockNetworkGraph {
 impl MockNetworkGraph {
     pub fn new(node_num: usize) -> Self {
         let temp_path = tempfile::tempdir().unwrap();
-        let store = Store::new(temp_path.path());
+        let store = Store::new(temp_path.path()).expect("create store failed");
         let keypairs = generate_key_pairs(node_num + 1);
         let (secret_key1, public_key1) = keypairs[0];
         let mut graph = NetworkGraph::new(store, public_key1.into());

src/fiber/tests/history.rs

@@ -651,7 +651,7 @@ fn test_history_eval_probability_range() {
 fn test_history_load_store() {
     let dir = tempdir().unwrap();
     let path = dir.path().join("test_history_load_store");
-    let store = Store::new(path);
+    let store = Store::new(path).expect("created store failed");
     let mut history = PaymentHistory::new(generate_pubkey().into(), None, store.clone());
     let from = generate_pubkey();
     let target = generate_pubkey();

src/fiber/tests/network.rs

@@ -5,6 +5,7 @@ use crate::fiber::tests::test_utils::generate_pubkey;
 use crate::invoice::InvoiceBuilder;
 use crate::{
     fiber::{
+        channel::ShutdownInfo,
         graph::{ChannelInfo, NetworkGraphStateStore},
         network::{get_chain_hash, NetworkActorStateStore, SendPaymentCommand},
         tests::test_utils::NetworkNodeConfigBuilder,
@@ -27,6 +28,7 @@ use ckb_types::{
     prelude::{Builder, Entity, Pack},
 };
 use core::time::Duration;
+use musig2::PartialSignature;
 use std::{borrow::Cow, str::FromStr};
 use tentacle::{
     multiaddr::{MultiAddr, Protocol},
@@ -648,6 +650,42 @@ async fn test_saving_and_connecting_to_node() {
     .await;
 }
 
+#[test]
+fn test_announcement_message_serialize() {
+    let capacity = 42;
+    let priv_key: Privkey = get_test_priv_key();
+    let pubkey = priv_key.x_only_pub_key().serialize();
+    let pubkey_hash = &blake2b_256(pubkey.as_slice())[0..20];
+    let tx = TransactionView::new_advanced_builder()
+        .output(
+            CellOutput::new_builder()
+                .capacity(capacity.pack())
+                .lock(ScriptBuilder::default().args(pubkey_hash.pack()).build())
+                .build(),
+        )
+        .output_data(vec![0u8; 8].pack())
+        .build();
+    let outpoint = tx.output_pts()[0].clone();
+    let mut channel_announcement =
+        create_fake_channel_announcement_mesage(priv_key, capacity, outpoint);
+
+    channel_announcement.udt_type_script = Some(ScriptBuilder::default().build());
+
+    eprintln!("channel_announcement: {:#?}", channel_announcement);
+    let serialized = bincode::serialize(&channel_announcement).unwrap();
+    let deserialized: ChannelAnnouncement = bincode::deserialize(&serialized).unwrap();
+    assert_eq!(channel_announcement, deserialized);
+
+    let shutdown_info = ShutdownInfo {
+        close_script: ScriptBuilder::default().build(),
+        fee_rate: 100 as u64,
+        signature: Some(PartialSignature::max()),
+    };
+    let serialized = bincode::serialize(&shutdown_info).unwrap();
+    let deserialized: ShutdownInfo = bincode::deserialize(&serialized).unwrap();
+    assert_eq!(shutdown_info, deserialized);
+}
+
 #[test]
 fn test_send_payment_validate_payment_hash() {
     let send_command = SendPaymentCommand {