vector: integrate engine health, admin management, and local deployment (#12517)

* vector: expose request metrics and registration-aware health endpoints

* vector: report immutable read cache and storage fetch metrics

* docs: describe vector engine health, metrics, and maintenance

* vector: count canceled requests when their handler is dropped

* vector: account for uncached large immutable storage reads

* vector: reliably release the stalled credential reader in tests

* vector: observe large reads as their streamed body is consumed

* vector: bound heartbeats on established master registration streams

* test: exercise hardened vector maintenance on Go and Rust volumes

* vector: register engines in their configured filer group

* vector: make retried index creation preserve compatible datasets

* vector: allow registration to advertise a separate address

* test: release stalled credential readers without blocking FIFO cleanup

* test: allow hardened vector coverage in a named filer group

* vector: discover engines in the same filer group as their gateway

* vector: fall back when the advertised address is empty

* admin: expose vector readiness, maintenance, and management tools (#12520)

* s3vectors: share vector discovery with trusted management clients

* admin: show vector engine readiness and recent maintenance activity

* admin: suggest S3 Vectors IAM actions from the supported operations

* admin: manage vector indexes, policies, tags, encryption, and query data

* admin: validate SSE-S3 support before creating encrypted vector buckets

* admin: discover vector engines in their registration group

* admin: probe vector engines concurrently with independent time budgets

* admin: verify encryption capability on the remote S3 gateway

* admin: defer discovery until after the HTTP server can start

* admin: distinguish unknown vector health services from failed engines

* admin: count workers that can execute vector index maintenance

* admin: allow read-only users to query vectors and inspect tags

* vector: scope gateway and admin discovery to the configured filer group

* admin: refresh policy state after mutations and serialize policy controls

* vector: drain active RPCs before closing replaced discovery connections

* vector: respect request cancellation while waiting for discovery

* vector: prefer another registered engine after a transport failure

* admin: check all masters before reporting no registered vector engines

* iam: honor attached vector policies without broadening legacy grants

* vector: allow short-lived discovery clients to close their connections

* vector: refresh cached master addresses for each discovery lookup

* iam: enforce vector bucket tag conditions in attached policies

* iam: authorize vector bucket creation with the requested tags

* test: exercise vector IAM grants and tag denies on Go and Rust volumes

* admin: reject authenticated vector mutations without an admin role

* test: run vector IAM assertions before the shared engine is stopped

* vector: add native mini startup, shell tools, and a runnable Compose example (#12522)

* s3vectors: share vector discovery with trusted management clients

* admin: show vector engine readiness and recent maintenance activity

* admin: suggest S3 Vectors IAM actions from the supported operations

* admin: manage vector indexes, policies, tags, encryption, and query data

* admin: validate SSE-S3 support before creating encrypted vector buckets

* admin: discover vector engines in their registration group

* admin: probe vector engines concurrently with independent time budgets

* admin: verify encryption capability on the remote S3 gateway

* admin: defer discovery until after the HTTP server can start

* admin: distinguish unknown vector health services from failed engines

* admin: count workers that can execute vector index maintenance

* admin: allow read-only users to query vectors and inspect tags

* vector: scope gateway and admin discovery to the configured filer group

* admin: refresh policy state after mutations and serialize policy controls

* vector: drain active RPCs before closing replaced discovery connections

* vector: respect request cancellation while waiting for discovery

* vector: prefer another registered engine after a transport failure

* admin: check all masters before reporting no registered vector engines

* iam: honor attached vector policies without broadening legacy grants

* vector: allow short-lived discovery clients to close their connections

* vector: refresh cached master addresses for each discovery lookup

* iam: enforce vector bucket tag conditions in attached policies

* iam: authorize vector bucket creation with the requested tags

* test: exercise vector IAM grants and tag denies on Go and Rust volumes

* admin: reject authenticated vector mutations without an admin role

* test: run vector IAM assertions before the shared engine is stopped

* mini: optionally start and supervise the vector engine

* mini: bootstrap vector buckets with existing Lance catalog helpers

* shell: manage vector buckets, indexes, and vector data

* docs: provide a runnable vector and maintenance worker Compose stack

* shell: scope vector discovery to the session and close its connections

* mini: honor the bind address for vector and metrics listeners

* mini: reserve vector ports before assigning gRPC and compatibility listeners

* mini: wait for vector storage setup and master registration

* mini: create requested vector buckets without requiring the Lance HTTP listener

* shell: refresh master addresses during vector discovery

* mini: resolve sibling vector executables with platform extensions

* vector: preserve uncertain creates and fix mini listener allocation (#12525)

* vector: retain catalog records when index creation has an unknown outcome

* mini: keep chosen listener ports reserved throughout allocation

* mini: resolve hostname binds before starting vector listeners

* vector: fix integration test imports and recover interrupted index creation (#12533)

* shell: register vector operations from the command package

* s3vectors: resume interrupted index initialization safely

* mini: exclude disabled WebDAV from listener allocation

* s3vectors: retain uncertain creates across refused retries

* s3vectors: claim pending indexes before retrying initialization

* s3vectors: compare resolved distance metrics during recovery

* s3vectors: preserve metadata keys when retrying initialization

* s3vectors: verify ownership before initializing new catalog records

weed/command/admin.go

@@ -516,14 +516,6 @@ func startAdminServer(ctx context.Context, options AdminOptions, enableUI bool,
 		return fmt.Errorf("apply admin.toml to plugin config: %w", err)
 	}
 
-	// Show discovered filers
-	filers := adminServer.GetAllFilers()
-	if len(filers) > 0 {
-		glog.Infof("Discovered filers: %s", strings.Join(filers, ", "))
-	} else {
-		glog.Infof("No filers discovered from masters")
-	}
-
 	// Start worker gRPC server for worker connections. The worker gRPC binds
 	// to its own address (workerGrpcBindIp) which, unlike the HTTP ip, does
 	// not auto-upgrade to 0.0.0.0 based on adminPassword, since the worker

weed/command/mini.go

@@ -7,6 +7,7 @@ import (
 	"encoding/json"
 	"errors"
 	"fmt"
+	"maps"
 	"math/bits"
 	"net"
 	"net/http"
@@ -415,6 +416,11 @@ var (
 	miniS3AutoCreateBucket          = cmdMini.Flag.Bool("s3.autoCreateBucket", true, "create the bucket on upload if it does not exist, for admin identities only")
 	miniBucket                      = cmdMini.Flag.String("bucket", "", "comma-separated S3 bucket names to create on startup if they do not already exist; leave empty to skip. Falls back to S3_BUCKET env var.")
 	miniTableBucket                 = cmdMini.Flag.String("tableBucket", "", "comma-separated S3 Tables buckets to create on startup if they do not already exist, each name[:FORMAT] with FORMAT one of ICEBERG (default) or LANCE, e.g. warehouse,vectors:LANCE; leave empty to skip. Falls back to S3_TABLE_BUCKET env var.")
+	miniVector                      = cmdMini.Flag.Bool("vector", false, "start the weed-vector engine for vector indexes and search")
+	miniVectorBinary                = cmdMini.Flag.String("vector.binary", "weed-vector", "path to the vector engine binary")
+	miniVectorPort                  = cmdMini.Flag.Int("vector.port", 8090, "vector engine gRPC port")
+	miniVectorMetricsPort           = cmdMini.Flag.Int("vector.metricsPort", 0, "optional vector HTTP health and metrics port")
+	miniVectorBucket                = cmdMini.Flag.String("vectorBucket", "", "comma-separated vector buckets to create; falls back to S3_VECTOR_BUCKET")
 )
 
 // getBindIp determines the bind IP address based on miniIp and miniBindIp flags
@@ -874,9 +880,9 @@ func ensurePortAvailableOnIP(portPtr *int, serviceName string, ip string, reserv
 	// Skip if this port is reserved for gRPC calculation
 	if reservedPorts[original] {
 		if isExplicitPort {
-			return fmt.Errorf("port %d for %s (specified by flag %s) is reserved for gRPC calculation and cannot be used", original, serviceName, flagName)
+			return fmt.Errorf("port %d for %s (specified by flag %s) is reserved by another mini service and cannot be used", original, serviceName, flagName)
 		}
-		glog.Warningf("Port %d for %s is reserved for gRPC calculation, finding alternative...", original, serviceName)
+		glog.Warningf("Port %d for %s is reserved by another mini service, finding alternative...", original, serviceName)
 		newPort := findAvailablePortOnIP(ip, original+1, 100, reservedPorts, listenOnLocalhost)
 		if newPort == 0 {
 			glog.Errorf("Could not find available port for %s starting from %d, will use original %d and fail on binding", serviceName, original+1, original)
@@ -950,22 +956,41 @@ func ensureAllPortsAvailableOnIP(bindIp string, listenOnLocalhost bool) error {
 			}{miniS3Options.portLance, "Lance", "s3.port.lance", nil, listenOnLocalhost})
 		}
 	}
-	// WebDAV and the Admin UI bind only -ip.bind; they never open a loopback
-	// companion listener, so ports held elsewhere cannot collide with them.
+	if *miniVector {
+		portConfigs = append(portConfigs, struct {
+			port      *int
+			name      string
+			flagName  string
+			grpcPtr   *int
+			localhost bool
+		}{miniVectorPort, "Vector", "vector.port", nil, false})
+		if *miniVectorMetricsPort > 0 {
+			portConfigs = append(portConfigs, struct {
+				port      *int
+				name      string
+				flagName  string
+				grpcPtr   *int
+				localhost bool
+			}{miniVectorMetricsPort, "Vector metrics", "vector.metricsPort", nil, false})
+		}
+	}
+
+	if *miniEnableWebDAV {
+		portConfigs = append(portConfigs, struct {
+			port      *int
+			name      string
+			flagName  string
+			grpcPtr   *int
+			localhost bool
+		}{miniWebDavOptions.port, "WebDAV", "webdav.port", nil, false})
+	}
 	portConfigs = append(portConfigs, struct {
 		port      *int
 		name      string
 		flagName  string
 		grpcPtr   *int
 		localhost bool
-	}{miniWebDavOptions.port, "WebDAV", "webdav.port", nil, false},
-		struct {
-			port      *int
-			name      string
-			flagName  string
-			grpcPtr   *int
-			localhost bool
-		}{miniAdminOptions.port, "Admin", "admin.port", miniAdminOptions.grpcPort, false})
+	}{miniAdminOptions.port, "Admin", "admin.port", miniAdminOptions.grpcPort, false})
 
 	// First, reserve all gRPC ports that will be calculated to prevent HTTP port allocation from using them
 	// This prevents collisions like: HTTP port moves to X, then gRPC port is calculated as Y where Y == X
@@ -978,6 +1003,16 @@ func ensureAllPortsAvailableOnIP(bindIp string, listenOnLocalhost bool) error {
 		}
 	}
 
+	chosenPorts := make(map[int]string)
+	for _, config := range portConfigs {
+		if *config.port > 0 && explicitFlags[config.flagName] {
+			if previous, exists := chosenPorts[*config.port]; exists {
+				return fmt.Errorf("port %d is shared by explicit flags %s and %s", *config.port, previous, config.flagName)
+			}
+			chosenPorts[*config.port] = config.flagName
+		}
+	}
+
 	applyMiniAdminCredentialFallback(&miniAdminOptions)
 	adminBindIP := miniAdminEffectiveBindIP(bindIp)
 
@@ -991,9 +1026,19 @@ func ensureAllPortsAvailableOnIP(bindIp string, listenOnLocalhost bool) error {
 			checkIP = adminBindIP
 		}
 		original := *config.port
-		if err := ensurePortAvailableOnIP(config.port, config.name, checkIP, reservedPorts, config.flagName, config.localhost); err != nil {
+		unavailablePorts := maps.Clone(reservedPorts)
+		for port, flag := range chosenPorts {
+			if flag != config.flagName {
+				unavailablePorts[port] = true
+			}
+		}
+
+		if err := ensurePortAvailableOnIP(config.port, config.name, checkIP, unavailablePorts, config.flagName, config.localhost); err != nil {
 			return err
 		}
+		if *config.port > 0 {
+			chosenPorts[*config.port] = config.flagName
+		}
 		// If port was changed, update the reserved gRPC ports mapping
 		if *config.port != original && config.grpcPtr != nil && *config.grpcPtr == 0 {
 			delete(reservedPorts, original+GrpcPortOffset)
@@ -1105,6 +1150,13 @@ func initializeGrpcPortsOnIP(bindIp string, listenOnLocalhost bool) map[int]bool
 		}
 	}
 
+	if *miniVector {
+		allocatedPorts[*miniVectorPort] = true
+		if *miniVectorMetricsPort > 0 {
+			allocatedPorts[*miniVectorMetricsPort] = true
+		}
+	}
+
 	grpcConfigs := []struct {
 		httpPort  *int
 		grpcPort  *int
@@ -1641,6 +1693,32 @@ func runMini(cmd *Command, args []string) bool {
 		os.Unsetenv("S3_TABLE_BUCKET")
 	}
 
+	vectorBuckets := *miniVectorBucket
+	if vectorBuckets == "" {
+		vectorBuckets = os.Getenv("S3_VECTOR_BUCKET")
+	}
+	var vectorEntries []tableBucketEntry
+	for _, name := range parseBucketList(vectorBuckets) {
+		vectorEntries = append(vectorEntries, tableBucketEntry{name: name, format: s3tables.FormatLance})
+	}
+	if len(vectorEntries) > 0 && !*miniEnableS3 {
+		glog.Errorf("vector buckets require the S3 gateway")
+		triggerMiniClientsShutdown(5 * time.Second)
+		return false
+	}
+	if ready, err := createMiniTableBuckets(vectorEntries); err != nil || len(ready) != len(vectorEntries) {
+		glog.Errorf("created %d of %d requested vector buckets: %v", len(ready), len(vectorEntries), err)
+		triggerMiniClientsShutdown(5 * time.Second)
+		return false
+	}
+	if *miniVector {
+		if err := startMiniVector(); err != nil {
+			glog.Errorf("start vector engine: %v", err)
+			triggerMiniClientsShutdown(5 * time.Second)
+			return false
+		}
+	}
+
 	// Print welcome message after all services are running
 	printWelcomeMessage()
 
@@ -2423,6 +2501,14 @@ func ensureMiniTableBuckets(buckets []tableBucketEntry) ([]tableBucketEntry, err
 		return nil, nil
 	}
 
+	return createMiniTableBuckets(servable)
+}
+
+func createMiniTableBuckets(servable []tableBucketEntry) ([]tableBucketEntry, error) {
+	if len(servable) == 0 {
+		return nil, nil
+	}
+
 	filerAddress := pb.NewServerAddress(*miniIp, *miniFilerOptions.port, *miniFilerOptions.portGrpc)
 	grpcDialOption := security.LoadClientTLS(util.GetViper(), "grpc.client")
 

weed/command/mini_vector.go

@@ -0,0 +1,130 @@
+package command
+
+import (
+	"context"
+	"fmt"
+	"net"
+	"os"
+	"os/exec"
+	"path/filepath"
+	"time"
+
+	"github.com/seaweedfs/seaweedfs/weed/glog"
+	"github.com/seaweedfs/seaweedfs/weed/pb"
+	"github.com/seaweedfs/seaweedfs/weed/pb/master_pb"
+	"github.com/seaweedfs/seaweedfs/weed/security"
+	"github.com/seaweedfs/seaweedfs/weed/util"
+)
+
+func startMiniVector() error {
+	binary := *miniVectorBinary
+	if binary == "weed-vector" {
+		if executable, err := os.Executable(); err == nil {
+			candidate := filepath.Join(filepath.Dir(executable), binary)
+			if resolved, err := exec.LookPath(candidate); err == nil {
+				binary = resolved
+			}
+		}
+	}
+	if _, err := exec.LookPath(binary); err != nil {
+		return fmt.Errorf("%s is unavailable: install the vector component or set -vector.binary: %w", binary, err)
+	}
+	address := util.JoinHostPort(*miniIp, *miniVectorPort)
+	bindIP := *miniBindIp
+	if bindIP == "" {
+		bindIP = *miniIp
+	}
+	if net.ParseIP(bindIP) == nil {
+		ctx, cancel := context.WithTimeout(miniClientsCtx(), 5*time.Second)
+		addresses, err := net.DefaultResolver.LookupIPAddr(ctx, bindIP)
+		cancel()
+		if err != nil {
+			return fmt.Errorf("resolve vector bind address %s: %w", bindIP, err)
+		}
+		if len(addresses) == 0 {
+			return fmt.Errorf("vector bind address %s has no IP addresses", bindIP)
+		}
+		bindIP = addresses[0].String()
+	}
+
+	args := []string{"--listen=" + util.JoinHostPort(bindIP, *miniVectorPort), "--advertise=" + address, "--masters=" + string(pb.NewServerAddress(*miniIp, *miniMasterOptions.port, *miniMasterOptions.portGrpc))}
+	if miniFilerOptions.filerGroup != nil && *miniFilerOptions.filerGroup != "" {
+		args = append(args, "--filer-group="+*miniFilerOptions.filerGroup)
+	}
+	if *miniVectorMetricsPort > 0 {
+		args = append(args, fmt.Sprintf("--metrics-port=%d", *miniVectorMetricsPort), "--metrics-ip="+bindIP)
+	}
+	if directory := util.ConfigurationFileDirectory.String(); directory != "" {
+		args = append(args, "--config-dir="+util.ResolvePath(directory))
+	}
+	command := exec.CommandContext(miniClientsCtx(), binary, args...)
+	command.Stdout, command.Stderr = os.Stdout, os.Stderr
+	if err := command.Start(); err != nil {
+		return err
+	}
+	done := trackMiniClient()
+	exited := make(chan error, 1)
+	go func() {
+		defer done()
+		err := command.Wait()
+		exited <- err
+		if err != nil && miniClientsCtx().Err() == nil {
+			glog.Errorf("vector engine exited: %v", err)
+		}
+	}()
+	ctx, cancel := context.WithTimeout(miniClientsCtx(), 30*time.Second)
+	defer cancel()
+	ticker := time.NewTicker(100 * time.Millisecond)
+	defer ticker.Stop()
+	for {
+		select {
+		case err := <-exited:
+			return fmt.Errorf("vector engine exited before readiness: %v", err)
+		case <-ctx.Done():
+			_ = command.Process.Kill()
+			return ctx.Err()
+		case <-ticker.C:
+			conn, err := net.DialTimeout("tcp", miniVectorLocalAddress(bindIP, *miniVectorPort), 100*time.Millisecond)
+			if err == nil {
+				_ = conn.Close()
+				if !miniVectorRegistered(ctx, address) {
+					continue
+				}
+				glog.V(0).Infof("vector engine ready on %s", address)
+				return nil
+			}
+		}
+	}
+}
+
+func miniVectorLocalAddress(bindIP string, port int) string {
+	if ip := net.ParseIP(bindIP); ip != nil && ip.IsUnspecified() {
+		if ip.To4() != nil {
+			bindIP = "127.0.0.1"
+		} else {
+			bindIP = "::1"
+		}
+	}
+	return util.JoinHostPort(bindIP, port)
+}
+
+func miniVectorRegistered(ctx context.Context, address string) bool {
+	ctx, cancel := context.WithTimeout(ctx, time.Second)
+	defer cancel()
+	master := pb.NewServerAddress(*miniIp, *miniMasterOptions.port, *miniMasterOptions.portGrpc)
+	ready := false
+	_ = pb.WithMasterClient(ctx, false, master, security.LoadClientTLS(util.GetViper(), "grpc.client"), false, func(client master_pb.SeaweedClient) error {
+		response, err := client.ListClusterNodes(ctx, &master_pb.ListClusterNodesRequest{ClientType: "vector", FilerGroup: *miniFilerOptions.filerGroup})
+		if err != nil {
+			return err
+		}
+		for _, node := range response.ClusterNodes {
+			if node.Address == address {
+				ready = true
+				break
+			}
+		}
+		return nil
+	})
+	return ready
+}

weed/command/mini_vector_test.go

@@ -0,0 +1,21 @@
+package command
+
+import (
+	"strings"
+	"testing"
+)
+
+func TestMiniVectorRejectsExplicitListenerConflict(t *testing.T) {
+	oldVector, oldPort, oldMetrics, oldFlags := *miniVector, *miniVectorPort, *miniVectorMetricsPort, explicitFlags
+	defer func() {
+		*miniVector, *miniVectorPort, *miniVectorMetricsPort, explicitFlags = oldVector, oldPort, oldMetrics, oldFlags
+	}()
+	*miniVector = true
+	*miniVectorPort = 8090
+	*miniVectorMetricsPort = 8090
+	explicitFlags = map[string]bool{"vector.port": true, "vector.metricsPort": true}
+	err := ensureAllPortsAvailableOnIP("127.0.0.1", false)
+	if err == nil || !strings.Contains(err.Error(), "shared by explicit flags vector.port and vector.metricsPort") {
+		t.Fatalf("listener conflict not rejected: %v", err)
+	}
+}

weed/command/s3.go

@@ -10,7 +10,6 @@ import (
 	"runtime"
 	"strconv"
 	"strings"
-	"sync"
 	"sync/atomic"
 	"time"
 
@@ -20,10 +19,8 @@ import (
 	"github.com/seaweedfs/seaweedfs/weed/glog"
 	"github.com/seaweedfs/seaweedfs/weed/pb"
 	"github.com/seaweedfs/seaweedfs/weed/pb/filer_pb"
-	"github.com/seaweedfs/seaweedfs/weed/pb/master_pb"
 	"github.com/seaweedfs/seaweedfs/weed/pb/s3_lifecycle_pb"
 	"github.com/seaweedfs/seaweedfs/weed/pb/s3_pb"
-	"github.com/seaweedfs/seaweedfs/weed/pb/vector_pb"
 	"github.com/seaweedfs/seaweedfs/weed/s3api"
 	"github.com/seaweedfs/seaweedfs/weed/s3api/iceberg"
 	"github.com/seaweedfs/seaweedfs/weed/s3api/lance"
@@ -37,8 +34,6 @@ import (
 	util_http "github.com/seaweedfs/seaweedfs/weed/util/http"
 	"github.com/seaweedfs/seaweedfs/weed/util/version"
 	"google.golang.org/grpc"
-	"google.golang.org/grpc/codes"
-	"google.golang.org/grpc/status"
 )
 
 var (
@@ -452,7 +447,7 @@ func (s3opt *S3Options) startS3Server() bool {
 	// backend is reachable — /s3vectors/ in the SigV4 credential scope picks
 	// the vectors router.
 	vectorsHandler := http.Handler(router)
-	if vectorsRouter := s3opt.vectorsRouter(s3ApiServer, grpcDialOption, masterAddresses); vectorsRouter != nil {
+	if vectorsRouter := s3opt.vectorsRouter(s3ApiServer, grpcDialOption, masterAddresses, filerGroup); vectorsRouter != nil {
 		vr, mainRouter := vectorsRouter, router
 		vectorsHandler = http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
 			if isVectorsRequest(r) {
@@ -739,12 +734,12 @@ func (s3opt *S3Options) startLanceServer(listenOnLocalhost bool) *http.Server {
 // 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.
-func (s3opt *S3Options) vectorsRouter(s3ApiServer *s3api.S3ApiServer, grpcDialOption grpc.DialOption, masterAddresses []pb.ServerAddress) *mux.Router {
+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")
 		return nil
 	}
-	backend := &vectorDiscovery{masters: masterAddresses, grpcDialOption: grpcDialOption, discover: discoverVectorWorkers}
+	backend := s3vectors.NewVectorDiscovery(func() []pb.ServerAddress { return masterAddresses }, grpcDialOption, filerGroup)
 	// In-process describe — the router resolves locations and vended
 	// storage options through the catalog directly, never over HTTP. The
 	// lookup itself runs under the trusted identity: the caller's vector
@@ -769,230 +764,6 @@ func (s3opt *S3Options) vectorsRouter(s3ApiServer *s3api.S3ApiServer, grpcDialOp
 	return vectorsRouter
 }
 
-// vectorDiscovery resolves the currently registered weed-vector through the
-// master and caches the connection; workers joining after gateway startup
-// are picked up on first query.
-type vectorDiscovery struct {
-	masters        []pb.ServerAddress
-	grpcDialOption grpc.DialOption
-	discover       func([]pb.ServerAddress, grpc.DialOption) []string
-	mu             sync.Mutex
-	worker         string
-	conn           *grpc.ClientConn
-	client         vector_pb.VectorServiceClient
-	nextResolve    time.Time
-	resolving      chan struct{} // closed when the in-flight resolve lands
-}
-
-// reset drops the cached worker so the next call rediscovers — invoked on
-// transport errors only, and only when the failed client is still the
-// cached one (a concurrent request may have already re-resolved).
-func (d *vectorDiscovery) reset(c vector_pb.VectorServiceClient) {
-	d.mu.Lock()
-	defer d.mu.Unlock()
-	if c != d.client {
-		return
-	}
-	if d.conn != nil {
-		d.conn.Close()
-	}
-	d.worker, d.conn, d.client = "", nil, nil
-	d.nextResolve = time.Time{}
-}
-
-// Only Unavailable proves a transport failure — a request deadline or an
-// application-level error can fire on a healthy connection.
-func isTransportError(err error) bool {
-	return status.Code(err) == codes.Unavailable
-}
-
-// resolve runs without d.mu held so a slow master cannot stall cached-client callers.
-func (d *vectorDiscovery) resolve() {
-	workers := d.discover(d.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.
-	defer func() {
-		interval := vectorRecheckInterval
-		if d.client != nil {
-			interval = vectorResolveInterval
-		}
-		d.nextResolve = time.Now().Add(interval)
-		close(d.resolving)
-		d.resolving = nil
-	}()
-	if len(workers) == 0 {
-		return
-	}
-	worker := workers[0]
-	for _, w := range workers {
-		if w == d.worker {
-			worker = w
-			break
-		}
-	}
-	if worker != d.worker {
-		conn, err := pb.GrpcDial(context.Background(), worker, false, d.grpcDialOption)
-		if err != nil {
-			glog.Warningf("S3 Vectors: dial weed-vector %s: %v", worker, err)
-			return
-		}
-		if d.conn != nil {
-			d.conn.Close()
-		}
-		d.worker, d.conn = worker, conn
-		d.client = vector_pb.NewVectorServiceClient(conn)
-	}
-}
-
-// 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()
-	if d.resolving == nil && !time.Now().Before(d.nextResolve) {
-		d.resolving = make(chan struct{})
-		go d.resolve()
-	}
-	cli, resolving := d.client, d.resolving
-	d.mu.Unlock()
-	if cli != nil {
-		return cli, nil
-	}
-	// No cached client: wait for the in-flight lookup instead of failing.
-	if resolving != nil {
-		<-resolving
-		d.mu.Lock()
-		cli = d.client
-		d.mu.Unlock()
-		if cli != nil {
-			return cli, nil
-		}
-	}
-	return nil, errNoVectorWorker
-}
-
-func (d *vectorDiscovery) CreateIndex(ctx context.Context, in *vector_pb.CreateIndexRequest, opts ...grpc.CallOption) (resp *vector_pb.CreateIndexResponse, err error) {
-	for attempt := 0; attempt < 2; attempt++ {
-		var c vector_pb.VectorServiceClient
-		if c, err = d.cli(); err != nil {
-			return nil, err
-		}
-		if resp, err = c.CreateIndex(ctx, in, opts...); err == nil {
-			return resp, nil
-		}
-		if !isTransportError(err) {
-			return nil, err
-		}
-		d.reset(c)
-	}
-	return nil, err
-}
-func (d *vectorDiscovery) DropIndex(ctx context.Context, in *vector_pb.DropIndexRequest, opts ...grpc.CallOption) (resp *vector_pb.DropIndexResponse, err error) {
-	for attempt := 0; attempt < 2; attempt++ {
-		var c vector_pb.VectorServiceClient
-		if c, err = d.cli(); err != nil {
-			return nil, err
-		}
-		if resp, err = c.DropIndex(ctx, in, opts...); err == nil {
-			return resp, nil
-		}
-		if !isTransportError(err) {
-			return nil, err
-		}
-		d.reset(c)
-	}
-	return nil, err
-}
-func (d *vectorDiscovery) PutVectors(ctx context.Context, in *vector_pb.PutVectorsRequest, opts ...grpc.CallOption) (resp *vector_pb.PutVectorsResponse, err error) {
-	for attempt := 0; attempt < 2; attempt++ {
-		var c vector_pb.VectorServiceClient
-		if c, err = d.cli(); err != nil {
-			return nil, err
-		}
-		if resp, err = c.PutVectors(ctx, in, opts...); err == nil {
-			return resp, nil
-		}
-		if !isTransportError(err) {
-			return nil, err
-		}
-		d.reset(c)
-	}
-	return nil, err
-}
-func (d *vectorDiscovery) GetVectors(ctx context.Context, in *vector_pb.GetVectorsRequest, opts ...grpc.CallOption) (resp *vector_pb.GetVectorsResponse, err error) {
-	for attempt := 0; attempt < 2; attempt++ {
-		var c vector_pb.VectorServiceClient
-		if c, err = d.cli(); err != nil {
-			return nil, err
-		}
-		if resp, err = c.GetVectors(ctx, in, opts...); err == nil {
-			return resp, nil
-		}
-		if !isTransportError(err) {
-			return nil, err
-		}
-		d.reset(c)
-	}
-	return nil, err
-}
-func (d *vectorDiscovery) DeleteVectors(ctx context.Context, in *vector_pb.DeleteVectorsRequest, opts ...grpc.CallOption) (resp *vector_pb.DeleteVectorsResponse, err error) {
-	for attempt := 0; attempt < 2; attempt++ {
-		var c vector_pb.VectorServiceClient
-		if c, err = d.cli(); err != nil {
-			return nil, err
-		}
-		if resp, err = c.DeleteVectors(ctx, in, opts...); err == nil {
-			return resp, nil
-		}
-		if !isTransportError(err) {
-			return nil, err
-		}
-		d.reset(c)
-	}
-	return nil, err
-}
-func (d *vectorDiscovery) ListVectors(ctx context.Context, in *vector_pb.ListVectorsRequest, opts ...grpc.CallOption) (resp *vector_pb.ListVectorsResponse, err error) {
-	for attempt := 0; attempt < 2; attempt++ {
-		var c vector_pb.VectorServiceClient
-		if c, err = d.cli(); err != nil {
-			return nil, err
-		}
-		if resp, err = c.ListVectors(ctx, in, opts...); err == nil {
-			return resp, nil
-		}
-		if !isTransportError(err) {
-			return nil, err
-		}
-		d.reset(c)
-	}
-	return nil, err
-}
-func (d *vectorDiscovery) QueryVectors(ctx context.Context, in *vector_pb.QueryVectorsRequest, opts ...grpc.CallOption) (resp *vector_pb.QueryVectorsResponse, err error) {
-	for attempt := 0; attempt < 2; attempt++ {
-		var c vector_pb.VectorServiceClient
-		if c, err = d.cli(); err != nil {
-			return nil, err
-		}
-		if resp, err = c.QueryVectors(ctx, in, opts...); err == nil {
-			return resp, nil
-		}
-		if !isTransportError(err) {
-			return nil, err
-		}
-		d.reset(c)
-	}
-	return nil, err
-}
-
 // isVectorsRequest routes by the SigV4 credential scope — the s3vectors
 // service name lives there, not in the path.
 func isVectorsRequest(r *http.Request) bool {
@@ -1134,32 +905,3 @@ func (v lanceCredentialVendor) VendTableCredentials(ctx context.Context, princip
 // 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.
-func discoverVectorWorkers(masters []pb.ServerAddress, grpcDialOption grpc.DialOption) []string {
-	for _, master := range masters {
-		ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
-		conn, err := pb.GrpcDial(ctx, master.ToGrpcAddress(), false, grpcDialOption)
-		if err != nil {
-			cancel()
-			continue
-		}
-		resp, err := master_pb.NewSeaweedClient(conn).ListClusterNodes(ctx, &master_pb.ListClusterNodesRequest{
-			ClientType: "vector",
-		})
-		conn.Close()
-		cancel()
-		if err != nil {
-			continue
-		}
-		workers := 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)
-		}
-		// A follower can answer with an empty list — workers register with
-		// the leader, so keep looking rather than accept "none."
-		if len(workers) > 0 {
-			return workers
-		}
-	}
-	return nil
-}

weed/command/s3_vector_discovery_test.go

@@ -1,245 +0,0 @@
-package command
-
-import (
-	"fmt"
-	"sync/atomic"
-	"testing"
-	"time"
-
-	"github.com/seaweedfs/seaweedfs/weed/pb"
-	"google.golang.org/grpc"
-	"google.golang.org/grpc/codes"
-	"google.golang.org/grpc/connectivity"
-	"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)
-	}
-}
-
-// A lookup slower than the throttle interval (an unreachable first master
-// costs the full dial timeout) must not leave the next lookup already due,
-// or every request queues behind back-to-back lookups.
-func TestVectorDiscoveryThrottlesFromLookupEnd(t *testing.T) {
-	const lookup = 600 * time.Millisecond
-	var calls int32
-	d := &vectorDiscovery{
-		grpcDialOption: grpc.WithTransportCredentials(insecure.NewCredentials()),
-		discover: func(masters []pb.ServerAddress, opt grpc.DialOption) []string {
-			atomic.AddInt32(&calls, 1)
-			time.Sleep(lookup)
-			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("slow lookup was repeated right away: %d lookups", got)
-	}
-
-	d.discover = func(masters []pb.ServerAddress, opt grpc.DialOption) []string {
-		time.Sleep(lookup)
-		return []string{"127.0.0.1:9301"}
-	}
-	for i := 0; i < 2; i++ {
-		d.mu.Lock()
-		d.nextResolve = time.Time{}
-		d.mu.Unlock()
-		if _, err := d.cli(); err != nil {
-			t.Fatalf("worker registered: expected a client, got %v", err)
-		}
-		waitResolved(d)
-	}
-	d.mu.Lock()
-	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)
-	}
-}
-
-func waitResolved(d *vectorDiscovery) {
-	d.mu.Lock()
-	resolving := d.resolving
-	d.mu.Unlock()
-	if resolving != nil {
-		<-resolving
-	}
-}
-
-// 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.
-func TestVectorDiscoveryLookupDoesNotBlockCachedClient(t *testing.T) {
-	d := &vectorDiscovery{
-		grpcDialOption: grpc.WithTransportCredentials(insecure.NewCredentials()),
-		discover: func(masters []pb.ServerAddress, opt grpc.DialOption) []string {
-			return []string{"127.0.0.1:9301"}
-		},
-	}
-	cached, err := d.cli()
-	if err != nil {
-		t.Fatalf("worker registered: expected a client, got %v", err)
-	}
-
-	var calls int32
-	entered, release := make(chan struct{}), make(chan struct{})
-	d.discover = func(masters []pb.ServerAddress, opt grpc.DialOption) []string {
-		if atomic.AddInt32(&calls, 1) == 1 {
-			close(entered)
-		}
-		<-release
-		return []string{"127.0.0.1:9301"}
-	}
-	d.mu.Lock()
-	d.nextResolve = time.Time{}
-	d.mu.Unlock()
-
-	const callers = 8
-	got := make(chan error, callers+1)
-	call := func() {
-		c, err := d.cli()
-		if err == nil && c != cached {
-			err = fmt.Errorf("got a different client")
-		}
-		got <- err
-	}
-	go call()
-	<-entered
-	for i := 0; i < callers; i++ {
-		go call()
-	}
-	timeout := time.After(2 * time.Second)
-	for i := 0; i <= callers; i++ {
-		select {
-		case err := <-got:
-			if err != nil {
-				close(release)
-				t.Fatalf("cached client: %v", err)
-			}
-		case <-timeout:
-			close(release)
-			t.Fatalf("cached client blocked behind an in-flight lookup")
-		}
-	}
-	close(release)
-	waitResolved(d)
-	if n := atomic.LoadInt32(&calls); n != 1 {
-		t.Fatalf("concurrent callers ran %d lookups, want 1", n)
-	}
-}
-
-// Callers with no worker wait for the in-flight lookup and share its result
-// rather than failing while it runs.
-func TestVectorDiscoveryUncachedCallersShareLookup(t *testing.T) {
-	var calls int32
-	release := make(chan struct{})
-	d := &vectorDiscovery{
-		grpcDialOption: grpc.WithTransportCredentials(insecure.NewCredentials()),
-		discover: func(masters []pb.ServerAddress, opt grpc.DialOption) []string {
-			atomic.AddInt32(&calls, 1)
-			<-release
-			return []string{"127.0.0.1:9301"}
-		},
-	}
-	const callers = 8
-	got := make(chan error, callers)
-	for i := 0; i < callers; i++ {
-		go func() {
-			_, err := d.cli()
-			got <- err
-		}()
-	}
-	time.Sleep(50 * time.Millisecond)
-	close(release)
-	for i := 0; i < callers; i++ {
-		if err := <-got; err != nil {
-			t.Fatalf("caller waiting on the lookup: %v", err)
-		}
-	}
-	if n := atomic.LoadInt32(&calls); n != 1 {
-		t.Fatalf("concurrent callers ran %d lookups, want 1", n)
-	}
-}
-
-// A failed dial to a new worker must leave the cached connection open.
-func TestVectorDiscoveryKeepsConnOnDialError(t *testing.T) {
-	d := &vectorDiscovery{
-		grpcDialOption: grpc.WithTransportCredentials(insecure.NewCredentials()),
-		discover: func(masters []pb.ServerAddress, opt grpc.DialOption) []string {
-			return []string{"127.0.0.1:9301"}
-		},
-	}
-	cached, err := d.cli()
-	if err != nil {
-		t.Fatalf("worker registered: expected a client, got %v", err)
-	}
-
-	// Without transport credentials the dial fails synchronously.
-	d.grpcDialOption = nil
-	d.discover = func(masters []pb.ServerAddress, opt grpc.DialOption) []string {
-		return []string{"127.0.0.1:9302"}
-	}
-	d.mu.Lock()
-	d.nextResolve = time.Time{}
-	d.mu.Unlock()
-	if c, err := d.cli(); err != nil || c != cached {
-		t.Fatalf("cached client: got %v, %v", c, err)
-	}
-	waitResolved(d)
-
-	d.mu.Lock()
-	defer d.mu.Unlock()
-	if d.client != cached || d.conn.GetState() == connectivity.Shutdown {
-		t.Fatalf("failed dial dropped the cached connection")
-	}
-}

weed/command/shell_s3vectors.go

@@ -0,0 +1,101 @@
+package command
+
+import (
+	"context"
+	"encoding/json"
+	"flag"
+	"fmt"
+	"io"
+	"net/http"
+	"os"
+	"slices"
+	"time"
+
+	"github.com/seaweedfs/seaweedfs/weed/s3api/lance"
+	"github.com/seaweedfs/seaweedfs/weed/s3api/s3_constants"
+	"github.com/seaweedfs/seaweedfs/weed/s3api/s3tables"
+	"github.com/seaweedfs/seaweedfs/weed/s3api/s3vectors"
+	"github.com/seaweedfs/seaweedfs/weed/shell"
+)
+
+func init() { shell.Commands = append(shell.Commands, &commandS3Vectors{}) }
+
+type commandS3Vectors struct{}
+
+func (c *commandS3Vectors) Name() string                 { return "s3vectors" }
+func (c *commandS3Vectors) HasTag(shell.CommandTag) bool { return false }
+func (c *commandS3Vectors) Help() string {
+	return `manage vector buckets, indexes, policies, tags and vectors
+
+s3vectors -operation CreateVectorBucket -request '{"vectorBucketName":"embeddings"}'
+s3vectors -operation ListVectorBuckets
+s3vectors -operation CreateIndex -file index.json
+s3vectors -operation QueryVectors -file query.json
+
+Requests use the S3 Vectors API JSON shape. The shell uses trusted cluster
+credentials. Index and vector operations require a registered vector engine.
+`
+}
+
+func (c *commandS3Vectors) Do(args []string, env *shell.CommandEnv, writer io.Writer) error {
+	flags := flag.NewFlagSet(c.Name(), flag.ContinueOnError)
+	operation := flags.String("operation", "", "S3 Vectors operation")
+	request := flags.String("request", "", "request JSON")
+	file := flags.String("file", "", "request JSON file")
+	if err := flags.Parse(args); err != nil {
+		return err
+	}
+	if !slices.Contains(s3vectors.Operations(), *operation) {
+		return fmt.Errorf("unknown operation %q; supported: %v", *operation, s3vectors.Operations())
+	}
+	if *request != "" && *file != "" {
+		return fmt.Errorf("use either -request or -file")
+	}
+	body := []byte(*request)
+	if *file != "" {
+		reader, err := os.Open(*file)
+		if err != nil {
+			return err
+		}
+		defer reader.Close()
+		body, err = io.ReadAll(io.LimitReader(reader, s3_constants.S3VectorsMaxRequestBodyBytes+1))
+		if err != nil {
+			return err
+		}
+	}
+	if len(body) == 0 {
+		body = []byte(`{}`)
+	}
+	if len(body) > s3_constants.S3VectorsMaxRequestBodyBytes {
+		return fmt.Errorf("request is too large")
+	}
+	if !json.Valid(body) {
+		return fmt.Errorf("request must be valid JSON")
+	}
+	ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second)
+	defer cancel()
+	backend := s3vectors.NewVectorDiscovery(env.MasterClient.ListCachedMasters, env.GrpcDialOption(), env.MasterClient.FilerGroup)
+	defer backend.Close()
+	server := s3vectors.NewServer(env, nil, backend)
+	server.SetRegion(s3tables.DefaultRegion)
+	server.SetAccountID(s3tables.DefaultAccountID)
+	catalog := lance.NewServer(env, nil)
+	server.SetDescribe(func(ctx context.Context, id []string) (string, map[string]string, error) {
+		r := (&http.Request{Method: http.MethodPost, Header: make(http.Header)}).WithContext(ctx)
+		return catalog.DescribeTableInternal(r, id)
+	})
+	var response json.RawMessage
+	if err := server.Execute(ctx, *operation, json.RawMessage(body), &response, s3_constants.AccountAdminId); err != nil {
+		return err
+	}
+	if len(response) > 0 {
+		var value interface{}
+		if err := json.Unmarshal(response, &value); err != nil {
+			return err
+		}
+		encoder := json.NewEncoder(writer)
+		encoder.SetIndent("", "  ")
+		return encoder.Encode(value)
+	}
+	return nil
+}