Refactor channel index to pubkey

Reuse prefix 64 with pubkey-based keys, update channel store/query interfaces, and add migration to rebuild channel index from channel states.

Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>

Author: quake

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

@@ -4614,13 +4614,13 @@ impl ChannelActorState {
     fn update_graph_for_remote_channel_change(&mut self) {
         if let Some(channel_update_info) = self.get_remote_channel_update_info() {
             if let Some(channel_outpoint) = self.get_funding_transaction_outpoint() {
-                let peer_id = self.get_remote_pubkey();
+                let pubkey = self.get_remote_pubkey();
                 self.network()
                     .send_message(NetworkActorMessage::new_event(
                         NetworkActorEvent::OwnedChannelUpdateEvent(
                             super::graph::OwnedChannelUpdateEvent::Updated(
                                 channel_outpoint,
-                                peer_id,
+                                pubkey,
                                 channel_update_info,
                             ),
                         ),
@@ -4646,14 +4646,14 @@ impl ChannelActorState {
         let Some(channel_outpoint) = self.get_funding_transaction_outpoint() else {
             return;
         };
-        let peer_id = self.get_local_pubkey();
+        let pubkey = self.get_local_pubkey();
         let channel_update_info = self.get_local_channel_update_info();
         self.network()
             .send_message(NetworkActorMessage::new_event(
                 NetworkActorEvent::OwnedChannelUpdateEvent(
                     super::graph::OwnedChannelUpdateEvent::Updated(
                         channel_outpoint,
-                        peer_id,
+                        pubkey,
                         channel_update_info,
                     ),
                 ),
@@ -7917,21 +7917,21 @@ pub trait ChannelActorStateStore {
     fn get_channel_actor_state(&self, id: &Hash256) -> Option<ChannelActorState>;
     fn insert_channel_actor_state(&self, state: ChannelActorState);
     fn delete_channel_actor_state(&self, id: &Hash256);
-    fn get_channel_ids_by_peer(&self, peer_id: &PeerId) -> Vec<Hash256>;
-    fn get_active_channel_ids_by_peer(&self, peer_id: &PeerId) -> Vec<Hash256> {
-        self.get_channel_ids_by_peer(peer_id)
+    fn get_channel_ids_by_pubkey(&self, pubkey: &Pubkey) -> Vec<Hash256>;
+    fn get_active_channel_ids_by_pubkey(&self, pubkey: &Pubkey) -> Vec<Hash256> {
+        self.get_channel_ids_by_pubkey(pubkey)
             .into_iter()
             .filter(
                 |id| matches!(self.get_channel_actor_state(id), Some(state) if !state.is_closed()),
             )
             .collect()
     }
-    fn get_channel_states(&self, peer_id: Option<PeerId>) -> Vec<(PeerId, Hash256, ChannelState)>;
+    fn get_channel_states(&self, pubkey: Option<Pubkey>) -> Vec<(Pubkey, Hash256, ChannelState)>;
     fn get_active_channel_states(
         &self,
-        peer_id: Option<PeerId>,
-    ) -> Vec<(PeerId, Hash256, ChannelState)> {
-        self.get_channel_states(peer_id)
+        pubkey: Option<Pubkey>,
+    ) -> Vec<(Pubkey, Hash256, ChannelState)> {
+        self.get_channel_states(pubkey)
             .into_iter()
             .filter(|(_, _, state)| !state.is_closed())
             .collect()

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

@@ -1144,7 +1144,8 @@ where
             NetworkActorCommand::MaintainConnections => {
                 debug!("Trying to connect to peers with mutual channels");
 
-                for (peer_id, channel_id, channel_state) in self.store.get_channel_states(None) {
+                for (pubkey, channel_id, channel_state) in self.store.get_channel_states(None) {
+                    let peer_id = pubkey.tentacle_peer_id();
                     if state.is_connected(&peer_id) {
                         continue;
                     }
@@ -1278,7 +1279,7 @@ where
                 }
             }
             NetworkActorCommand::CheckChannelsShutdown => {
-                for (_peer_id, channel_id, channel_state) in self.store.get_channel_states(None) {
+                for (_pubkey, channel_id, channel_state) in self.store.get_channel_states(None) {
                     if matches!(
                         channel_state,
                         ChannelState::ChannelReady | ChannelState::ShuttingDown(..)
@@ -1327,7 +1328,8 @@ where
                 let mut with_channel_down_peers = HashSet::new();
                 let mut ready_channels_count = 0;
                 let mut shuttingdown_channels_count = 0;
-                for (peer_id, channel_id, channel_state) in self.store.get_channel_states(None) {
+                for (pubkey, channel_id, channel_state) in self.store.get_channel_states(None) {
+                    let peer_id = pubkey.tentacle_peer_id();
                     if matches!(channel_state, ChannelState::ChannelReady) {
                         if let Some(actor_state) = self.store.get_channel_actor_state(&channel_id) {
                             ready_channels_count += 1;
@@ -4168,9 +4170,10 @@ where
 
         if let Some(info) = self.peer_session_map.get_mut(&peer_id) {
             info.features = Some(init_msg.features);
+            let peer_pubkey = info.pubkey;
             debug_event!(_myself, "PeerInit");
 
-            for channel_id in self.store.get_active_channel_ids_by_peer(&peer_id) {
+            for channel_id in self.store.get_active_channel_ids_by_pubkey(&peer_pubkey) {
                 if let Err(e) = self.reestablish_channel(&peer_id, channel_id).await {
                     error!("Failed to reestablish channel {:x}: {:?}", &channel_id, &e);
                 }

crates/fiber-lib/src/fiber/tests/settle_tlc_set_command_tests.rs

@@ -3,9 +3,6 @@
 use std::cell::RefCell;
 use std::collections::HashMap;
 
-use ckb_types::packed::{OutPoint, Script};
-use tentacle::secio::PeerId;
-
 use crate::fiber::channel::InboundTlcStatus;
 use crate::fiber::channel::{
     AppliedFlags, ChannelActorState, ChannelActorStateStore, ChannelBasePublicKeys,
@@ -15,13 +12,16 @@ use crate::fiber::channel::{
 use crate::fiber::hash_algorithm::HashAlgorithm;
 use crate::fiber::payment::PaymentCustomRecords;
 use crate::fiber::settle_tlc_set_command::{SettleTlcSetCommand, TlcSettlement};
-use crate::fiber::types::{Hash256, HoldTlc, RemoveTlcReason, TlcErrorCode, NO_SHARED_SECRET};
+use crate::fiber::types::{
+    Hash256, HoldTlc, Pubkey, RemoveTlcReason, TlcErrorCode, NO_SHARED_SECRET,
+};
 use crate::gen_rand_sha256_hash;
 use crate::invoice::{CkbInvoice, CkbInvoiceStatus, Currency, InvoiceBuilder, InvoiceError};
 use crate::invoice::{InvoiceStore, PreimageStore};
 use crate::now_timestamp_as_millis_u64;
 use crate::tests::gen_utils::gen_rand_fiber_public_key;
 use crate::time::SystemTime;
+use ckb_types::packed::{OutPoint, Script};
 
 /// Mock store for testing that implements PreimageStore, InvoiceStore, and ChannelActorStateStore
 struct MockStore {
@@ -137,11 +137,11 @@ impl ChannelActorStateStore for MockStore {
         self.channel_states.borrow_mut().remove(id);
     }
 
-    fn get_channel_ids_by_peer(&self, _peer_id: &PeerId) -> Vec<Hash256> {
+    fn get_channel_ids_by_pubkey(&self, _pubkey: &Pubkey) -> Vec<Hash256> {
         self.channel_states.borrow().keys().cloned().collect()
     }
 
-    fn get_channel_states(&self, _peer_id: Option<PeerId>) -> Vec<(PeerId, Hash256, ChannelState)> {
+    fn get_channel_states(&self, _pubkey: Option<Pubkey>) -> Vec<(Pubkey, Hash256, ChannelState)> {
         vec![]
     }
 

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

@@ -589,10 +589,7 @@ where
     ) -> Result<ListChannelsResult, ErrorObjectOwned> {
         let only_pending = params.only_pending.unwrap_or_default();
         let include_closed = params.include_closed.unwrap_or_default();
-        let filter_peer_id = params
-            .pubkey
-            .as_ref()
-            .map(|pubkey| pubkey.tentacle_peer_id());
+        let filter_pubkey = params.pubkey;
 
         // The two filter options are mutually exclusive: `only_pending` narrows to channels
         // that are still opening (or failed to open), while `include_closed` broadens to
@@ -609,15 +606,15 @@ where
         let channel_states = if only_pending {
             // For pending mode, fetch all channel states (including non-active ones
             // like ABANDONED / FUNDING_ABORTED which are "closed" but represent failed openings)
-            self.store.get_channel_states(filter_peer_id.clone())
+            self.store.get_channel_states(filter_pubkey)
         } else if include_closed {
-            self.store.get_channel_states(filter_peer_id.clone())
+            self.store.get_channel_states(filter_pubkey)
         } else {
-            self.store.get_active_channel_states(filter_peer_id.clone())
+            self.store.get_active_channel_states(filter_pubkey)
         };
         let mut channels: Vec<_> = channel_states
             .into_iter()
-            .filter_map(|(_peer_id, channel_id, _state)| {
+            .filter_map(|(_pubkey, channel_id, _state)| {
                 self.store
                     .get_channel_actor_state(&channel_id)
                     .and_then(|state| {
@@ -717,7 +714,7 @@ where
                     continue;
                 }
                 // Apply pubkey filter if provided
-                if let Some(ref filter_pubkey) = params.pubkey {
+                if let Some(filter_pubkey) = filter_pubkey.as_ref() {
                     if filter_pubkey != &record.pubkey {
                         continue;
                     }
@@ -775,7 +772,7 @@ where
             };
             for pending in pending_accept {
                 // Apply pubkey filter if provided
-                if let Some(ref filter_pubkey) = params.pubkey {
+                if let Some(filter_pubkey) = filter_pubkey.as_ref() {
                     if filter_pubkey != &pending.pubkey {
                         continue;
                     }

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

@@ -25,9 +25,9 @@ use crate::fiber::gossip::GossipMessageStore;
 use crate::fiber::payment::{
     Attempt, AttemptStatus, PaymentCustomRecords, PaymentSession, PaymentStatus,
 };
-use crate::fiber::types::{HoldTlc, CURSOR_SIZE};
 #[cfg(feature = "watchtower")]
-use crate::fiber::types::{Privkey, Pubkey};
+use crate::fiber::types::Privkey;
+use crate::fiber::types::{HoldTlc, Pubkey, CURSOR_SIZE};
 use crate::{
     fiber::{
         channel::{
@@ -231,7 +231,7 @@ pub enum KeyValue {
     CkbInvoice(Hash256, CkbInvoice),
     Preimage(Hash256, Hash256),
     CkbInvoiceStatus(Hash256, CkbInvoiceStatus),
-    PeerIdChannelId((PeerId, Hash256), ChannelState),
+    PubkeyChannelId((Pubkey, Hash256), ChannelState),
     OutPointChannelId(OutPoint, Hash256),
     BroadcastMessageTimestamp(BroadcastMessageID, u64),
     BroadcastMessage(Cursor, BroadcastMessage),
@@ -293,12 +293,15 @@ impl StoreKeyValue for KeyValue {
             KeyValue::CkbInvoiceStatus(id, _) => {
                 [&[CKB_INVOICE_STATUS_PREFIX], id.as_ref()].concat()
             }
-            KeyValue::PeerIdChannelId((peer_id, channel_id), _) => [
-                &[PEER_ID_CHANNEL_ID_PREFIX],
-                peer_id.as_bytes(),
-                channel_id.as_ref(),
-            ]
-            .concat(),
+            KeyValue::PubkeyChannelId((pubkey, channel_id), _) => {
+                let pubkey_bytes = pubkey.serialize();
+                [
+                    &[PEER_ID_CHANNEL_ID_PREFIX][..],
+                    &pubkey_bytes[..],
+                    channel_id.as_ref(),
+                ]
+                .concat()
+            }
             KeyValue::OutPointChannelId(outpoint, _) => {
                 [&[CHANNEL_OUTPOINT_CHANNEL_ID_PREFIX], outpoint.as_slice()].concat()
             }
@@ -383,7 +386,7 @@ impl StoreKeyValue for KeyValue {
             KeyValue::CkbInvoice(_, invoice) => serialize_to_vec(invoice, "CkbInvoice"),
             KeyValue::Preimage(_, preimage) => serialize_to_vec(preimage, "Hash256"),
             KeyValue::CkbInvoiceStatus(_, status) => serialize_to_vec(status, "CkbInvoiceStatus"),
-            KeyValue::PeerIdChannelId(_, state) => serialize_to_vec(state, "ChannelState"),
+            KeyValue::PubkeyChannelId(_, state) => serialize_to_vec(state, "ChannelState"),
             KeyValue::OutPointChannelId(_, channel_id) => serialize_to_vec(channel_id, "ChannelId"),
             KeyValue::PaymentSession(_, payment_session) => {
                 serialize_to_vec(payment_session, "PaymentSession")
@@ -444,8 +447,8 @@ impl ChannelActorStateStore for Store {
     fn insert_channel_actor_state(&self, state: ChannelActorState) {
         let mut batch = self.batch();
 
-        batch.put_kv(KeyValue::PeerIdChannelId(
-            (state.get_remote_peer_id(), state.id),
+        batch.put_kv(KeyValue::PubkeyChannelId(
+            (state.get_remote_pubkey(), state.id),
             state.state,
         ));
         if let Some(outpoint) = state.get_funding_transaction_outpoint() {
@@ -459,10 +462,11 @@ impl ChannelActorStateStore for Store {
         if let Some(state) = self.get_channel_actor_state(id) {
             let mut batch = self.batch();
             batch.delete([&[CHANNEL_ACTOR_STATE_PREFIX], id.as_ref()].concat());
+            let remote_pubkey_bytes = state.get_remote_pubkey().serialize();
             batch.delete(
                 [
-                    &[PEER_ID_CHANNEL_ID_PREFIX],
-                    state.get_remote_peer_id().as_bytes(),
+                    &[PEER_ID_CHANNEL_ID_PREFIX][..],
+                    &remote_pubkey_bytes[..],
                     id.as_ref(),
                 ]
                 .concat(),
@@ -474,8 +478,9 @@ impl ChannelActorStateStore for Store {
         }
     }
 
-    fn get_channel_ids_by_peer(&self, peer_id: &tentacle::secio::PeerId) -> Vec<Hash256> {
-        let prefix = [&[PEER_ID_CHANNEL_ID_PREFIX], peer_id.as_bytes()].concat();
+    fn get_channel_ids_by_pubkey(&self, pubkey: &Pubkey) -> Vec<Hash256> {
+        let pubkey_bytes = pubkey.serialize();
+        let prefix = [&[PEER_ID_CHANNEL_ID_PREFIX][..], &pubkey_bytes[..]].concat();
         let iter = self.prefix_iterator(&prefix);
         iter.map(|(key, _)| {
             let channel_id: [u8; 32] = key[prefix.len()..]
@@ -486,21 +491,24 @@ impl ChannelActorStateStore for Store {
         .collect()
     }
 
-    fn get_channel_states(&self, peer_id: Option<PeerId>) -> Vec<(PeerId, Hash256, ChannelState)> {
-        let prefix = match peer_id {
-            Some(peer_id) => [&[PEER_ID_CHANNEL_ID_PREFIX], peer_id.as_bytes()].concat(),
+    fn get_channel_states(&self, pubkey: Option<Pubkey>) -> Vec<(Pubkey, Hash256, ChannelState)> {
+        let prefix = match pubkey {
+            Some(pubkey) => {
+                let pubkey_bytes = pubkey.serialize();
+                [&[PEER_ID_CHANNEL_ID_PREFIX][..], &pubkey_bytes[..]].concat()
+            }
             None => vec![PEER_ID_CHANNEL_ID_PREFIX],
         };
         self.prefix_iterator(&prefix)
             .map(|(key, value)| {
                 let key_len = key.len();
-                let peer_id = PeerId::from_bytes(key[1..key_len - 32].into())
-                    .expect("deserialize peer id should be OK");
+                let pubkey = Pubkey::from_slice(&key[1..key_len - 32])
+                    .expect("deserialize pubkey should be OK");
                 let channel_id: [u8; 32] = key[key_len - 32..]
                     .try_into()
                     .expect("channel id should be 32 bytes");
                 let state = deserialize_from(value.as_ref(), "ChannelState");
-                (peer_id, channel_id.into(), state)
+                (pubkey, channel_id.into(), state)
             })
             .collect()
     }

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

@@ -576,9 +576,9 @@ fn test_channel_actor_state_store() {
     assert!(get_state.is_some());
     assert!(!get_state.unwrap().is_tlc_forwarding_enabled());
 
-    let remote_peer_id = state.get_remote_peer_id();
+    let remote_pubkey = state.get_remote_pubkey();
     assert_eq!(
-        store.get_channel_ids_by_peer(&remote_peer_id),
+        store.get_channel_ids_by_pubkey(&remote_pubkey),
         vec![state.id]
     );
     let channel_point = state.must_get_funding_transaction_outpoint();
@@ -588,7 +588,7 @@ fn test_channel_actor_state_store() {
 
     store.delete_channel_actor_state(&state.id);
     assert!(store.get_channel_actor_state(&state.id).is_none());
-    assert_eq!(store.get_channel_ids_by_peer(&remote_peer_id), vec![]);
+    assert_eq!(store.get_channel_ids_by_pubkey(&remote_pubkey), vec![]);
     let channel_point = state.must_get_funding_transaction_outpoint();
     assert!(store
         .get_channel_state_by_outpoint(&channel_point)
@@ -1022,12 +1022,12 @@ fn test_store_sample_channel_actor_state() {
             sample.shutdown_transaction_hash
         );
 
-        // Verify peer-id index
-        let remote_peer_id = sample.get_remote_peer_id();
-        let channel_ids = store.get_channel_ids_by_peer(&remote_peer_id);
+        // Verify pubkey index
+        let remote_pubkey = sample.get_remote_pubkey();
+        let channel_ids = store.get_channel_ids_by_pubkey(&remote_pubkey);
         assert!(
             channel_ids.contains(&sample.id),
-            "peer-id index should contain the channel id"
+            "pubkey index should contain the channel id"
         );
     }