message: NO-TICKET: log when event stream server disconnects a lagging client
diff: @@ -5,8 +5,11 @@ use futures::{Stream, StreamExt};
  use once_cell::sync::Lazy;
  use semver::Version;
  use serde::{Deserialize, Serialize};
  -use tokio::sync::{broadcast, mpsc};
  -use tracing::{error, trace};
  +use tokio::sync::{
  + broadcast::{self, RecvError},
  + mpsc,
  +};
  +use tracing::{error, info, trace};
  use warp::{
  filters::BoxedFilter,
  sse::{self, ServerSentEvent as WarpServerSentEvent},
  @@ -168,14 +171,15 @@ pub(super) fn create_channels_and_filter(
  fn stream_to_client(
  initial_events: mpsc::UnboundedReceiver<ServerSentEvent>,
  ongoing_events: broadcast::Receiver<BroadcastChannelMessage>,
  -) -> impl Stream<Item = Result<impl WarpServerSentEvent, broadcast::RecvError>> + 'static {
  +) -> impl Stream<Item = Result<impl WarpServerSentEvent, RecvError>> + 'static {
  initial_events
  .map(|event| Ok(BroadcastChannelMessage::ServerSentEvent(event)))
  .chain(ongoing_events)
  .map(|result| {
  trace!(?result);
  - match result? {
  - BroadcastChannelMessage::ServerSentEvent(event) => match (event.id, &event.data) {
  + match result {
  + Ok(BroadcastChannelMessage::ServerSentEvent(event)) => {
  + match (event.id, &event.data) {
  (None, &SseData::ApiVersion { .. }) => Ok(sse::json(event.data).boxed()),
  (Some(id), &SseData::BlockFinalized { .. })
  | (Some(id), &SseData::BlockAdded { .. })
  @@ -185,8 +189,18 @@ fn stream_to_client(
  Ok((sse::id(id), sse::json(event.data)).boxed())
  }
  _ => unreachable!("only ApiVersion may have no event ID"),
  - },
  - BroadcastChannelMessage::Shutdown => Err(broadcast::RecvError::Closed),
  + }
  + }
  + Ok(BroadcastChannelMessage::Shutdown) | Err(RecvError::Closed) => {
  + Err(RecvError::Closed)
  + }
  + Err(RecvError::Lagged(amount)) => {
  + info!(
  + "client lagged by {} events - dropping event stream connection to client",
  + amount
  + );
  + Err(RecvError::Lagged(amount))
  + }
  }
  })
  }
  