fix: address Macroscope review feedback

- Cap exponential backoff at 30s to prevent unbounded growth (V3+V4)
- Add context cancellation check in V4 listener outer reconnect loop
- Close old gRPC connection before creating new one in refreshV4Client
- Handle idempotency key marshal failure with fallback instead of silent empty hash

Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>

Author: neekolas

pkg/xmtp/client.go

@@ -51,11 +51,11 @@ func NewClient(ctx context.Context, apiAddress string, useTls bool, clientVersio
 	return v1.NewMessageApiClient(conn), nil
 }
 
-func NewV4Client(ctx context.Context, apiAddress string, useTls bool, clientVersion, appVersion string) (notificationApi.NotificationApiClient, error) {
+func NewV4Client(ctx context.Context, apiAddress string, useTls bool, clientVersion, appVersion string) (notificationApi.NotificationApiClient, *grpc.ClientConn, error) {
 	conn, err := newConn(apiAddress, useTls, clientVersion, appVersion)
 	if err != nil {
-		return nil, err
+		return nil, nil, err
 	}
 
-	return notificationApi.NewNotificationApiClient(conn), nil
+	return notificationApi.NewNotificationApiClient(conn), conn, nil
 }

pkg/xmtp/client_test.go

@@ -16,9 +16,11 @@ func TestXmtpdNotificationApiImportable(t *testing.T) {
 }
 
 func TestNewV4Client_ReturnsClient(t *testing.T) {
-	client, err := NewV4Client(context.Background(), "localhost:0", false, "test", "test")
+	client, conn, err := NewV4Client(context.Background(), "localhost:0", false, "test", "test")
 	require.NoError(t, err)
 	require.NotNil(t, client)
+	require.NotNil(t, conn)
+	_ = conn.Close()
 }
 
 func TestXmtpdEnvelopeTypesImportable(t *testing.T) {

pkg/xmtp/common.go

@@ -9,8 +9,18 @@ import (
 )
 
 const STARTING_SLEEP_TIME = 100 * time.Millisecond
+const MAX_SLEEP_TIME = 30 * time.Second
 const DELIVERY_TIMEOUT = 15 * time.Second
 
+// cappedBackoff doubles sleepTime up to MAX_SLEEP_TIME.
+func cappedBackoff(sleepTime time.Duration) time.Duration {
+	sleepTime *= 2
+	if sleepTime > MAX_SLEEP_TIME {
+		sleepTime = MAX_SLEEP_TIME
+	}
+	return sleepTime
+}
+
 // NotificationListener is the interface implemented by both V3 and V4 listeners
 type NotificationListener interface {
 	Start()

pkg/xmtp/v3_listener.go

@@ -86,7 +86,7 @@ func (l *Listener) startMessageListener() {
 		if err != nil {
 			l.logger.Error("error connecting to stream", zap.Error(err))
 			time.Sleep(sleepTime)
-			sleepTime = sleepTime * 2
+			sleepTime = cappedBackoff(sleepTime)
 			if err = l.refreshClient(); err != nil {
 				l.logger.Error("error refreshing client", zap.Error(err))
 			}
@@ -109,7 +109,7 @@ func (l *Listener) startMessageListener() {
 					l.logger.Error("error reading from stream", zap.Error(err))
 					// Wait 100ms to avoid hammering the API and getting rate limited
 					time.Sleep(sleepTime)
-					sleepTime = sleepTime * 2
+					sleepTime = cappedBackoff(sleepTime)
 					if err = l.refreshClient(); err != nil {
 						l.logger.Error("error refreshing client", zap.Error(err))
 					}

pkg/xmtp/v4_listener.go

@@ -4,6 +4,7 @@ import (
 	"context"
 	"crypto/sha1"
 	"encoding/hex"
+	"fmt"
 	"io"
 	"time"
 
@@ -16,6 +17,7 @@ import (
 	notificationApi "github.com/xmtp/xmtpd/pkg/proto/xmtpv4/message_api"
 	"github.com/xmtp/xmtpd/pkg/topic"
 	"go.uber.org/zap"
+	"google.golang.org/grpc"
 	"google.golang.org/protobuf/proto"
 )
 
@@ -25,6 +27,7 @@ type V4Listener struct {
 	ctx             context.Context
 	cancelFunc      func()
 	v4Client        notificationApi.NotificationApiClient
+	v4Conn          *grpc.ClientConn
 	opts            options.XmtpOptions
 	envelopeChannel chan *envelopesProto.OriginatorEnvelope
 	installations   interfaces.Installations
@@ -43,7 +46,7 @@ func NewV4Listener(
 	clientVersion string,
 	appVersion string,
 ) (*V4Listener, error) {
-	client, err := NewV4Client(ctx, opts.GrpcAddress, opts.UseTls, clientVersion, appVersion)
+	client, conn, err := NewV4Client(ctx, opts.GrpcAddress, opts.UseTls, clientVersion, appVersion)
 	if err != nil {
 		return nil, err
 	}
@@ -56,6 +59,7 @@ func NewV4Listener(
 		cancelFunc:      cancel,
 		logger:          namedLogger,
 		v4Client:        client,
+		v4Conn:          conn,
 		opts:            opts,
 		envelopeChannel: make(chan *envelopesProto.OriginatorEnvelope, 100),
 		installations:   installations,
@@ -83,11 +87,18 @@ func (l *V4Listener) startEnvelopeListener() {
 	l.logger.Info("starting V4 envelope listener")
 	sleepTime := STARTING_SLEEP_TIME
 	for {
+		select {
+		case <-l.ctx.Done():
+			close(l.envelopeChannel)
+			return
+		default:
+		}
+
 		stream, err := l.v4Client.SubscribeAllEnvelopes(l.ctx, &notificationApi.SubscribeAllEnvelopesRequest{})
 		if err != nil {
 			l.logger.Error("error connecting to V4 stream", zap.Error(err))
 			time.Sleep(sleepTime)
-			sleepTime = sleepTime * 2
+			sleepTime = cappedBackoff(sleepTime)
 			if err = l.refreshV4Client(); err != nil {
 				l.logger.Error("error refreshing V4 client", zap.Error(err))
 			}
@@ -108,17 +119,15 @@ func (l *V4Listener) startEnvelopeListener() {
 
 				if err != nil {
 					l.logger.Error("error reading from V4 stream", zap.Error(err))
-					// Wait to avoid hammering the API and getting rate limited
 					time.Sleep(sleepTime)
-					sleepTime = sleepTime * 2
+					sleepTime = cappedBackoff(sleepTime)
 					if err = l.refreshV4Client(); err != nil {
 						l.logger.Error("error refreshing V4 client", zap.Error(err))
 					}
 					break streamLoop
 				}
 
 				if resp != nil {
-					// Reset the sleep time on first successful message
 					sleepTime = STARTING_SLEEP_TIME
 					for _, env := range resp.GetEnvelopes() {
 						l.envelopeChannel <- env
@@ -366,19 +375,25 @@ func logV4ConversionIssue(logger *zap.Logger, clientEnvProto *envelopesProto.Cli
 }
 
 func buildV4IdempotencyKey(env *envelopesProto.OriginatorEnvelope) string {
-	h := sha1.New()
 	b, err := proto.Marshal(env)
-	if err == nil {
-		h.Write(b)
+	if err != nil {
+		// Fall back to a timestamp-based key if marshaling fails
+		return hex.EncodeToString([]byte(fmt.Sprintf("v4-%d", time.Now().UnixNano())))
 	}
+	h := sha1.New()
+	h.Write(b)
 	return hex.EncodeToString(h.Sum(nil))
 }
 
 func (l *V4Listener) refreshV4Client() error {
-	client, err := NewV4Client(l.ctx, l.opts.GrpcAddress, l.opts.UseTls, l.clientVersion, l.appVersion)
+	if l.v4Conn != nil {
+		_ = l.v4Conn.Close()
+	}
+	client, conn, err := NewV4Client(l.ctx, l.opts.GrpcAddress, l.opts.UseTls, l.clientVersion, l.appVersion)
 	if err != nil {
 		return err
 	}
 	l.v4Client = client
+	l.v4Conn = conn
 	return nil
 }