vector: recheck master discovery quickly while no worker is cached (#12199)

The vectors gateway backs off discovery for 5s between resolutions, but
the backoff also suppressed rediscovery after an empty lookup. A worker
that landed its KeepConnected registration a moment after the first
query stayed invisible for the whole window — longer than a client's
retry budget — so the S3 Vectors e2e flaked: CreateIndex raced the
registration, got InternalFailure, and every boto retry fell inside the
same 5s fast-fail window.

With no cached backend, recheck every 500ms instead; the 5s interval now
only bounds membership refreshes against a live client, and an absent
worker still cannot flood the master with lookups. The empty-membership
error also becomes codes.Unavailable so the API answers 503 rather than
InternalFailure.

cluster.ps lists registered vector nodes like the other node types, and
the e2e harness polls it for the engine's registration instead of only
checking the gRPC port — a listening engine is not discoverable until
the registration lands.

test/vectors/run_test.sh

@@ -48,4 +48,18 @@ for _ in $(seq 1 60); do
 done
 (exec 3<>/dev/tcp/127.0.0.1/9301) 2>/dev/null || { echo "vector server on :9301 never came up" >&2; exit 1; }
 
+# A listening engine still has to land its KeepConnected registration before
+# the gateway's lazy discovery can find it — poll the cluster list for the
+# vector node rather than race the first request against it.
+vector_registered() {
+  # grep -q would close the pipe on first match and SIGPIPE the shell mid-write
+  # under `set -o pipefail`; a full read lets cluster.ps finish.
+  printf 'cluster.ps\n' | ./weed shell -master 127.0.0.1:9333 2>/dev/null | grep "vector servers" >/dev/null
+}
+for _ in $(seq 1 60); do
+  vector_registered && break
+  sleep 1
+done
+vector_registered || { echo "vector server never registered with the master" >&2; exit 1; }
+
 python3 -m pytest -xvs s3vectors_test.py

weed/command/s3.go

@@ -717,7 +717,7 @@ func (s3opt *S3Options) vectorsRouter(s3ApiServer *s3api.S3ApiServer, grpcDialOp
 		glog.Warningf("S3 Vectors API disabled: no weed-vector registered on master")
 		return nil
 	}
-	backend := &vectorDiscovery{masters: masterAddresses, grpcDialOption: grpcDialOption}
+	backend := &vectorDiscovery{masters: masterAddresses, grpcDialOption: grpcDialOption, discover: discoverVectorWorkers}
 	vectorsRouter := mux.NewRouter().SkipClean(true)
 
 	vectorsServer := s3vectors.NewServer(s3ApiServer, s3ApiServer, backend)
@@ -733,6 +733,7 @@ func (s3opt *S3Options) vectorsRouter(s3ApiServer *s3api.S3ApiServer, grpcDialOp
 type vectorDiscovery struct {
 	masters        []pb.ServerAddress
 	grpcDialOption grpc.DialOption
+	discover       func([]pb.ServerAddress, grpc.DialOption) []string
 	mu             sync.Mutex
 	worker         string
 	conn           *grpc.ClientConn
@@ -765,9 +766,9 @@ func isTransportError(err error) bool {
 func (d *vectorDiscovery) resolve() (vector_pb.VectorServiceClient, error) {
 	d.mu.Lock()
 	defer d.mu.Unlock()
-	workers := discoverVectorWorkers(d.masters, d.grpcDialOption)
+	workers := d.discover(d.masters, d.grpcDialOption)
 	if len(workers) == 0 {
-		return nil, fmt.Errorf("no weed-vector registered on master")
+		return nil, errNoVectorWorker
 	}
 	worker := workers[0]
 	for _, w := range workers {
@@ -786,27 +787,41 @@ func (d *vectorDiscovery) resolve() (vector_pb.VectorServiceClient, error) {
 		}
 		d.worker, d.conn = worker, conn
 		d.client = vector_pb.NewVectorServiceClient(conn)
+		// cli() scheduled this resolve on the short recheck window while no
+		// client existed; a live client only refreshes every resolve interval.
+		d.nextResolve = time.Now().Add(vectorResolveInterval)
 	}
 	return d.client, nil
 }
 
-const vectorResolveInterval = 5 * time.Second
+// A live client refreshes membership at most every vectorResolveInterval.
+// With no usable backend the recheck is much more frequent: a worker that
+// registers a moment after a lookup must surface inside a caller's retry
+// budget, while an absent one still cannot flood the master with lookups.
+const (
+	vectorResolveInterval = 5 * time.Second
+	vectorRecheckInterval = 500 * time.Millisecond
+)
+
+var errNoVectorWorker = status.Error(codes.Unavailable, "no weed-vector registered on master")
 
 func (d *vectorDiscovery) cli() (vector_pb.VectorServiceClient, error) {
 	d.mu.Lock()
 	cli := d.client
-	past := !time.Now().Before(d.nextResolve)
-	if cli != nil && !past {
+	if !time.Now().Before(d.nextResolve) {
+		interval := vectorRecheckInterval
+		if cli != nil {
+			interval = vectorResolveInterval
+		}
+		d.nextResolve = time.Now().Add(interval)
 		d.mu.Unlock()
+		return d.resolve()
+	}
+	d.mu.Unlock()
+	if cli != nil {
 		return cli, nil
 	}
-	if !past && cli == nil {
-		d.mu.Unlock()
-		return nil, fmt.Errorf("no weed-vector registered on master")
-	}
-	d.nextResolve = time.Now().Add(vectorResolveInterval)
-	d.mu.Unlock()
-	return d.resolve()
+	return nil, errNoVectorWorker
 }
 
 func (d *vectorDiscovery) CreateIndex(ctx context.Context, in *vector_pb.CreateIndexRequest, opts ...grpc.CallOption) (resp *vector_pb.CreateIndexResponse, err error) {

weed/command/s3_vector_discovery_test.go

@@ -0,0 +1,61 @@
+package command
+
+import (
+	"sync/atomic"
+	"testing"
+	"time"
+
+	"github.com/seaweedfs/seaweedfs/weed/pb"
+	"google.golang.org/grpc"
+	"google.golang.org/grpc/codes"
+	"google.golang.org/grpc/credentials/insecure"
+	"google.golang.org/grpc/status"
+)
+
+// A worker that registers a moment after a failed lookup must surface inside
+// a caller's retry budget: the no-client recheck window is much shorter than
+// the live-client membership refresh.
+func TestVectorDiscoveryRechecksRegistration(t *testing.T) {
+	var calls int32
+	d := &vectorDiscovery{
+		discover: func(masters []pb.ServerAddress, opt grpc.DialOption) []string {
+			atomic.AddInt32(&calls, 1)
+			return nil
+		},
+	}
+
+	if _, err := d.cli(); status.Code(err) != codes.Unavailable {
+		t.Fatalf("no registered worker: expected Unavailable, got %v", err)
+	}
+	if _, err := d.cli(); status.Code(err) != codes.Unavailable {
+		t.Fatalf("no registered worker: expected Unavailable, got %v", err)
+	}
+	if got := atomic.LoadInt32(&calls); got != 1 {
+		t.Fatalf("lookup inside the recheck window hit master again: %d lookups", got)
+	}
+
+	// Once the recheck interval passes the next call re-discovers.
+	d.mu.Lock()
+	d.nextResolve = time.Now().Add(-time.Second)
+	d.mu.Unlock()
+	d.discover = func(masters []pb.ServerAddress, opt grpc.DialOption) []string {
+		atomic.AddInt32(&calls, 1)
+		return []string{"127.0.0.1:9301"}
+	}
+	d.grpcDialOption = grpc.WithTransportCredentials(insecure.NewCredentials())
+
+	if _, err := d.cli(); err != nil {
+		t.Fatalf("worker registered: expected a client, got %v", err)
+	}
+	if got := atomic.LoadInt32(&calls); got != 2 {
+		t.Fatalf("expired recheck window did not rediscover: %d lookups", got)
+	}
+
+	// A live client rides out the refresh interval without rediscovery.
+	if _, err := d.cli(); err != nil {
+		t.Fatalf("cached client rejected: %v", err)
+	}
+	if got := atomic.LoadInt32(&calls); got != 2 {
+		t.Fatalf("live client rediscovered inside the refresh window: %d lookups", got)
+	}
+}

weed/shell/command_cluster_ps.go

@@ -75,6 +75,10 @@ func (c *commandClusterPs) Do(args []string, commandEnv *CommandEnv, writer io.W
 	if err != nil {
 		return
 	}
+	vectorNodes, err := listClusterNodes(cluster.VectorType)
+	if err != nil {
+		return
+	}
 
 	if len(mqBrokerNodes) > 0 {
 		fmt.Fprintf(writer, "* message queue brokers %d\n", len(mqBrokerNodes))
@@ -102,6 +106,19 @@ func (c *commandClusterPs) Do(args []string, commandEnv *CommandEnv, writer io.W
 		}
 	}
 
+	if len(vectorNodes) > 0 {
+		fmt.Fprintf(writer, "* vector servers %d\n", len(vectorNodes))
+		for _, node := range vectorNodes {
+			fmt.Fprintf(writer, "  * %s (%v)\n", node.Address, node.Version)
+			if node.DataCenter != "" {
+				fmt.Fprintf(writer, "    DataCenter: %v\n", node.DataCenter)
+			}
+			if node.Rack != "" {
+				fmt.Fprintf(writer, "    Rack: %v\n", node.Rack)
+			}
+		}
+	}
+
 	filerSignatures := make(map[*master_pb.ListClusterNodesResponse_ClusterNode]int32)
 	fmt.Fprintf(writer, "* filers %d\n", len(filerNodes))
 	for _, node := range filerNodes {