Have a question on "grpc_retry" and how does retry works on streaming the response if something fails?

[TechGeeky]

I have a golang client which is streaming response from a golang server. I want to add logic to my grpc client to retry if something fails while streaming the response.

Below is my client which streams the response from golang server:

func main() {
    timeout := time.Duration(1000) * time.Millisecond
    ctx, _ := context.WithTimeout(context.Background(), timeout)
    conn, err := grpc.DialContext(ctx, "localhost:50005", grpc.WithInsecure())
    if err != nil {
        log.Fatalf("can not connect with server %v", err)
    }

    // create stream
    client := pb.NewCustomerServiceClient(conn)
    req := &pb.Request{ClientId: 12345}
    stream, err := client.FetchResponse(context.Background(), req)
    if err != nil {
        log.Fatalf("open stream error %v", err)
    }
    done := make(chan bool)
    go func() {
        for {
            resp, err := stream.Recv()
            if err == io.EOF {
                done <- true
                return
            }
            if err != nil {
                log.Fatalf("can not receive %v", err)
            }
            log.Printf("Resp received: %s", resp.GetCustomerUrl())
        }
    }()

    <-done
    log.Printf("finished")
}

I am kinda confuse on how to use grpc_retry here to retry my client on whenever streaming response fails. I just want to add simple retry logic here so that my client can retry if any single response fails while streaming. Also I have few other questions on how does retry works on streaming response -

• Does it retry again from start if one single response failed while streaming it? Also if last response fails to receive at the client then will it start again from first?
• Or does it retry only what failed like particular response? Can we even retry that? Is it possible

[TechGeeky]

Can someone help me on this please? As I am totally new to grpc world so trying to figure out on to do retry efficiently?

[johanbrandhorst]

Hi, thanks for the issue. Before I help answer your question, let me encourage you to not use streaming if you are a beginner at gRPC. It adds a lot of complexity that needs to be managed in your application. Even this basic example you have shared has many issues, and would probably be better implemented via polling or listing 😬.

As for the specifics of the question, I am actually not sure exactly what the behavior is and encourage you to simply look at the source code to find answers to your questions, or implementing tests that allow you to do some more debugging of the code.

[TechGeeky]

Yeah I agree but our use case is dependent on streaming only. Our grpc server has streaming endpoint already available so that is why I am trying to handle case on the client side where I call streaming endpoint. I am just curious about how do we usually retry at client side while we are streaming? Is there any example that I can take a look to figure out how people are doing this on their end. I will appreciate some link or guidance on this. @johanbrandhorst

[johanbrandhorst]

I have no personal experience with automatically retrying streaming calls. It would seem to me not obvious what semantics you want, so unlikely that it could be implemented generically.

[imcom]

IIUC, this grpc_retry for client streaming is just broken ... here is the core retry logic ... where we have

func (s *serverStreamingRetryingStream) receiveMsgAndIndicateRetry(m interface{}) (bool, error) {
	s.mu.RLock()
	wasGood := s.receivedGood
	s.mu.RUnlock()
	err := s.getStream().RecvMsg(m)
	if err == nil || err == io.EOF {
		s.mu.Lock()
		s.receivedGood = true
		s.mu.Unlock()
		return false, err
	} else if wasGood {
		// previous RecvMsg in the stream succeeded, no retry logic should interfere
		return false, err
	}

Note that the wasGood once it is true ... it will be always true ... hence skipping all retry logic even if there is an error. The retry can take place only if the very first call failed with retry-able errors ...

A screenshot from my debug session can demonstrate it more clearly