s3vectors: name weed-vector processes servers (#12627)

* s3vectors: call the registered weed-vector processes servers

The gateway's discovery named them workers, the name of the admin
plugin workers. Also drops two comments left without a function.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_011UpvafvK8JpagF94MHWqwK

* test/s3vectors: build the Enterprise engine as weed-vector

In an Enterprise checkout, make ./weed-vector builds the distributed
engine, which also serves VectorService, instead of a separate
weed-vector-worker. An open-source checkout builds weed-vector-server
as before.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_011UpvafvK8JpagF94MHWqwK

* test/s3vectors: distributed mode starts two more weed-vector engines

It started weed-vector-worker, which no target builds now, and waited
for a gateway log line that no longer exists. The Enterprise
weed-vector's engines all join the placement controller.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_011UpvafvK8JpagF94MHWqwK

---------

Co-authored-by: Claude Opus 5.5 <noreply@anthropic.com>

.gitignore

@@ -147,7 +147,6 @@ telemetry/server/telemetry-server
 __pycache__/
 /test/s3vectors/weed
 /test/s3vectors/weed-vector
-/test/s3vectors/weed-vector-worker
 /test/s3vectors/weed-vector-cli
 /test/s3vectors/weed-worker
 /test/s3vectors/weed-volume

test/s3vectors/Makefile

@@ -1,10 +1,7 @@
 WEED := ./weed
 VEC := ./weed-vector
 CLI := ./weed-vector-cli
-WORKER_MANIFEST := ../../rust/sw-vector-distributed/Cargo.toml
-ifneq ($(wildcard $(WORKER_MANIFEST)),)
-WORKER := ./weed-vector-worker
-endif
+ENGINE_MANIFEST := ../../rust/sw-vector-distributed/Cargo.toml
 VOLUME := ./weed-volume
 LANCE_WORKER := ./weed-worker
 CARGO_PROFILE ?= release
@@ -21,7 +18,11 @@ $(WEED): FORCE
 	go build -o $(WEED) ../../weed
 
 $(VEC): FORCE
+ifneq ($(wildcard $(ENGINE_MANIFEST)),)
+	cargo build --locked --profile $(CARGO_PROFILE) --manifest-path ../../rust/Cargo.toml -p sw-vector-distributed --bin weed-vector
+else
 	cd ../../seaweed-vector && cargo build --locked --profile $(CARGO_PROFILE) -p weed-vector-server
+endif
 	cp $(CARGO_OUTPUT)/weed-vector $(VEC)
 
 $(CLI): FORCE
@@ -32,19 +33,13 @@ $(LANCE_WORKER): FORCE
 	cargo build --locked --profile $(CARGO_PROFILE) --manifest-path ../../seaweed-worker/Cargo.toml -p weed-lance-worker --bin weed-worker
 	cp $(CARGO_OUTPUT)/weed-worker $(LANCE_WORKER)
 
-ifneq ($(WORKER),)
-$(WORKER): $(VEC) FORCE
-	cargo build --locked --profile $(CARGO_PROFILE) --manifest-path ../../rust/Cargo.toml -p sw-vector-distributed --bin weed-vector
-	cp $(CARGO_OUTPUT)/weed-vector $(WORKER)
-endif
-
-$(VOLUME): $(VEC) $(WORKER) FORCE
+$(VOLUME): $(VEC) FORCE
 	cargo build --locked --profile $(CARGO_PROFILE) --manifest-path ../../seaweed-volume/Cargo.toml --bin weed-volume
 	cp $(CARGO_OUTPUT)/weed-volume $(VOLUME)
 
 FORCE:
 
-build: $(WEED) $(VEC) $(WORKER) $(VOLUME)
+build: $(WEED) $(VEC) $(VOLUME)
 ifneq ($(VECTOR_WORKER),)
 build: $(CLI) $(LANCE_WORKER)
 endif
@@ -56,4 +51,4 @@ perf: build $(CLI)
 	./run_perf.sh
 
 clean:
-	rm -f $(WEED) $(VEC) $(CLI) $(LANCE_WORKER) ./weed-vector-worker $(VOLUME)
+	rm -f $(WEED) $(VEC) $(CLI) $(LANCE_WORKER) $(VOLUME)

test/s3vectors/conftest.py

@@ -242,7 +242,6 @@ def run_cluster(param, tmp_path_factory, identities):
     # VECTOR_WORKER=1 adds the admin server and the Lance maintenance worker,
     # for the tests that build vector indexes through a plugin job.
     lance_worker = binary("LANCE_WORKER_BINARY", "weed-worker") if os.environ.get("VECTOR_WORKER") else None
-    worker = binary("VECTOR_WORKER_BINARY", "weed-vector-worker") if mode == "distributed" else None
     volume_binary = binary("RUST_VOLUME_SERVER_BINARY", "weed-volume") if volume == "rust" else weed
     processes = {}
     with contextlib.ExitStack() as stack:
@@ -252,13 +251,13 @@ def run_cluster(param, tmp_path_factory, identities):
                 sock = reservations.enter_context(socket.socket())
                 sock.bind(("127.0.0.1", 0))
                 ports.append(sock.getsockname()[1])
-        (master, master_grpc, vol, vol_grpc, filer, filer_grpc, s3, s3_grpc, lance, vector, worker1, worker2,
+        (master, master_grpc, vol, vol_grpc, filer, filer_grpc, s3, s3_grpc, lance, vector, engine1, engine2,
          admin, admin_grpc, vector_metrics) = ports
         master_addr = f"127.0.0.1:{master}.{master_grpc}"
         filer_addr = f"127.0.0.1:{filer}.{filer_grpc}"
 
         # VECTOR_READ_JWT=1 makes volume reads need a JWT; every weed process
-        # and the workers sign with the key.
+        # and the engines sign with the key.
         read_jwt_key = os.environ.get("VECTOR_READ_JWT") and uuid.uuid4().hex
         # The cluster's filer signing key: the gateway's catalog authenticates
         # the worker and the CLI by the token they derive from it.
@@ -346,18 +345,19 @@ def run_cluster(param, tmp_path_factory, identities):
             "vector engine never registered with the master",
         )
 
-        if worker:
-            for i, port in enumerate((worker1, worker2)):
-                start(f"worker-{i}", [worker, "--masters", master_addr, "--listen", f"127.0.0.1:{port}"])
-                wait_for(lambda: listening(port), processes, "vector worker not ready")
+        # Distributed mode needs the Enterprise weed-vector, whose engines all
+        # join the masters' placement controller.
+        if mode == "distributed":
+            for i, port in enumerate((engine1, engine2), 1):
+                start(f"vector-{i}", [engine, "--masters", master_addr, "--listen", f"127.0.0.1:{port}",
+                                      "--filer-group", filer_group, "--state-dir", str(directory / f"vector-{i}"),
+                                      "--capacity-bytes", str(1 << 30), "--drain-timeout", "5s"])
+                wait_for(lambda: listening(port), processes, f"vector-{i} not ready")
             wait_for(
-                lambda: all(
-                    f"S3 Vectors: discovered search worker 127.0.0.1:{port}"
-                    in (directory / "s3.log").read_text()
-                    for port in (worker1, worker2)
-                ),
+                lambda: all(f"joined at 127.0.0.1:{port}" in (directory / "master.log").read_text()
+                            for port in (engine1, engine2)),
                 processes,
-                "gateway did not discover both search workers",
+                "the engines never joined the placement controller",
             )
         admin_url = None
         if lance_worker:

weed/command/s3.go

@@ -734,9 +734,9 @@ func (s3opt *S3Options) startLanceServer(listenOnLocalhost bool) *http.Server {
 	return httpS
 }
 
-// vectorsRouter builds the S3 Vectors API router — workers register to the
-// master and the backend resolves them lazily, so a worker that comes up
-// after the gateway still gets picked up on first query.
+// vectorsRouter builds the S3 Vectors API router — weed-vector servers
+// register to the master and the backend resolves them lazily, so a server
+// that comes up after the gateway still gets picked up on first query.
 func (s3opt *S3Options) vectorsRouter(s3ApiServer *s3api.S3ApiServer, grpcDialOption grpc.DialOption, masterAddresses []pb.ServerAddress, filerGroup string) *mux.Router {
 	if len(masterAddresses) == 0 {
 		glog.Warningf("S3 Vectors API disabled: no weed-vector registered on master")
@@ -902,10 +902,3 @@ func (v lanceCredentialVendor) VendTableCredentials(ctx context.Context, princip
 		Expiration:      credentials.Expiration,
 	}, nil
 }
-
-// discoverVectorWorker resolves one registered weed-vector server through the
-// master's cluster-nodes list; OSS fans every vector call to that single
-// backend.
-// discoverVectorWorkers returns every registered weed-vector address —
-// callers keep their current worker when it is still listed so
-// membership-order changes cannot churn live connections.

weed/s3api/s3vectors/discovery.go

@@ -15,7 +15,7 @@ import (
 
 func NewVectorDiscovery(masters func() []pb.ServerAddress, dialOption grpc.DialOption, filerGroup string) *vectorDiscovery {
 	return &vectorDiscovery{masters: masters, grpcDialOption: dialOption, discover: func(m []pb.ServerAddress, opt grpc.DialOption) []string {
-		return discoverVectorWorkers(m, opt, filerGroup)
+		return discoverVectorServers(m, opt, filerGroup)
 	}}
 }
 
@@ -25,7 +25,7 @@ type vectorDiscovery struct {
 	discover       func([]pb.ServerAddress, grpc.DialOption) []string
 	mu             sync.Mutex
 	closed         bool
-	worker         string
+	server         string
 	failed         string
 	active         map[vector_pb.VectorServiceClient]int
 	retired        map[vector_pb.VectorServiceClient]*grpc.ClientConn
@@ -40,7 +40,7 @@ func (d *vectorDiscovery) Close() error {
 	defer d.mu.Unlock()
 	d.closed = true
 	d.retire()
-	d.worker, d.conn, d.client = "", nil, nil
+	d.server, d.conn, d.client = "", nil, nil
 	return nil
 }
 
@@ -50,9 +50,9 @@ func (d *vectorDiscovery) reset(c vector_pb.VectorServiceClient) {
 	if c != d.client {
 		return
 	}
-	d.failed = d.worker
+	d.failed = d.server
 	d.retire()
-	d.worker, d.conn, d.client = "", nil, nil
+	d.server, d.conn, d.client = "", nil, nil
 	d.nextResolve = time.Time{}
 }
 
@@ -92,7 +92,7 @@ func (d *vectorDiscovery) resolve() {
 	if d.masters != nil {
 		masters = d.masters()
 	}
-	workers := d.discover(masters, d.grpcDialOption)
+	servers := d.discover(masters, d.grpcDialOption)
 	d.mu.Lock()
 	defer d.mu.Unlock()
 	// Measured from lookup end so a slow lookup cannot leave the next one already due.
@@ -105,30 +105,30 @@ func (d *vectorDiscovery) resolve() {
 		close(d.resolving)
 		d.resolving = nil
 	}()
-	if d.closed || len(workers) == 0 {
+	if d.closed || len(servers) == 0 {
 		return
 	}
-	worker := workers[0]
-	for _, w := range workers {
+	server := servers[0]
+	for _, w := range servers {
 		if w != d.failed {
-			worker = w
+			server = w
 			break
 		}
 	}
-	for _, w := range workers {
-		if w == d.worker && w != d.failed {
-			worker = w
+	for _, w := range servers {
+		if w == d.server && w != d.failed {
+			server = w
 			break
 		}
 	}
-	if worker != d.worker {
-		conn, err := pb.GrpcDial(context.Background(), worker, false, d.grpcDialOption)
+	if server != d.server {
+		conn, err := pb.GrpcDial(context.Background(), server, false, d.grpcDialOption)
 		if err != nil {
-			glog.Warningf("S3 Vectors: dial weed-vector %s: %v", worker, err)
+			glog.Warningf("S3 Vectors: dial weed-vector %s: %v", server, err)
 			return
 		}
 		d.retire()
-		d.worker, d.conn = worker, conn
+		d.server, d.conn = server, conn
 		d.client = vector_pb.NewVectorServiceClient(conn)
 	}
 }
@@ -138,7 +138,7 @@ const (
 	vectorRecheckInterval = 500 * time.Millisecond
 )
 
-var errNoVectorWorker = status.Error(codes.Unavailable, "no weed-vector registered on master")
+var errNoVectorServer = status.Error(codes.Unavailable, "no weed-vector registered on master")
 
 func (d *vectorDiscovery) cli(ctx context.Context) (vector_pb.VectorServiceClient, error) {
 	if err := ctx.Err(); err != nil {
@@ -168,7 +168,7 @@ func (d *vectorDiscovery) cli(ctx context.Context) (vector_pb.VectorServiceClien
 		return nil, status.Error(codes.Canceled, "vector discovery closed")
 	}
 	if d.client == nil {
-		return nil, errNoVectorWorker
+		return nil, errNoVectorServer
 	}
 	if d.active == nil {
 		d.active = make(map[vector_pb.VectorServiceClient]int)
@@ -326,7 +326,7 @@ func (d *vectorDiscovery) QueryVectors(ctx context.Context, in *vector_pb.QueryV
 	return nil, err
 }
 
-func discoverVectorWorkers(masters []pb.ServerAddress, grpcDialOption grpc.DialOption, filerGroup string) []string {
+func discoverVectorServers(masters []pb.ServerAddress, grpcDialOption grpc.DialOption, filerGroup string) []string {
 	for _, master := range masters {
 		ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
 		conn, err := pb.GrpcDial(ctx, master.ToGrpcAddress(), false, grpcDialOption)
@@ -343,15 +343,15 @@ func discoverVectorWorkers(masters []pb.ServerAddress, grpcDialOption grpc.DialO
 		if err != nil {
 			continue
 		}
-		workers := make([]string, 0, len(resp.ClusterNodes))
+		servers := make([]string, 0, len(resp.ClusterNodes))
 		for _, node := range resp.ClusterNodes {
 			glog.V(0).Infof("S3 Vectors: discovered weed-vector at %s via master %s", node.Address, master)
-			workers = append(workers, node.Address)
+			servers = append(servers, node.Address)
 		}
-		// A follower can answer with an empty list — workers register with
+		// A follower can answer with an empty list — servers register with
 		// the leader, so keep looking rather than accept "none."
-		if len(workers) > 0 {
-			return workers
+		if len(servers) > 0 {
+			return servers
 		}
 	}
 	return nil

weed/s3api/s3vectors/discovery_test.go

@@ -15,7 +15,7 @@ import (
 	"google.golang.org/grpc/status"
 )
 
-// A worker that registers a moment after a failed lookup must surface inside
+// A server 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) {
@@ -28,10 +28,10 @@ func TestVectorDiscoveryRechecksRegistration(t *testing.T) {
 	}
 
 	if _, err := d.cli(t.Context()); status.Code(err) != codes.Unavailable {
-		t.Fatalf("no registered worker: expected Unavailable, got %v", err)
+		t.Fatalf("no registered server: expected Unavailable, got %v", err)
 	}
 	if _, err := d.cli(t.Context()); status.Code(err) != codes.Unavailable {
-		t.Fatalf("no registered worker: expected Unavailable, got %v", err)
+		t.Fatalf("no registered server: 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)
@@ -48,7 +48,7 @@ func TestVectorDiscoveryRechecksRegistration(t *testing.T) {
 	d.grpcDialOption = grpc.WithTransportCredentials(insecure.NewCredentials())
 
 	if _, err := d.cli(t.Context()); err != nil {
-		t.Fatalf("worker registered: expected a client, got %v", err)
+		t.Fatalf("server 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)
@@ -79,10 +79,10 @@ func TestVectorDiscoveryThrottlesFromLookupEnd(t *testing.T) {
 	}
 
 	if _, err := d.cli(t.Context()); status.Code(err) != codes.Unavailable {
-		t.Fatalf("no registered worker: expected Unavailable, got %v", err)
+		t.Fatalf("no registered server: expected Unavailable, got %v", err)
 	}
 	if _, err := d.cli(t.Context()); status.Code(err) != codes.Unavailable {
-		t.Fatalf("no registered worker: expected Unavailable, got %v", err)
+		t.Fatalf("no registered server: expected Unavailable, got %v", err)
 	}
 	if got := atomic.LoadInt32(&calls); got != 1 {
 		t.Fatalf("slow lookup was repeated right away: %d lookups", got)
@@ -97,7 +97,7 @@ func TestVectorDiscoveryThrottlesFromLookupEnd(t *testing.T) {
 		d.nextResolve = time.Time{}
 		d.mu.Unlock()
 		if _, err := d.cli(t.Context()); err != nil {
-			t.Fatalf("worker registered: expected a client, got %v", err)
+			t.Fatalf("server registered: expected a client, got %v", err)
 		}
 		waitResolved(d)
 	}
@@ -105,7 +105,7 @@ func TestVectorDiscoveryThrottlesFromLookupEnd(t *testing.T) {
 	left := time.Until(d.nextResolve)
 	d.mu.Unlock()
 	if left < vectorResolveInterval-lookup/2 {
-		t.Fatalf("refresh of a live worker is due in %v, want ~%v", left, vectorResolveInterval)
+		t.Fatalf("refresh of a live server is due in %v, want ~%v", left, vectorResolveInterval)
 	}
 }
 
@@ -119,7 +119,7 @@ func waitResolved(d *vectorDiscovery) {
 }
 
 // A lookup stuck on an unreachable master must not hold up requests that
-// already have a live worker, and concurrent callers start only one lookup.
+// already have a live server, and concurrent callers start only one lookup.
 func TestVectorDiscoveryLookupDoesNotBlockCachedClient(t *testing.T) {
 	d := &vectorDiscovery{
 		grpcDialOption: grpc.WithTransportCredentials(insecure.NewCredentials()),
@@ -129,7 +129,7 @@ func TestVectorDiscoveryLookupDoesNotBlockCachedClient(t *testing.T) {
 	}
 	cached, err := d.cli(t.Context())
 	if err != nil {
-		t.Fatalf("worker registered: expected a client, got %v", err)
+		t.Fatalf("server registered: expected a client, got %v", err)
 	}
 
 	var calls int32
@@ -179,7 +179,7 @@ func TestVectorDiscoveryLookupDoesNotBlockCachedClient(t *testing.T) {
 	}
 }
 
-// Callers with no worker wait for the in-flight lookup and share its result
+// Callers with no server wait for the in-flight lookup and share its result
 // rather than failing while it runs.
 func TestVectorDiscoveryUncachedCallersShareLookup(t *testing.T) {
 	var calls int32
@@ -212,7 +212,7 @@ func TestVectorDiscoveryUncachedCallersShareLookup(t *testing.T) {
 	}
 }
 
-// A failed dial to a new worker must leave the cached connection open.
+// A failed dial to a new server must leave the cached connection open.
 func TestVectorDiscoveryKeepsConnOnDialError(t *testing.T) {
 	d := &vectorDiscovery{
 		grpcDialOption: grpc.WithTransportCredentials(insecure.NewCredentials()),
@@ -222,7 +222,7 @@ func TestVectorDiscoveryKeepsConnOnDialError(t *testing.T) {
 	}
 	cached, err := d.cli(t.Context())
 	if err != nil {
-		t.Fatalf("worker registered: expected a client, got %v", err)
+		t.Fatalf("server registered: expected a client, got %v", err)
 	}
 
 	// Without transport credentials the dial fails synchronously.
@@ -287,7 +287,7 @@ func TestVectorDiscoveryWaitingRespectsCancellation(t *testing.T) {
 	waitResolved(d)
 }
 
-func TestVectorDiscoveryRetriesAnotherWorker(t *testing.T) {
+func TestVectorDiscoveryRetriesAnotherServer(t *testing.T) {
 	d := &vectorDiscovery{grpcDialOption: grpc.WithTransportCredentials(insecure.NewCredentials()), discover: func([]pb.ServerAddress, grpc.DialOption) []string {
 		return []string{"127.0.0.1:9301", "127.0.0.1:9302"}
 	}}
@@ -302,8 +302,8 @@ func TestVectorDiscoveryRetriesAnotherWorker(t *testing.T) {
 		t.Fatal(err)
 	}
 	defer d.release(second)
-	if d.worker != "127.0.0.1:9302" {
-		t.Fatalf("retry selected failed worker: %s", d.worker)
+	if d.server != "127.0.0.1:9302" {
+		t.Fatalf("retry selected failed server: %s", d.server)
 	}
 }
 

weed/s3api/s3vectors/handlers.go

@@ -1255,7 +1255,7 @@ func (s *Server) handleCreateIndex(w http.ResponseWriter, r *http.Request) {
 			vectorErr(w, r, arns, err)
 			return
 		}
-		if !errors.Is(err, errNoVectorWorker) {
+		if !errors.Is(err, errNoVectorServer) {
 			switch status.Code(err) {
 			case codes.InvalidArgument, codes.PermissionDenied, codes.Unauthenticated, codes.FailedPrecondition, codes.AlreadyExists, codes.NotFound, codes.OutOfRange, codes.Unimplemented:
 			default: