From 511727cf3a020823647fda89518f13495d79097e Mon Sep 17 00:00:00 2001 From: Chris Lu Date: Wed, 28 Oct 2026 09:22:05 -0700 Subject: [PATCH] vector: DeleteIndex format guard, sticky discovery, advertise normalize (#12171) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit * vector: DeleteIndex format guard, sticky discovery, advertise normalize - handleDeleteIndex verifies the table carries the Lance format before DeleteTable — an Iceberg table under the same namespace is not an index and must not be removed by it. - discovery keeps the current worker while it is still registered — membership-order churn can no longer cancel in-flight calls, and the resolver closes each master conn inside a 5s deadline. - advertised_addr strips scheme and .grpcPort so the UDP probe reaches every supported master form. * vector: empty-follower retry, format check without read grant - discoverVectorWorkers keeps looking past an empty master reply — followers list nothing when workers registered with the leader. - the DeleteIndex format check uses the internal catalog read, so DeleteTable-only callers don't gain a GetTable requirement, and the format must be Lance exactly — format-less tables no longer pass. --- seaweed-vector/crates/server/src/main.rs | 15 ++++++++++- weed/command/s3.go | 34 +++++++++++++++++++----- weed/s3api/s3vectors/handlers.go | 20 ++++++++++++++ 3 files changed, 61 insertions(+), 8 deletions(-) diff --git a/seaweed-vector/crates/server/src/main.rs b/seaweed-vector/crates/server/src/main.rs index ca3f9ea077..de52878e47 100644 --- a/seaweed-vector/crates/server/src/main.rs +++ b/seaweed-vector/crates/server/src/main.rs @@ -46,8 +46,21 @@ fn advertised_addr(listen: &str, masters: &[String]) -> String { } let port = listen.rsplit_once(':').map(|(_, p)| p).unwrap_or_default(); if let Some(master) = masters.first() { + // Masters may arrive as host:http.grpc or scheme://host:http — + // the UDP probe only needs a dialable host:port. + let probe = master + .rsplit_once("://") + .map(|(_, r)| r) + .unwrap_or(master); + let probe = match probe.rsplit_once(':') { + Some((h, p)) => match p.split_once('.') { + Some((_, g)) => format!("{h}:{g}"), + None => probe.to_string(), + }, + None => probe.to_string(), + }; if let Ok(sock) = std::net::UdpSocket::bind((host, 0u16)) { - if sock.connect(master).is_ok() { + if sock.connect(&probe).is_ok() { if let Ok(local) = sock.local_addr() { return format!("{}:{}", local.ip(), port); } diff --git a/weed/command/s3.go b/weed/command/s3.go index 6d2802246b..cb9f4d8b33 100644 --- a/weed/command/s3.go +++ b/weed/command/s3.go @@ -765,10 +765,17 @@ func isTransportError(err error) bool { func (d *vectorDiscovery) resolve() (vector_pb.VectorServiceClient, error) { d.mu.Lock() defer d.mu.Unlock() - worker := discoverVectorWorker(d.masters, d.grpcDialOption) - if worker == "" { + workers := discoverVectorWorkers(d.masters, d.grpcDialOption) + if len(workers) == 0 { return nil, fmt.Errorf("no weed-vector registered on master") } + worker := workers[0] + for _, w := range workers { + if w == d.worker { + worker = w + break + } + } if worker != d.worker { if d.conn != nil { d.conn.Close() @@ -1053,22 +1060,35 @@ func (v lanceCredentialVendor) VendTableCredentials(ctx context.Context, princip // discoverVectorWorker resolves one registered weed-vector server through the // master's cluster-nodes list; OSS fans every vector call to that single // backend. -func discoverVectorWorker(masters []pb.ServerAddress, grpcDialOption grpc.DialOption) string { +// 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 { - conn, err := pb.GrpcDial(context.Background(), master.ToGrpcAddress(), false, grpcDialOption) + 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(context.Background(), &master_pb.ListClusterNodesRequest{ + 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) - return node.Address + 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 "" + return nil } diff --git a/weed/s3api/s3vectors/handlers.go b/weed/s3api/s3vectors/handlers.go index c887b9843e..6e57250ca1 100644 --- a/weed/s3api/s3vectors/handlers.go +++ b/weed/s3api/s3vectors/handlers.go @@ -662,6 +662,26 @@ func (s *Server) handleDeleteIndex(w http.ResponseWriter, r *http.Request) { writeError(w, http.StatusBadRequest, "ValidationException", err.Error()) return } + // A named table is an index only while it carries the Lance format — + // an Iceberg or format-less table under the same namespace is not an + // index and must not be deleted by this operation. + // The format check needs no read grant — the delete below is the + // authorization boundary — and only a Lance-formatted record is a + // vector index. + var record map[string]interface{} + if err := s.internalCatalog(r.Context(), "GetTable", &s3tables.GetTableRequest{ + TableBucketARN: s.s3tablesArn(arns.bucket), + Namespace: []string{indexNamespace}, + Name: arns.index, + }, &record); err != nil { + catalogErr(w, err) + return + } + if format, _ := record["format"].(string); format != s3tables.FormatLance { + writeError(w, http.StatusBadRequest, "ValidationException", + fmt.Sprintf("index %q is not a vector index", arns.index)) + return + } // Deleting the catalog record is the delete: the engine derives an index's // existence from its table, and DropIndex is unimplemented on the shared // service.