support sentinel TopologyRefreshInterval (#1201)

Author: FZambia

go.mod

@@ -14,7 +14,7 @@ require (
 	github.com/aws/aws-sdk-go-v2/service/sqs v1.46.5
 	github.com/aws/aws-sdk-go-v2/service/sts v1.45.5
 	github.com/aws/smithy-go v1.27.7
-	github.com/centrifugal/centrifuge v0.38.1-0.20260812044137-a63771cce903
+	github.com/centrifugal/centrifuge v0.38.1-0.20260812170737-ccb335a5d97b
 	github.com/centrifugal/protocol v0.20.1-0.20260811164823-815457ed4d09
 	github.com/cristalhq/jwt/v5 v5.4.0
 	github.com/go-viper/mapstructure/v2 v2.5.0
@@ -39,7 +39,7 @@ require (
 	github.com/stretchr/testify v1.11.1
 	github.com/tidwall/gjson v1.19.0
 	github.com/tidwall/sjson v1.2.5
-	github.com/twmb/franz-go v1.21.5
+	github.com/twmb/franz-go v1.21.6
 	github.com/twmb/franz-go/pkg/kadm v1.18.0
 	github.com/twmb/franz-go/pkg/kmsg v1.13.1
 	github.com/valyala/fasttemplate v1.2.2
@@ -128,7 +128,7 @@ require (
 	github.com/prometheus/common v0.70.1 // indirect
 	github.com/prometheus/procfs v0.21.1 // indirect
 	github.com/quic-go/qpack v0.6.0 // indirect
-	github.com/redis/rueidis v1.0.76
+	github.com/redis/rueidis v1.0.77-0.20260811083940-91dfdfd3b348
 	github.com/segmentio/asm v1.2.1 // indirect
 	github.com/segmentio/encoding v0.5.4
 	github.com/spf13/cast v1.10.0 // indirect

go.sum

@@ -73,8 +73,8 @@ github.com/beorn7/perks v1.0.1/go.mod h1:G2ZrVWU2WbWT9wwq4/hrbKbnv/1ERSJQ0ibhJ6r
 github.com/cenkalti/backoff/v5 v5.0.3 h1:ZN+IMa753KfX5hd8vVaMixjnqRZ3y8CuJKRKj1xcsSM=
 github.com/cenkalti/backoff/v5 v5.0.3/go.mod h1:rkhZdG3JZukswDf7f0cwqPNk4K0sa+F97BxZthm/crw=
 github.com/census-instrumentation/opencensus-proto v0.2.1/go.mod h1:f6KPmirojxKA12rnyqOA5BBL4O983OfeGPqjHWSTneU=
-github.com/centrifugal/centrifuge v0.38.1-0.20260812044137-a63771cce903 h1:WLW8wcahmGCnNABQ5QSMq50IcHNbN6QVV6Doq4D1mhs=
-github.com/centrifugal/centrifuge v0.38.1-0.20260812044137-a63771cce903/go.mod h1:ETqrSYvOTIJhPg3BmEzSAF9ok/XK77HVObFkXWFGCNg=
+github.com/centrifugal/centrifuge v0.38.1-0.20260812170737-ccb335a5d97b h1:vxak6TBgKi0GyjDexinQ8xuUBHl65tSpmuSckI7fohg=
+github.com/centrifugal/centrifuge v0.38.1-0.20260812170737-ccb335a5d97b/go.mod h1:9d+DLsD9PDJrVKitBtQ2Yw+EmkdMXiN+0MD7oDpwBqw=
 github.com/centrifugal/protocol v0.20.1-0.20260811164823-815457ed4d09 h1:8SJBvmcS/KXNu8h/79VyYgey4WNreVM9NR3NrGNLFSQ=
 github.com/centrifugal/protocol v0.20.1-0.20260811164823-815457ed4d09/go.mod h1:3pinAfcb+bH8Yp8Ewhf9QTRTzsQ9veP/QOrCk5D5SSE=
 github.com/cespare/xxhash/v2 v2.3.0 h1:UL815xU9SqsFlibzuggzjXhog7bL6oX9BbNZnL2UFvs=
@@ -251,8 +251,8 @@ github.com/quic-go/webtransport-go v0.12.0 h1:CpnKNwZvdV0LD73xoHO8QaR0NI3llqpWRw
 github.com/quic-go/webtransport-go v0.12.0/go.mod h1:GHne8aRFJ24h73pAMrcywXtuaz/ShBXCLXLvG/NPFdU=
 github.com/rakutentech/jwk-go v1.2.0 h1:vNJwedPkRR+32V5WGNj0JP4COes93BGERvzQLBjLy4c=
 github.com/rakutentech/jwk-go v1.2.0/go.mod h1:pI0bYVntqaJ27RCpaC75MTUacheW0Rk4+8XzWWe1OWM=
-github.com/redis/rueidis v1.0.76 h1:RdDWuvlYBSp+bTrBvaXqJnNEL3VVzsnjo+0psPFgLc4=
-github.com/redis/rueidis v1.0.76/go.mod h1:UsfHPSbomB6QAVMk4iiFkzRy0nh9o7scDGa+SitvBY4=
+github.com/redis/rueidis v1.0.77-0.20260811083940-91dfdfd3b348 h1:ShhM52F+S4XJ937CmGnx0cGgZsfBB+Zp0Tr48VoxLLo=
+github.com/redis/rueidis v1.0.77-0.20260811083940-91dfdfd3b348/go.mod h1:UsfHPSbomB6QAVMk4iiFkzRy0nh9o7scDGa+SitvBY4=
 github.com/rogpeppe/go-internal v1.14.1 h1:UQB4HGPB6osV0SQTLymcB4TgvyWu6ZyliaW0tI/otEQ=
 github.com/rogpeppe/go-internal v1.14.1/go.mod h1:MaRKkUm5W0goXpeCfT7UZI6fk/L7L7so1lCWt35ZSgc=
 github.com/rs/zerolog v1.35.1 h1:m7xQeoiLIiV0BCEY4Hs+j2NG4Gp2o2KPKmhnnLiazKI=
@@ -302,8 +302,8 @@ github.com/tidwall/pretty v1.2.1 h1:qjsOFOWWQl+N3RsoF5/ssm1pHmJJwhjlSbZ51I6wMl4=
 github.com/tidwall/pretty v1.2.1/go.mod h1:ITEVvHYasfjBbM0u2Pg8T2nJnzm8xPwvNhhsoaGGjNU=
 github.com/tidwall/sjson v1.2.5 h1:kLy8mja+1c9jlljvWTlSazM7cKDRfJuR/bOJhcY5NcY=
 github.com/tidwall/sjson v1.2.5/go.mod h1:Fvgq9kS/6ociJEDnK0Fk1cpYF4FIW6ZF7LAe+6jwd28=
-github.com/twmb/franz-go v1.21.5 h1:cVYI2+JTTKSvohhy8bCOleYrS7G79ZBrLVFIJsoHm8M=
-github.com/twmb/franz-go v1.21.5/go.mod h1:rfoMTnVk7107fhTGxfEKIHP/e7tPe6oyij/ywzO0czk=
+github.com/twmb/franz-go v1.21.6 h1:+v0dQJVIIuw9uPmPWmPrkoUHs1pPeV8MSwA4eU/Y2kY=
+github.com/twmb/franz-go v1.21.6/go.mod h1:wMepkgCatAdV9vCsuwM+wr+C1fl7KV/41+uHGAjt/wc=
 github.com/twmb/franz-go/pkg/kadm v1.18.0 h1:WRf/LZmDdcDXwX7WMbtDU++v+b3NzYh2bCGoPMmzirw=
 github.com/twmb/franz-go/pkg/kadm v1.18.0/go.mod h1:XeLhGoLXLFzK8/ryv5FfpxPxGwj4oFEGpPJMB/x6KDE=
 github.com/twmb/franz-go/pkg/kmsg v1.13.1 h1:fG5kItwysTk5UXqVwb64EpQEy3TydF3vYYK21nUQ+bI=

internal/redisshard/shard.go

@@ -18,6 +18,18 @@ import (
 const (
 	defaultRedisIOTimeout      = 4 * time.Second
 	defaultRedisConnectTimeout = time.Second
+
+	// sentinelTopologyRefreshInterval makes the Sentinel client periodically
+	// re-check its targets with Sentinels instead of relying solely on the
+	// +switch-master event. That event is the only way the client learns about
+	// a failover, and it can be missed — a demoted master keeps its connection
+	// alive and a role change sends no MOVED, so a client which missed the
+	// event stays pinned to a node that is now a read-only replica until
+	// restart. This is easy to hit with a Sentinel co-located with each Redis
+	// node (as in the Bitnami Redis Helm chart), where the client loses its
+	// Sentinel connection at the same moment the master dies and Redis PUB/SUB
+	// does not replay the missed message.
+	sentinelTopologyRefreshInterval = 5 * time.Second
 )
 
 type RedisShard struct {
@@ -103,8 +115,11 @@ func NewRedisShard(conf RedisShardConfig) (*RedisShard, error) {
 			addressOpts.ClientOption, addressOpts.IsCluster, addressOpts.IsSentinel, addressOpts.ReplicaClientEnabled
 	}
 
-	if isSentinel && options.Sentinel.MasterSet == "" {
-		return nil, errors.New("sentinel master name must be configured for Redis Sentinel setup")
+	if isSentinel {
+		if options.Sentinel.MasterSet == "" {
+			return nil, errors.New("sentinel master name must be configured for Redis Sentinel setup")
+		}
+		options.Sentinel.TopologyRefreshInterval = sentinelTopologyRefreshInterval
 	}
 
 	client, err := rueidis.NewClient(options)