add store pub-sub

Add StoreWithPubSub to emit store update events. It uses ractor
OutputPort for pub-sub.

This is a rework of StoreWithHooks in nervosnetwork#511

- Use decorator pattern to add additional features on Store
- Minimize the interface change that original codes relying on Store
  traits dot not need to change.
- Avoid using an Actor to fetch extra data asynchronously. This resolves
  the issue mentioned in
  nervosnetwork#615 (comment)
  . The channel actor cleans preimage on success payment automatically.
  Thus we may fail to get the preimage when settling the cch payment in
  the BTC end.

Author: doitian

crates/fiber-bin/src/main.rs

@@ -14,6 +14,7 @@ use fnn::rpc::watchtower::{
     UpdateLocalSettlementParams, UpdateRevocationParams, WatchtowerRpcClient,
 };
 use fnn::store::Store;
+use fnn::store::StoreWithPubSub;
 use fnn::tasks::{
     cancel_tasks_and_wait_for_completion, new_tokio_cancellation_token, new_tokio_task_tracker,
 };
@@ -84,6 +85,7 @@ pub async fn main() -> Result<(), ExitMessage> {
         .store_path();
 
     let store = Store::new(store_path).map_err(|err| ExitMessage(err.to_string()))?;
+    let store = StoreWithPubSub::new(store);
 
     let tracker = new_tokio_task_tracker();
     let token = new_tokio_cancellation_token();

crates/fiber-lib/src/fiber/channel.rs

@@ -7599,6 +7599,62 @@ pub trait ChannelActorStateStore {
     fn get_payment_custom_records(&self, payment_hash: &Hash256) -> Option<PaymentCustomRecords>;
 }
 
+/// Used for delegating the store trait
+pub trait ChannelActorStateStoreDeref {
+    type Target: ChannelActorStateStore;
+    fn channel_actor_state_store_deref(&self) -> &Self::Target;
+}
+
+impl<T: ChannelActorStateStoreDeref> ChannelActorStateStore for T {
+    fn get_channel_actor_state(&self, id: &Hash256) -> Option<ChannelActorState> {
+        self.channel_actor_state_store_deref()
+            .get_channel_actor_state(id)
+    }
+    fn insert_channel_actor_state(&self, state: ChannelActorState) {
+        self.channel_actor_state_store_deref()
+            .insert_channel_actor_state(state);
+    }
+    fn delete_channel_actor_state(&self, id: &Hash256) {
+        self.channel_actor_state_store_deref()
+            .delete_channel_actor_state(id);
+    }
+    fn get_channel_ids_by_peer(&self, peer_id: &PeerId) -> Vec<Hash256> {
+        self.channel_actor_state_store_deref()
+            .get_channel_ids_by_peer(peer_id)
+    }
+    fn get_active_channel_ids_by_peer(&self, peer_id: &PeerId) -> Vec<Hash256> {
+        self.channel_actor_state_store_deref()
+            .get_active_channel_ids_by_peer(peer_id)
+    }
+    fn get_channel_states(&self, peer_id: Option<PeerId>) -> Vec<(PeerId, Hash256, ChannelState)> {
+        self.channel_actor_state_store_deref()
+            .get_channel_states(peer_id)
+    }
+    fn get_active_channel_states(
+        &self,
+        peer_id: Option<PeerId>,
+    ) -> Vec<(PeerId, Hash256, ChannelState)> {
+        self.channel_actor_state_store_deref()
+            .get_active_channel_states(peer_id)
+    }
+    fn get_channel_state_by_outpoint(&self, id: &OutPoint) -> Option<ChannelActorState> {
+        self.channel_actor_state_store_deref()
+            .get_channel_state_by_outpoint(id)
+    }
+    fn insert_payment_custom_records(
+        &self,
+        payment_hash: &Hash256,
+        custom_records: PaymentCustomRecords,
+    ) {
+        self.channel_actor_state_store_deref()
+            .insert_payment_custom_records(payment_hash, custom_records);
+    }
+    fn get_payment_custom_records(&self, payment_hash: &Hash256) -> Option<PaymentCustomRecords> {
+        self.channel_actor_state_store_deref()
+            .get_payment_custom_records(payment_hash)
+    }
+}
+
 /// A wrapper on CommitmentTransaction that has a partial signature along with
 /// the ckb transaction.
 #[derive(Clone, Debug)]

crates/fiber-lib/src/fiber/gossip.rs

@@ -242,6 +242,131 @@ pub trait GossipMessageStore {
     fn delete_channel_timestamps(&self, outpoint: &OutPoint);
 }
 
+/// Used for delegating the store trait
+pub trait GossipMessageStoreDeref {
+    type Target: GossipMessageStore;
+    fn gossip_message_store_deref(&self) -> &Self::Target;
+}
+
+impl<T: GossipMessageStoreDeref> GossipMessageStore for T {
+    fn get_broadcast_messages_iter(
+        &self,
+        after_cursor: &Cursor,
+    ) -> impl IntoIterator<Item = BroadcastMessageWithTimestamp> {
+        self.gossip_message_store_deref()
+            .get_broadcast_messages_iter(after_cursor)
+    }
+
+    fn get_broadcast_messages(
+        &self,
+        after_cursor: &Cursor,
+        count: Option<u16>,
+    ) -> Vec<BroadcastMessageWithTimestamp> {
+        self.gossip_message_store_deref()
+            .get_broadcast_messages(after_cursor, count)
+    }
+
+    fn query_broadcast_messages<I: IntoIterator<Item = BroadcastMessageQuery>>(
+        &self,
+        queries: I,
+    ) -> (Vec<BroadcastMessageWithTimestamp>, Vec<u16>) {
+        self.gossip_message_store_deref()
+            .query_broadcast_messages(queries)
+    }
+
+    fn query_broadcast_message(
+        &self,
+        query: BroadcastMessageQuery,
+    ) -> Option<BroadcastMessageWithTimestamp> {
+        self.gossip_message_store_deref()
+            .query_broadcast_message(query)
+    }
+
+    fn get_broadcast_message_with_cursor(
+        &self,
+        cursor: &Cursor,
+    ) -> Option<BroadcastMessageWithTimestamp> {
+        self.gossip_message_store_deref()
+            .get_broadcast_message_with_cursor(cursor)
+    }
+
+    fn get_latest_broadcast_message_cursor(&self) -> Option<Cursor> {
+        self.gossip_message_store_deref()
+            .get_latest_broadcast_message_cursor()
+    }
+
+    fn get_latest_channel_announcement_timestamp(&self, outpoint: &OutPoint) -> Option<u64> {
+        self.gossip_message_store_deref()
+            .get_latest_channel_announcement_timestamp(outpoint)
+    }
+
+    fn get_latest_channel_update_timestamp(
+        &self,
+        outpoint: &OutPoint,
+        is_node1: bool,
+    ) -> Option<u64> {
+        self.gossip_message_store_deref()
+            .get_latest_channel_update_timestamp(outpoint, is_node1)
+    }
+
+    fn get_latest_node_announcement_timestamp(&self, pk: &Pubkey) -> Option<u64> {
+        self.gossip_message_store_deref()
+            .get_latest_node_announcement_timestamp(pk)
+    }
+
+    fn get_latest_channel_announcement(
+        &self,
+        outpoint: &OutPoint,
+    ) -> Option<(u64, ChannelAnnouncement)> {
+        self.gossip_message_store_deref()
+            .get_latest_channel_announcement(outpoint)
+    }
+
+    fn get_latest_channel_update(
+        &self,
+        outpoint: &OutPoint,
+        is_node1: bool,
+    ) -> Option<ChannelUpdate> {
+        self.gossip_message_store_deref()
+            .get_latest_channel_update(outpoint, is_node1)
+    }
+
+    fn get_latest_node_announcement(&self, pk: &Pubkey) -> Option<NodeAnnouncement> {
+        self.gossip_message_store_deref()
+            .get_latest_node_announcement(pk)
+    }
+
+    fn delete_broadcast_message(&self, cursor: &Cursor) {
+        self.gossip_message_store_deref()
+            .delete_broadcast_message(cursor);
+    }
+
+    fn save_channel_announcement(&self, timestamp: u64, channel_announcement: ChannelAnnouncement) {
+        self.gossip_message_store_deref()
+            .save_channel_announcement(timestamp, channel_announcement);
+    }
+
+    fn save_channel_update(&self, channel_update: ChannelUpdate) {
+        self.gossip_message_store_deref()
+            .save_channel_update(channel_update)
+    }
+
+    fn save_node_announcement(&self, node_announcement: NodeAnnouncement) {
+        self.gossip_message_store_deref()
+            .save_node_announcement(node_announcement);
+    }
+
+    fn get_channel_timestamps_iter(&self) -> impl IntoIterator<Item = (OutPoint, [u64; 3])> {
+        self.gossip_message_store_deref()
+            .get_channel_timestamps_iter()
+    }
+
+    fn delete_channel_timestamps(&self, outpoint: &OutPoint) {
+        self.gossip_message_store_deref()
+            .delete_channel_timestamps(outpoint);
+    }
+}
+
 // A batch of gossip messages has been added to the store since the last time
 // we pulled new messages/messages are pushed to us.
 #[derive(Clone, Debug)]

crates/fiber-lib/src/fiber/graph.rs

@@ -1567,7 +1567,7 @@ pub trait NetworkGraphStateStore {
 }
 
 /// The status of a payment, will update as the payment progresses.
-#[derive(Clone, Debug, PartialEq, Serialize, Deserialize)]
+#[derive(Clone, Debug, Eq, PartialEq, Serialize, Deserialize)]
 pub enum PaymentSessionStatus {
     /// initial status, payment session is created, no HTLC is sent
     Created,

crates/fiber-lib/src/fiber/network.rs

@@ -2337,6 +2337,23 @@ pub trait NetworkActorStateStore {
     fn insert_network_actor_state(&self, id: &PeerId, state: PersistentNetworkActorState);
 }
 
+/// Used for delegating the store trait
+pub trait NetworkActorStateStoreDeref {
+    type Target: NetworkActorStateStore;
+    fn network_actor_state_store_deref(&self) -> &Self::Target;
+}
+
+impl<T: NetworkActorStateStoreDeref> NetworkActorStateStore for T {
+    fn get_network_actor_state(&self, id: &PeerId) -> Option<PersistentNetworkActorState> {
+        self.network_actor_state_store_deref()
+            .get_network_actor_state(id)
+    }
+    fn insert_network_actor_state(&self, id: &PeerId, state: PersistentNetworkActorState) {
+        self.network_actor_state_store_deref()
+            .insert_network_actor_state(id, state);
+    }
+}
+
 static CHANNEL_ACTOR_NAME_PREFIX: AtomicU64 = AtomicU64::new(0u64);
 
 // ractor requires that the actor name is unique, so we add a prefix to the actor name.

crates/fiber-lib/src/invoice/store.rs

@@ -29,3 +29,29 @@ pub trait PreimageStore {
     /// Search for the stored preimage with the given payment hash prefix, should be the first 20 bytes of the payment hash.
     fn search_preimage(&self, payment_hash_prefix: &[u8]) -> Option<Hash256>;
 }
+
+/// Used for delegating the store trait
+pub trait PreimageStoreDeref {
+    type Target: PreimageStore;
+    fn preimage_store_deref(&self) -> &Self::Target;
+}
+
+impl<T: PreimageStoreDeref> PreimageStore for T {
+    fn insert_preimage(&self, payment_hash: Hash256, preimage: Hash256) {
+        self.preimage_store_deref()
+            .insert_preimage(payment_hash, preimage);
+    }
+
+    fn remove_preimage(&self, payment_hash: &Hash256) {
+        self.preimage_store_deref().remove_preimage(payment_hash);
+    }
+
+    fn get_preimage(&self, payment_hash: &Hash256) -> Option<Hash256> {
+        self.preimage_store_deref().get_preimage(payment_hash)
+    }
+
+    fn search_preimage(&self, payment_hash_prefix: &[u8]) -> Option<Hash256> {
+        self.preimage_store_deref()
+            .search_preimage(payment_hash_prefix)
+    }
+}

crates/fiber-lib/src/store/mod.rs

@@ -2,6 +2,9 @@ pub mod db_migrate;
 pub mod migration;
 mod schema;
 pub mod store_impl;
-pub use store_impl::Store;
+pub mod subscription;
+
+pub use store_impl::{Store, StoreWithPubSub};
+
 #[cfg(test)]
 mod tests;

crates/fiber-lib/src/store/store_impl/browser.rs

@@ -1,4 +1,5 @@
 pub struct Store {}
+pub struct StoreWithPubSub {}
 pub struct Batch {}
 pub enum IteratorMode<'a> {
     Start,