@@ -2699,11 +2699,12 @@ mod tests {
2699
2699
use crate :: ln:: types:: ChannelId ;
2700
2700
use crate :: ln:: features:: { InitFeatures , NodeFeatures } ;
2701
2701
use crate :: ln:: peer_channel_encryptor:: PeerChannelEncryptor ;
2702
- use crate :: ln:: peer_handler:: { CustomMessageHandler , PeerManager , MessageHandler , SocketDescriptor , IgnoringMessageHandler , filter_addresses, ErroringMessageHandler , MAX_BUFFER_DRAIN_TICK_INTERVALS_PER_PEER } ;
2702
+ use crate :: ln:: peer_handler:: { CustomMessageHandler , OnionMessageHandler , PeerManager , MessageHandler , SocketDescriptor , IgnoringMessageHandler , filter_addresses, ErroringMessageHandler , MAX_BUFFER_DRAIN_TICK_INTERVALS_PER_PEER } ;
2703
2703
use crate :: ln:: { msgs, wire} ;
2704
2704
use crate :: ln:: msgs:: { Init , LightningError , SocketAddress } ;
2705
2705
use crate :: util:: test_utils;
2706
2706
2707
+
2707
2708
use bitcoin:: Network ;
2708
2709
use bitcoin:: blockdata:: constants:: ChainHash ;
2709
2710
use bitcoin:: secp256k1:: { PublicKey , SecretKey } ;
@@ -2780,6 +2781,76 @@ mod tests {
2780
2781
}
2781
2782
}
2782
2783
2784
+ struct TestPeerTrackingMessageHandler {
2785
+ features : InitFeatures ,
2786
+ pub peer_connected_called : AtomicBool ,
2787
+ pub peer_disconnected_called : AtomicBool ,
2788
+ }
2789
+
2790
+ impl TestPeerTrackingMessageHandler {
2791
+ pub fn new ( features : InitFeatures ) -> Self {
2792
+ Self {
2793
+ features,
2794
+ peer_connected_called : AtomicBool :: new ( false ) ,
2795
+ peer_disconnected_called : AtomicBool :: new ( false ) ,
2796
+ }
2797
+ }
2798
+ }
2799
+
2800
+ impl wire:: CustomMessageReader for TestPeerTrackingMessageHandler {
2801
+ type CustomMessage = Infallible ;
2802
+ fn read < R : io:: Read > ( & self , _: u16 , _: & mut R ) -> Result < Option < Self :: CustomMessage > , msgs:: DecodeError > {
2803
+ Ok ( None )
2804
+ }
2805
+ }
2806
+
2807
+ impl CustomMessageHandler for TestPeerTrackingMessageHandler {
2808
+ fn handle_custom_message ( & self , _: Infallible , _: & PublicKey ) -> Result < ( ) , LightningError > {
2809
+ unreachable ! ( ) ;
2810
+ }
2811
+
2812
+ fn get_and_clear_pending_msg ( & self ) -> Vec < ( PublicKey , Self :: CustomMessage ) > { Vec :: new ( ) }
2813
+
2814
+ fn peer_disconnected ( & self , _their_node_id : & PublicKey ) {
2815
+ assert ! ( self . peer_connected_called. load( Ordering :: Acquire ) ) ;
2816
+ assert ! ( !self . peer_disconnected_called. load( Ordering :: Acquire ) ) ;
2817
+ self . peer_disconnected_called . store ( true , Ordering :: Release ) ;
2818
+ }
2819
+
2820
+ fn peer_connected ( & self , _their_node_id : & PublicKey , _msg : & Init , _inbound : bool ) -> Result < ( ) , ( ) > {
2821
+ assert ! ( !self . peer_connected_called. load( Ordering :: Acquire ) ) ;
2822
+ assert ! ( !self . peer_disconnected_called. load( Ordering :: Acquire ) ) ;
2823
+ self . peer_connected_called . store ( true , Ordering :: Release ) ;
2824
+ Err ( ( ) )
2825
+ }
2826
+
2827
+ fn provided_node_features ( & self ) -> NodeFeatures { NodeFeatures :: empty ( ) }
2828
+
2829
+ fn provided_init_features ( & self , _: & PublicKey ) -> InitFeatures {
2830
+ self . features . clone ( )
2831
+ }
2832
+ }
2833
+
2834
+ impl OnionMessageHandler for TestPeerTrackingMessageHandler {
2835
+ fn handle_onion_message ( & self , _peer_node_id : & PublicKey , _msg : & msgs:: OnionMessage ) { }
2836
+ fn next_onion_message_for_peer ( & self , _peer_node_id : PublicKey ) -> Option < msgs:: OnionMessage > { None }
2837
+ fn peer_disconnected ( & self , _their_node_id : & PublicKey ) {
2838
+ assert ! ( self . peer_connected_called. load( Ordering :: Acquire ) ) ;
2839
+ assert ! ( !self . peer_disconnected_called. load( Ordering :: Acquire ) ) ;
2840
+ self . peer_disconnected_called . store ( true , Ordering :: Release ) ;
2841
+ }
2842
+
2843
+ fn peer_connected ( & self , _their_node_id : & PublicKey , _msg : & Init , _inbound : bool ) -> Result < ( ) , ( ) > {
2844
+ assert ! ( !self . peer_connected_called. load( Ordering :: Acquire ) ) ;
2845
+ assert ! ( !self . peer_disconnected_called. load( Ordering :: Acquire ) ) ;
2846
+ self . peer_connected_called . store ( true , Ordering :: Release ) ;
2847
+ Err ( ( ) )
2848
+ }
2849
+ fn timer_tick_occurred ( & self ) { }
2850
+ fn provided_node_features ( & self ) -> NodeFeatures { NodeFeatures :: empty ( ) }
2851
+ fn provided_init_features ( & self , _their_node_id : & PublicKey ) -> InitFeatures { self . features . clone ( ) }
2852
+ }
2853
+
2783
2854
fn create_peermgr_cfgs ( peer_count : usize ) -> Vec < PeerManagerCfg > {
2784
2855
let mut cfgs = Vec :: new ( ) ;
2785
2856
for i in 0 ..peer_count {
@@ -3164,6 +3235,99 @@ mod tests {
3164
3235
assert_eq ! ( peers[ 0 ] . peers. read( ) . unwrap( ) . len( ) , 0 ) ;
3165
3236
}
3166
3237
3238
+ #[ test]
3239
+ fn test_peer_connected_error_disconnects ( ) {
3240
+
3241
+ struct PeerTrackingPeerManagerConfig {
3242
+ logger : test_utils:: TestLogger ,
3243
+ node_signer : test_utils:: TestNodeSigner ,
3244
+ chan_handler : test_utils:: TestChannelMessageHandler ,
3245
+ route_handler : test_utils:: TestRoutingMessageHandler ,
3246
+ onion_message_handler : TestPeerTrackingMessageHandler ,
3247
+ custom_message_handler : TestPeerTrackingMessageHandler ,
3248
+ }
3249
+
3250
+ fn create_cfgs ( peers : u8 ) -> Vec < PeerTrackingPeerManagerConfig > {
3251
+ let mut cfgs = vec ! [ ] ;
3252
+ for i in 0 ..peers {
3253
+ let features = {
3254
+ let mut feature_bits = vec ! [ 0u8 ; 33 ] ;
3255
+ feature_bits[ 32 ] = 0b00000001 ;
3256
+ InitFeatures :: from_le_bytes ( feature_bits)
3257
+ } ;
3258
+ let node_secret = SecretKey :: from_slice ( & [ 42 + i as u8 ; 32 ] ) . unwrap ( ) ;
3259
+ cfgs. push ( PeerTrackingPeerManagerConfig {
3260
+ logger : test_utils:: TestLogger :: new ( ) ,
3261
+ node_signer : test_utils:: TestNodeSigner :: new ( node_secret) ,
3262
+ chan_handler : test_utils:: TestChannelMessageHandler :: new ( ChainHash :: using_genesis_block ( Network :: Testnet ) ) ,
3263
+ route_handler : test_utils:: TestRoutingMessageHandler :: new ( ) ,
3264
+ onion_message_handler : TestPeerTrackingMessageHandler :: new ( features. clone ( ) ) ,
3265
+ custom_message_handler : TestPeerTrackingMessageHandler :: new ( features. clone ( ) ) ,
3266
+ } ) ;
3267
+ }
3268
+ cfgs
3269
+ }
3270
+
3271
+ type PeerTrackingPeerManager < ' a > = PeerManager < FileDescriptor , & ' a test_utils:: TestChannelMessageHandler , & ' a test_utils:: TestRoutingMessageHandler , & ' a TestPeerTrackingMessageHandler , & ' a test_utils:: TestLogger , & ' a TestPeerTrackingMessageHandler , & ' a test_utils:: TestNodeSigner > ;
3272
+
3273
+ fn create_network < ' a > ( peer_count : usize , cfgs : & ' a Vec < PeerTrackingPeerManagerConfig > ) -> Vec < PeerTrackingPeerManager < ' a > > {
3274
+ let mut peers = Vec :: new ( ) ;
3275
+ for i in 0 ..peer_count {
3276
+ let ephemeral_bytes = [ i as u8 ; 32 ] ;
3277
+ let msg_handler = MessageHandler {
3278
+ chan_handler : & cfgs[ i] . chan_handler , route_handler : & cfgs[ i] . route_handler ,
3279
+ onion_message_handler : & cfgs[ i] . onion_message_handler , custom_message_handler : & cfgs[ i] . custom_message_handler
3280
+ } ;
3281
+ let peer = PeerManager :: new ( msg_handler, 0 , & ephemeral_bytes, & cfgs[ i] . logger , & cfgs[ i] . node_signer ) ;
3282
+ peers. push ( peer) ;
3283
+ }
3284
+
3285
+ peers
3286
+ }
3287
+
3288
+ fn try_establish_connection < ' a > ( peer_a : & PeerTrackingPeerManager < ' a > , peer_b : & PeerTrackingPeerManager < ' a > ) -> ( FileDescriptor , FileDescriptor ) {
3289
+ let id_a = peer_a. node_signer . get_node_id ( Recipient :: Node ) . unwrap ( ) ;
3290
+ let mut fd_a = FileDescriptor {
3291
+ fd : 1 , outbound_data : Arc :: new ( Mutex :: new ( Vec :: new ( ) ) ) ,
3292
+ disconnect : Arc :: new ( AtomicBool :: new ( false ) ) ,
3293
+ } ;
3294
+ let addr_a = SocketAddress :: TcpIpV4 { addr : [ 127 , 0 , 0 , 1 ] , port : 1000 } ;
3295
+ let mut fd_b = FileDescriptor {
3296
+ fd : 1 , outbound_data : Arc :: new ( Mutex :: new ( Vec :: new ( ) ) ) ,
3297
+ disconnect : Arc :: new ( AtomicBool :: new ( false ) ) ,
3298
+ } ;
3299
+ let addr_b = SocketAddress :: TcpIpV4 { addr : [ 127 , 0 , 0 , 1 ] , port : 1001 } ;
3300
+ let initial_data = peer_b. new_outbound_connection ( id_a, fd_b. clone ( ) , Some ( addr_a. clone ( ) ) ) . unwrap ( ) ;
3301
+ peer_a. new_inbound_connection ( fd_a. clone ( ) , Some ( addr_b. clone ( ) ) ) . unwrap ( ) ;
3302
+
3303
+ let _res = peer_a. read_event ( & mut fd_a, & initial_data) ;
3304
+ peer_a. process_events ( ) ;
3305
+
3306
+ let a_data = fd_a. outbound_data . lock ( ) . unwrap ( ) . split_off ( 0 ) ;
3307
+
3308
+ let _res = peer_b. read_event ( & mut fd_b, & a_data) ;
3309
+
3310
+ peer_b. process_events ( ) ;
3311
+ let b_data = fd_b. outbound_data . lock ( ) . unwrap ( ) . split_off ( 0 ) ;
3312
+ let _res = peer_a. read_event ( & mut fd_a, & b_data) ;
3313
+
3314
+ peer_a. process_events ( ) ;
3315
+ let a_data = fd_a. outbound_data . lock ( ) . unwrap ( ) . split_off ( 0 ) ;
3316
+
3317
+ let _res = peer_b. read_event ( & mut fd_b, & a_data) ;
3318
+ ( fd_a. clone ( ) , fd_b. clone ( ) )
3319
+ }
3320
+
3321
+ let cfgs = create_cfgs ( 2 ) ;
3322
+ let peers = create_network ( 2 , & cfgs) ;
3323
+ let ( _sd1, _sd2) = try_establish_connection ( & peers[ 0 ] , & peers[ 1 ] ) ;
3324
+
3325
+ assert ! ( cfgs[ 0 ] . custom_message_handler. peer_connected_called. load( Ordering :: Acquire ) ) ;
3326
+ assert ! ( cfgs[ 0 ] . custom_message_handler. peer_disconnected_called. load( Ordering :: Acquire ) ) ;
3327
+ assert ! ( cfgs[ 0 ] . onion_message_handler. peer_connected_called. load( Ordering :: Acquire ) ) ;
3328
+ assert ! ( cfgs[ 0 ] . onion_message_handler. peer_disconnected_called. load( Ordering :: Acquire ) ) ;
3329
+ }
3330
+
3167
3331
#[ test]
3168
3332
fn test_do_attempt_write_data ( ) {
3169
3333
// Create 2 peers with custom TestRoutingMessageHandlers and connect them.
0 commit comments