STR-404: Yellowstone-grpc-client stream auto-reconnect

[Ozodimgba]

Closes #STR-404

[Ozodimgba]

@leafaar and @lvboudre this should classify as minor bump...the breaking changes are minimal?

[lvboudre]

Proposed changes

I propose this:

DedupStream object which implements Stream trait and apply the dedup logic you did.

Since we are return impl Stream you can return really complex type with lots of nested generics too.

#[pin_project]
struct DedupStream<GrpcStream> {
  pub(crate) state: DedupState, // this dedup state can recovered by another object on failure.
  #[pin]
  inner: GrpcStream
}

impl<S> Stream< for DedupStream<S> 
  where S: Stream<Item = Result<SubscribeUpdate, Status>> + Send + Unpin + 'static
{
   type Item = <S as Stream>::Item;


   fn poll_next(
        mut self: Pin<&mut Self>,
        cx: &mut Context<'_>,
    ) -> Poll<Option<Self::Item>> {
       let mut this = self.project();
       loop  {
           let poll = this.inner.poll_next(cx);
           match poll {
              Poll::Ready(result) => {
                   // do dedup or propagate the error
              }
             Poll:Pending => // return pending
       }
    }
}

Now for the retry logic, you make make another Stream that wraps you dedup stream.

struct NeedReconnect {
   attempt: usize,
   dedup_state: DedupState,
}

struct ReconnectingFut {
  dedup_state: DedupState,
}

// This is just a prototype but it is essentially a state machine, that each state
// can be represented a concrete type, therefore compiled check.
enum AutoReconnectState<GrpcStream> {
   Connected(DedupStream<GrpcStream>),
   NeedReconnect(NeedReconnect),
   ReconnectingFut(ReconnectingFut) // maybe make it generic?
}


pub struct AutoReconnect<GrpcStream> {
  pub(crate) endpoint: Endpoint, // preconfigured endpoint that builds `Channel`,
  // invariant : Dedup_state can only be `Some` for `inner` is `None`
  state: AutoReconnectState<GrpcStream>,
  reconnect_config: ReconnectionConfig // whatever is the reconnect logic.
}


impl<GrpcStream> Stream for AutoReconnect<GrpcStream> 
   where GrpcStream : // Put all the same condition as before
{
    type  Item = <GrpcStream as Stream>::Item;

    fn poll_next(
        mut self: Pin<&mut Self>,
        cx: &mut Context<'_>,
    ) -> Poll<Option<Self::Item>> {
       let mut this = self.project();
       //  either poll if alive or apply your reconnection state machine.
    }
}

Using custom Stream types will make the code more elegant instead of cramming all the reconnection logic/state machine inside a loop.

Dedup window

I didn't see in the case any way to prune the dedup state where you stored all dedup keys. We need a maximum retention window that must be configurable by the customer.

[leafaar]

I left some comments, looks good, though I think we need to support DeshredTransactions as well, it's another Subscribe method though, so it can be alittle bit annoying