From e7ecb5f1f7e18fd2cfbba709f24de1e59be5b0f2 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Zolt=C3=A1n=20Papp?= Date: Wed, 1 Apr 2026 12:11:39 +0200 Subject: [PATCH] [client] Add backoff reset condition to prevent short-lived retry cycles Refine backoff reset logic to ensure it only occurs for sufficiently long-lived stream connections, avoiding interference with `MaxElapsedTime`. --- flow/client/client.go | 12 +++++++++--- 1 file changed, 9 insertions(+), 3 deletions(-) diff --git a/flow/client/client.go b/flow/client/client.go index f996d939f..ec4080765 100644 --- a/flow/client/client.go +++ b/flow/client/client.go @@ -130,14 +130,20 @@ func (c *GRPCClient) Receive(ctx context.Context, interval time.Duration, msgHan return err } - // we have a successful connection, reset the backoff so that if receive fails later, - // the next retry starts with a short delay instead of continuing the already-elapsed timer - backOff.Reset() + streamStart := time.Now() if err := c.receive(stream, msgHandler); err != nil { if isContextDone(err) { return backoff.Permanent(err) } + + // Reset the backoff so the next retry starts with a short delay instead of + // continuing the already-elapsed timer. Only do this if the stream was healthy + // long enough; short-lived connect/drop cycles must not defeat MaxElapsedTime. + if time.Since(streamStart) >= interval { + backOff.Reset() + } + // RST_STREAM/PROTOCOL_ERROR — connection is corrupt, recreate immediately if s, ok := status.FromError(err); ok && s.Code() == codes.Internal { log.Warnf("connection corrupt, attempting reconnection: %v", err)