From 43626431c3d90b6e469980814c73101b35ffed11 Mon Sep 17 00:00:00 2001 From: Chris Lu Date: Thu, 29 Oct 2026 01:15:29 -0700 Subject: [PATCH] vector: recheck master discovery quickly while no worker is cached (#12199) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 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 | 14 ++++++ weed/command/s3.go | 41 +++++++++++----- weed/command/s3_vector_discovery_test.go | 61 ++++++++++++++++++++++++ weed/shell/command_cluster_ps.go | 17 +++++++ 4 files changed, 120 insertions(+), 13 deletions(-) create mode 100644 weed/command/s3_vector_discovery_test.go diff --git a/test/vectors/run_test.sh b/test/vectors/run_test.sh index ab3826f580..3a44eb9009 100755 --- a/test/vectors/run_test.sh +++ b/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 diff --git a/weed/command/s3.go b/weed/command/s3.go index cb9f4d8b33..972b83bafd 100644 --- a/weed/command/s3.go +++ b/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) { diff --git a/weed/command/s3_vector_discovery_test.go b/weed/command/s3_vector_discovery_test.go new file mode 100644 index 0000000000..3c6f561995 --- /dev/null +++ b/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) + } +} diff --git a/weed/shell/command_cluster_ps.go b/weed/shell/command_cluster_ps.go index 0c1e04fa8f..0e1d3c4dff 100644 --- a/weed/shell/command_cluster_ps.go +++ b/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 {