vector: DeleteIndex format guard, sticky discovery, advertise normalize (#12171)

* 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.
This commit is contained in:
Chris Lu authored and GitHub committed 2026-10-28 09:22:05 -07:00
1 parent ae95b2c777
commit 511727cf3a
3 files changed
+61 -8

No files matched your search

+14 -1
View File
@@ -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);
}
+27 -7
View File
@@ -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
}
+20
View File
@@ -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.