vector: confine DropIndex to the table's own path, bound master dials, check the gRPC port (#12294)

* vector: refuse to drop a location that is not the table's own

DropIndex removed everything under whatever location the catalog
described. That location is caller-supplied through the Lance
namespace and only confined to the bucket, so a table registered at
s3://bkt/vector, or at another index's directory, wiped every index in
the bucket or that index. drop now accepts only the colocated path the
gateway's tableLocation derives from the table id, after trimming a
trailing slash, and the RPC answers FailedPrecondition otherwise. This
mirrors the ancestor and live-entry guards on the catalog's DeleteTable.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

* vector: bound the master dial and try a leader hint once

A remembered leader hint was never cleared and the dial had no connect
timeout, so a black-holed former leader added the kernel's SYN timeout
(about two minutes) to every reconnect pass before the configured
masters were tried, and registration lapsed. The dial now gives up
after 5s, the same bound wdclient puts on a master call, and a hint is
tried once ahead of the configured masters, as wdclient.tryAllMasters
keeps nextHintedLeader for one cycle only.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

* vector: refuse a master port whose gRPC port would pass 65535

grpc_addr derived the gRPC port as http + 10000 in u16, which panics in
debug builds and in release silently dials a wrapped port for HTTP
ports above 55535. Go's pb.ServerToGrpcAddress adds in int, so the port
never wraps and the dial fails instead. The addition is now checked,
and weed-vector refuses to start with an error asking for the explicit
host:http.grpc form.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

* vector: refuse a numeric master port past 65535

A master given as m1:65536 or m1:9333.65536 failed both u16 parses and
fell through to the raw address, so startup accepted it and every dial
failed. Ports now parse as u64, as Go's ServerToGrpcAddress does with
ParseUint, and one outside u16 is refused at startup.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

* vector: bound the master's first keep-connected answer

The connect timeout only covers TCP. A master that accepts the stream
and never answers held stream_to forever: grpc-go sends the response
headers with the first message, so the call never returned and later
seeds were never tried. A healthy master always answers at once (a
follower with its leader hint), so the stream open and the first
response now share the same 5s bound; after that the long-lived stream
stays untimed.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

---------

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

seaweed-vector/crates/server/src/main.rs

@@ -73,10 +73,7 @@ fn advertised_addr(listen: &str, masters: &[String]) -> String {
     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 = 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}"),
@@ -101,10 +98,12 @@ async fn main() -> Result<()> {
     let args = Args::parse_from(normalize_args(std::env::args().collect()));
     if let Some(masters) = &args.masters {
         let masters: Vec<String> = masters.split(',').map(|s| s.trim().to_string()).collect();
+        for master in &masters {
+            weed_vector_server::master_client::grpc_addr(master)?;
+        }
         let advertise = advertised_addr(&args.listen, &masters);
         tokio::spawn(weed_vector_server::master_client::keep_connected(
-            masters,
-            advertise,
+            masters, advertise,
         ));
     }
     let fallback: FallbackOptions = args.storage_options.into_iter().collect();

seaweed-vector/crates/server/src/master_client.rs

@@ -4,11 +4,11 @@
 
 use std::time::Duration;
 
-use anyhow::Result;
+use anyhow::{Context, Result};
 use tokio::time::sleep;
 use tonic::transport::Endpoint;
 
-use crate::master_pb::{seaweed_client::SeaweedClient, KeepConnectedRequest};
+use crate::master_pb::{KeepConnectedRequest, seaweed_client::SeaweedClient};
 
 fn registration(client_addr: &str) -> KeepConnectedRequest {
     KeepConnectedRequest {
@@ -27,19 +27,30 @@ fn registration(client_addr: &str) -> KeepConnectedRequest {
 /// weed conventions (server_address.ToGrpcAddress).
 // A master address may carry an explicit gRPC port as `host:http.grpc` —
 // use it directly; otherwise the gRPC port is http + 10000.
-fn grpc_addr(master: &str) -> String {
+pub fn grpc_addr(master: &str) -> Result<String> {
     let (scheme, rest) = master.split_once("://").unwrap_or(("http", master));
     if let Some((host, port)) = rest.rsplit_once(':') {
         if let Some((_, grpc)) = port.split_once('.') {
-            if grpc.parse::<u16>().is_ok() {
-                return format!("{scheme}://{host}:{grpc}");
+            if let Ok(grpc) = grpc.parse::<u64>() {
+                let grpc = u16::try_from(grpc)
+                    .with_context(|| format!("master {master}: gRPC port {grpc} is past 65535"))?;
+                return Ok(format!("{scheme}://{host}:{grpc}"));
             }
         }
-        if let Ok(port) = port.parse::<u16>() {
-            return format!("{scheme}://{host}:{}", port + 10000);
+        if let Ok(port) = port.parse::<u64>() {
+            let grpc = port
+                .checked_add(10000)
+                .and_then(|grpc| u16::try_from(grpc).ok())
+                .with_context(|| {
+                    format!(
+                        "master {master}: gRPC port {port}+10000 is past 65535; \
+                         give it explicitly as {host}:{port}.<grpc port>"
+                    )
+                })?;
+            return Ok(format!("{scheme}://{host}:{grpc}"));
         }
     }
-    format!("{scheme}://{rest}")
+    Ok(format!("{scheme}://{rest}"))
 }
 
 // `host:http.grpc` normalizes to `host:http` so a leader hint naming the
@@ -60,12 +71,7 @@ fn http_addr(master: &str) -> String {
 pub async fn keep_connected(masters: Vec<String>, client_addr: String) {
     let mut leader: Option<String> = None;
     loop {
-        let mut seeds: Vec<String> = Vec::new();
-        if let Some(l) = &leader {
-            seeds.push(l.clone());
-        }
-        seeds.extend(masters.iter().cloned());
-        for master in &seeds {
+        for master in &seeds(&mut leader, &masters) {
             match stream_to(master, &client_addr).await {
                 Ok(Some(l)) => {
                     leader = Some(l);
@@ -79,21 +85,46 @@ pub async fn keep_connected(masters: Vec<String>, client_addr: String) {
     }
 }
 
+/// A hint is tried once, as wdclient.tryAllMasters keeps it for one cycle, so
+/// a former leader that went away does not delay every later pass.
+fn seeds(leader: &mut Option<String>, masters: &[String]) -> Vec<String> {
+    leader
+        .take()
+        .into_iter()
+        .chain(masters.iter().cloned())
+        .collect()
+}
+
+/// Bounds the TCP connect and then the master's first answer, the 5s wdclient
+/// gives a master call; a black-holed or silent master otherwise holds up the
+/// other seeds.
+const CONNECT_TIMEOUT: Duration = Duration::from_secs(5);
+
 async fn stream_to(master: &str, client_addr: &str) -> Result<Option<String>> {
-    let channel = Endpoint::new(grpc_addr(master))?.connect().await?;
+    let channel = Endpoint::new(grpc_addr(master)?)?
+        .connect_timeout(CONNECT_TIMEOUT)
+        .connect()
+        .await?;
+    // A master answers at once, a follower with its leader hint.
+    let answer_by = tokio::time::Instant::now() + CONNECT_TIMEOUT;
     let mut client = SeaweedClient::new(channel);
     let (tx, rx) = tokio::sync::mpsc::channel(8);
     tx.send(registration(client_addr)).await?;
-    let mut stream = client
-        .keep_connected(tokio_stream::wrappers::ReceiverStream::new(rx))
-        .await?
-        .into_inner();
+    let mut stream = tokio::time::timeout_at(
+        answer_by,
+        client.keep_connected(tokio_stream::wrappers::ReceiverStream::new(rx)),
+    )
+    .await
+    .context("master did not answer keep-connected")??
+    .into_inner();
+    let mut answered = false;
     // The master keeps the node live while the stream is open; resend the
     // registration periodically so a reconnect never leaves a stale entry.
     loop {
         tokio::select! {
             msg = stream.message() => match msg? {
                 Some(resp) => {
+                    answered = true;
                     // A volume_location appears on every response — only
                     // follow it when the hint points at a different master,
                     // otherwise the leader keeps identifying itself and we
@@ -108,9 +139,63 @@ async fn stream_to(master: &str, client_addr: &str) -> Result<Option<String>> {
                 }
                 None => anyhow::bail!("master closed the keep-connected stream"),
             },
+            _ = tokio::time::sleep_until(answer_by), if !answered => {
+                anyhow::bail!("master did not answer keep-connected");
+            }
             _ = sleep(Duration::from_secs(5)) => {
                 tx.send(registration(client_addr)).await?;
             }
         }
     }
 }
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[test]
+    fn leader_hint_is_tried_once() {
+        let masters = vec!["m1:9333".to_string(), "m2:9333".to_string()];
+        let mut leader = Some("m3:9333".to_string());
+        assert_eq!(
+            seeds(&mut leader, &masters),
+            ["m3:9333", "m1:9333", "m2:9333"]
+        );
+        assert_eq!(seeds(&mut leader, &masters), ["m1:9333", "m2:9333"]);
+    }
+
+    #[tokio::test]
+    async fn dial_to_a_black_holed_master_is_bounded() {
+        // TEST-NET-1 (RFC 5737) is reserved, so nothing answers the SYN.
+        let dial = stream_to("192.0.2.1:9333", "127.0.0.1:8080");
+        let result = tokio::time::timeout(CONNECT_TIMEOUT * 3, dial).await;
+        assert!(matches!(result, Ok(Err(_))), "dial was not bounded");
+    }
+
+    #[tokio::test]
+    async fn a_master_that_never_answers_is_bounded() {
+        let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
+        let port = listener.local_addr().unwrap().port();
+        tokio::spawn(async move {
+            let (mut conn, _) = listener.accept().await.unwrap();
+            let _ = tokio::io::copy(&mut conn, &mut tokio::io::sink()).await;
+        });
+        let master = format!("127.0.0.1:9333.{port}");
+        let open = stream_to(&master, "127.0.0.1:8080");
+        match tokio::time::timeout(CONNECT_TIMEOUT * 3, open).await {
+            Ok(Err(e)) => assert!(format!("{e:#}").contains("did not answer"), "{e:#}"),
+            other => panic!("open was not bounded: {other:?}"),
+        }
+    }
+
+    #[test]
+    fn grpc_addr_refuses_a_port_past_65535() {
+        assert_eq!(grpc_addr("m1:9333").unwrap(), "http://m1:19333");
+        assert_eq!(grpc_addr("m1:55535").unwrap(), "http://m1:65535");
+        assert_eq!(grpc_addr("m1:60000.7000").unwrap(), "http://m1:7000");
+        assert!(grpc_addr("m1:55536").is_err());
+        assert!(grpc_addr("https://m1:60000").is_err());
+        assert!(grpc_addr("m1:65536").is_err());
+        assert!(grpc_addr("m1:9333.65536").is_err());
+    }
+}

seaweed-vector/crates/server/src/service.rs

@@ -333,7 +333,13 @@ impl VectorService for VectorServer {
             .ok_or_else(|| Status::invalid_argument("missing table"))?;
         dataset::drop(&self.client, &table.id, &self.fallback)
             .await
-            .map_err(internal)?;
+            .map_err(|err| {
+                if err.is::<dataset::DropRefused>() {
+                    Status::failed_precondition(format!("{err:#}"))
+                } else {
+                    internal(err)
+                }
+            })?;
         Ok(Response::new(pb::DropIndexResponse {}))
     }
 
@@ -644,4 +650,46 @@ mod tests {
         ];
         assert!(batch_of(&vectors, 4, schema).is_err());
     }
+
+    /// A catalog that answers one describe with `location`.
+    async fn catalog_describing(location: &str) -> String {
+        use tokio::io::{AsyncReadExt, AsyncWriteExt};
+        let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
+        let url = format!("http://{}", listener.local_addr().unwrap());
+        let body = serde_json::json!({ "location": location }).to_string();
+        tokio::spawn(async move {
+            let (mut conn, _) = listener.accept().await.unwrap();
+            let mut request = [0u8; 8192];
+            let _ = conn.read(&mut request).await;
+            let response = format!(
+                "HTTP/1.1 200 OK\r\ncontent-type: application/json\r\ncontent-length: {}\r\nconnection: close\r\n\r\n{body}",
+                body.len()
+            );
+            conn.write_all(response.as_bytes()).await.unwrap();
+            let _ = conn.shutdown().await;
+            let _ = tokio::io::copy(&mut conn, &mut tokio::io::sink()).await;
+        });
+        url
+    }
+
+    #[tokio::test]
+    async fn drop_index_refuses_a_location_outside_the_table() {
+        for location in ["s3://bkt/vector", "s3://bkt/vector/prod"] {
+            let server = VectorServer::new(
+                NamespaceClient::new(catalog_describing(location).await),
+                FallbackOptions::new(),
+            );
+            let request = pb::DropIndexRequest {
+                table: Some(pb::Table {
+                    id: vec!["bkt".into(), "vector".into(), "x".into()],
+                    version: 0,
+                }),
+            };
+            let err = server
+                .drop_index(Request::new(request))
+                .await
+                .expect_err(location);
+            assert_eq!(err.code(), tonic::Code::FailedPrecondition, "{location}");
+        }
+    }
 }

seaweed-worker/crates/lance/src/dataset.rs

@@ -78,6 +78,7 @@ pub async fn drop(
         Err(err) if describe_not_found(&err) => return Ok(()),
         Err(err) => return Err(err),
     };
+    check_drop_location(id, &description.location)?;
     let mut options: HashMap<String, String> = fallback.clone();
     options.extend(description.storage_options.clone());
     let (store, path) = ObjectStore::from_uri_and_params(
@@ -95,6 +96,46 @@ pub async fn drop(
     Ok(())
 }
 
+/// Where the catalog puts a table it was given no location for, as the
+/// gateway's tableLocation derives it.
+pub fn table_location(id: &[String]) -> Option<String> {
+    match id {
+        [bucket, ns @ .., name] if !ns.is_empty() => {
+            Some(format!("s3://{bucket}/{}/{name}", ns.join(".")))
+        }
+        _ => None,
+    }
+}
+
+/// A drop refused before anything was removed.
+#[derive(Debug)]
+pub struct DropRefused(String);
+
+impl std::fmt::Display for DropRefused {
+    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+        f.write_str(&self.0)
+    }
+}
+
+impl std::error::Error for DropRefused {}
+
+/// A location is caller-supplied and may name the bucket, the namespace, or
+/// another table, so only the table's own colocated directory is removed.
+pub fn check_drop_location(id: &[String], location: &str) -> Result<(), DropRefused> {
+    let Some(expected) = table_location(id) else {
+        return Err(DropRefused(format!(
+            "table id {id:?} needs a bucket, a namespace and a name"
+        )));
+    };
+    if location.trim_end_matches('/') != expected {
+        return Err(DropRefused(format!(
+            "refusing to drop table {}: location {location} is not its colocated path {expected}",
+            id.join("/")
+        )));
+    }
+    Ok(())
+}
+
 fn describe_not_found(err: &anyhow::Error) -> bool {
     err.chain().any(|cause| {
         cause
@@ -136,3 +177,53 @@ fn schema_json(dataset: &Dataset) -> Option<String> {
     )
     .ok()
 }
+
+#[cfg(test)]
+mod tests {
+    use super::{check_drop_location, table_location};
+
+    fn id(parts: &[&str]) -> Vec<String> {
+        parts.iter().map(|part| part.to_string()).collect()
+    }
+
+    #[test]
+    fn table_location_matches_the_gateway_layout() {
+        assert_eq!(
+            table_location(&id(&["bkt", "vector", "x"])).as_deref(),
+            Some("s3://bkt/vector/x")
+        );
+        assert_eq!(
+            table_location(&id(&["bkt", "a", "b", "t"])).as_deref(),
+            Some("s3://bkt/a.b/t")
+        );
+        assert_eq!(table_location(&id(&["bkt", "t"])), None);
+    }
+
+    #[test]
+    fn drop_location_accepts_the_colocated_path() {
+        let table = id(&["bkt", "vector", "x"]);
+        for location in ["s3://bkt/vector/x", "s3://bkt/vector/x/"] {
+            assert!(check_drop_location(&table, location).is_ok(), "{location}");
+        }
+    }
+
+    #[test]
+    fn drop_location_refuses_anything_else() {
+        let table = id(&["bkt", "vector", "x"]);
+        for location in [
+            // ancestors: the bucket and the namespace root
+            "s3://bkt",
+            "s3://bkt/",
+            "s3://bkt/vector",
+            "s3://bkt/vector/",
+            // a sibling index's directory
+            "s3://bkt/vector/prod",
+            "s3://bkt/vector/x/data",
+            "s3://other/vector/x",
+            "",
+        ] {
+            assert!(check_drop_location(&table, location).is_err(), "{location}");
+        }
+        assert!(check_drop_location(&id(&["bkt", "x"]), "s3://bkt/x").is_err());
+    }
+}