message: Disconnect silent peers
diff: @@ -28,8 +28,10 @@ const OPERATIONS_BATCH_SIZE: usize = 10;
  #[derive(Clone, Debug)]
  pub struct CheckChainCompleteness;
  +#[derive(Clone, Debug)]
  +pub struct DisconnectSilentPeers;
  -#[actor(CheckChainCompleteness, NetworkChannelMsg, ShellChannelMsg, SystemEvent)]
  +#[actor(CheckChainCompleteness, DisconnectSilentPeers, NetworkChannelMsg, ShellChannelMsg, SystemEvent)]
  pub struct ChainManager {
  /// All events generated by the network layer will end up in this channel
  network_channel: NetworkChannelRef,
  @@ -53,6 +55,9 @@ pub struct ChainManager {
  pub type ChainManagerRef = ActorRef<ChainManagerMsg>;
  impl ChainManager {
  +
  + const PEER_RESPONSE_TIMEOUT: Duration = Duration::from_secs(60);
  +
  pub fn actor(sys: &impl ActorRefFactory, network_channel: NetworkChannelRef, shell_channel: ShellChannelRef, rocks_db: Arc<rocksdb::DB>, tezos_storage_init_info: &TezosStorageInitInfo) -> Result<ChainManagerRef, CreateError> {
  sys.actor_of(
  Props::new_args(ChainManager::new, (network_channel, shell_channel, rocks_db, tezos_storage_init_info.chain_id.clone())),
  @@ -302,6 +307,22 @@ impl ChainManager {
  Ok(())
  }
  + fn disconnect_silent_peers(&mut self, ctx: &Context<ChainManagerMsg>) {
  + &self.peers.values()
  + .for_each(|peer_state| {
  + let silence_duration = if peer_state.request_last > peer_state.response_last {
  + peer_state.request_last - peer_state.response_last
  + } else {
  + Duration::from_secs(0)
  + };
  +
  + if silence_duration > Self::PEER_RESPONSE_TIMEOUT {
  + info!("Timeout when waiting for response from peer {:?}. Disconnecting peer.", &peer_state.peer_ref);
  + ctx.system.stop(peer_state.peer_ref.clone());
  + }
  + });
  + }
  +
  fn process_shell_channel_message(&mut self, _ctx: &Context<ChainManagerMsg>, msg: ShellChannelMsg) -> Result<(), Error> {
  match msg {
  ShellChannelMsg::BlockApplied(message) => {
  @@ -334,6 +355,13 @@ impl Actor for ChainManager {
  ctx.myself(),
  None,
  CheckChainCompleteness.into());
  +
  + ctx.schedule::<Self::Msg, _>(
  + Self::PEER_RESPONSE_TIMEOUT.clone(),
  + Self::PEER_RESPONSE_TIMEOUT / 2,
  + ctx.myself(),
  + None,
  + DisconnectSilentPeers.into());
  }
  fn sys_recv(&mut self, ctx: &Context<Self::Msg>, msg: SystemMsg, sender: Option<BasicActorRef>) {
  @@ -376,6 +404,14 @@ impl Receive<CheckChainCompleteness> for ChainManager {
  }
  }
  }
  +impl Receive<DisconnectSilentPeers> for ChainManager {
  + type Msg = ChainManagerMsg;
  +
  + fn receive(&mut self, ctx: &Context<Self::Msg>, _msg: DisconnectSilentPeers, _sender: Sender) {
  + self.disconnect_silent_peers(ctx)
  + }
  +}
  +
  impl Receive<NetworkChannelMsg> for ChainManager {
  type Msg = ChainManagerMsg;
  