mirror of
https://github.com/seaweedfs/seaweedfs.git
synced 2026-10-30 08:50:48 +01:00
Chris Lu
43626431c3
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.
62 lines
1.9 KiB
Go
62 lines
1.9 KiB
Go
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)
|
|
}
|
|
}
|