Handle errors from SetAddrs in Valkey ring updater (#4024)

We were ignoring errors returned from `ring.SetAddr`, so [if the
creation of the new valkey client
failed](https://github.com/zalando/skipper/blob/8fb2ebe40f5f7aafe9db010c85c5efc8bc283ac3/net/valkey.go#L222),
the `clientMap` stays in the old (or incomlete) state, but in the
[updater](https://github.com/zalando/skipper/blob/8fb2ebe40f5f7aafe9db010c85c5efc8bc283ac3/net/valkey.go#L483)
it was considered successful and we never tried to recreate the client
or update clientMap.

---------

Signed-off-by: Aleksandr Ponimaskin <aleksandr.ponimaskin@zalando.de>

Author: a4180p

net/valkey.go

@@ -434,9 +434,13 @@ func (vrc *ValkeyRingClient) startUpdater(ctx context.Context) {
 
 	init := true
 	if len(old) != 0 {
-		vrc.SetAddrs(ctx, old)
-		vrc.log.Infof("Valkey updater initial set to %d shards", len(old))
-		init = false
+		err := vrc.SetAddrs(ctx, old)
+		if err != nil {
+			vrc.log.Errorf("Failed to init valkey ring: %v", err)
+		} else {
+			vrc.log.Infof("Valkey updater initial set to %d shards", len(old))
+			init = false
+		}
 	}
 
 	for {
@@ -456,18 +460,26 @@ func (vrc *ValkeyRingClient) startUpdater(ctx context.Context) {
 
 		if init {
 			if len(addrs) != 0 {
+				err := vrc.SetAddrs(ctx, addrs)
+				if err != nil {
+					vrc.log.Errorf("Failed to init valkey ring: %v", err)
+					continue
+				}
+
 				init = false
-				vrc.SetAddrs(ctx, addrs)
 				vrc.log.Infof("Valkey updater initial set to %d shards", len(addrs))
 				old = addrs
 			}
 			continue
 		}
 
 		if !slices.Equal(old, addrs) {
-			vrc.SetAddrs(ctx, addrs)
+			err := vrc.SetAddrs(ctx, addrs)
+			if err != nil {
+				vrc.log.Errorf("Failed to update valkey ring: %v", err)
+				continue
+			}
 			vrc.log.Infof("Valkey updater updated old(%d) -> new(%d)", len(old), len(addrs))
-
 			old = addrs
 		}
 	}
@@ -477,9 +489,12 @@ func (vrc *ValkeyRingClient) StartSpan(operationName string, opts ...opentracing
 	return vrc.tracer.StartSpan(operationName, opts...)
 }
 
-func (vrc *ValkeyRingClient) SetAddrs(ctx context.Context, addrs []string) {
-	vrc.ring.SetAddr(addrs)
+func (vrc *ValkeyRingClient) SetAddrs(ctx context.Context, addrs []string) error {
+	if err := vrc.ring.SetAddr(addrs); err != nil {
+		return err
+	}
 	vrc.metrics.UpdateGauge(vrc.metricsPrefix+"shards", float64(vrc.ring.Len()))
+	return nil
 }
 
 func (vrc *ValkeyRingClient) RingAvailable(ctx context.Context) bool {

net/valkey_test.go

@@ -3,7 +3,9 @@ package net
 import (
 	"context"
 	"fmt"
+	"net"
 	"strings"
+	"sync"
 	"testing"
 	"testing/synctest"
 	"time"
@@ -17,6 +19,19 @@ import (
 	"github.com/zalando/skipper/tracing/tracingtest"
 )
 
+func (vr *valkeyRing) hasAddr(addr string) bool {
+	vr.mu.Lock()
+	defer vr.mu.Unlock()
+	_, ok := vr.clientMap[addr]
+	return ok
+}
+
+func (vr *valkeyRing) safeLen() int {
+	vr.mu.Lock()
+	defer vr.mu.Unlock()
+	return vr.Len()
+}
+
 func TestValkeyDifference(t *testing.T) {
 	for _, tt := range []struct {
 		name string
@@ -1593,3 +1608,59 @@ func TestValkeyRingUpdateShardsDeterministic(t *testing.T) {
 		require.Equal(t, snapABC[slot], snapCBA[slot], "shard assignment should be the same")
 	}
 }
+
+func TestValkeyUpdaterRetriesAfterSetAddrsFailure(t *testing.T) {
+	valkeyAddr, done := valkeytest.NewTestValkey(t)
+	defer done()
+
+	newValkeyAddr, done := valkeytest.NewTestValkey(t)
+	defer done()
+
+	l, err := net.Listen("tcp", "127.0.0.1:0")
+	if err != nil {
+		t.Fatalf("Failed to listen: %v", err)
+	}
+	deadAddr := l.Addr().String()
+	l.Close()
+
+	var mu sync.Mutex
+	callCount := 0
+
+	cli, err := NewValkeyRingClient(&ValkeyOptions{
+		Addrs: []string{},
+		AddrUpdater: func() ([]string, error) {
+			mu.Lock()
+			defer mu.Unlock()
+			callCount++
+			if callCount == 1 {
+				return []string{valkeyAddr}, nil
+			}
+			if callCount == 2 {
+				// new address, but unreachable
+				return []string{deadAddr}, nil
+			}
+			return []string{newValkeyAddr}, nil
+		},
+		UpdateInterval: 50 * time.Millisecond,
+	})
+	if err != nil {
+		t.Fatalf("Failed to create valkey ring client: %v", err)
+	}
+	defer cli.Close()
+
+	time.Sleep(cli.options.UpdateInterval + 10*time.Millisecond)
+	if n := cli.ring.safeLen(); n != 1 {
+		t.Fatalf("expected original shard to remain after failed update, got %d shards", n)
+	}
+	if !cli.ring.hasAddr(valkeyAddr) {
+		t.Fatal("expected original shard address to still be in clientMap")
+	}
+
+	time.Sleep(cli.options.UpdateInterval + 10*time.Millisecond)
+	if n := cli.ring.safeLen(); n != 1 {
+		t.Fatalf("expected 1 shard after successful retry, got %d", n)
+	}
+	if !cli.ring.hasAddr(newValkeyAddr) {
+		t.Fatal("expected new shard address to be in clientMap")
+	}
+}