From 6cdddcd76d28f4835d93ffe0cdb1f591bb406d92 Mon Sep 17 00:00:00 2001 From: Chris Lu Date: Sat, 31 Oct 2026 18:42:18 -0700 Subject: [PATCH] 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 * 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 * 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 * 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 * 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 --------- Co-authored-by: Claude Opus 5.5 --- seaweed-vector/crates/server/src/main.rs | 11 +- .../crates/server/src/master_client.rs | 123 +++++++++++++++--- seaweed-vector/crates/server/src/service.rs | 50 ++++++- seaweed-worker/crates/lance/src/dataset.rs | 91 +++++++++++++ 4 files changed, 249 insertions(+), 26 deletions(-) diff --git a/seaweed-vector/crates/server/src/main.rs b/seaweed-vector/crates/server/src/main.rs index 15adc1047f..453aa8ed5f 100644 --- a/seaweed-vector/crates/server/src/main.rs +++ b/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 = 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(); diff --git a/seaweed-vector/crates/server/src/master_client.rs b/seaweed-vector/crates/server/src/master_client.rs index 614cdcd7f1..31f14948c3 100644 --- a/seaweed-vector/crates/server/src/master_client.rs +++ b/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 { 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::().is_ok() { - return format!("{scheme}://{host}:{grpc}"); + if let Ok(grpc) = grpc.parse::() { + 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::() { - return format!("{scheme}://{host}:{}", port + 10000); + if let Ok(port) = port.parse::() { + 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}." + ) + })?; + 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, client_addr: String) { let mut leader: Option = None; loop { - let mut seeds: Vec = 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, 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, masters: &[String]) -> Vec { + 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> { - 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> { } 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()); + } +} diff --git a/seaweed-vector/crates/server/src/service.rs b/seaweed-vector/crates/server/src/service.rs index 64819dad67..4287130b6e 100644 --- a/seaweed-vector/crates/server/src/service.rs +++ b/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::() { + 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}"); + } + } } diff --git a/seaweed-worker/crates/lance/src/dataset.rs b/seaweed-worker/crates/lance/src/dataset.rs index 4ae382c023..19eb1b82e4 100644 --- a/seaweed-worker/crates/lance/src/dataset.rs +++ b/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 = 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 { + 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 { ) .ok() } + +#[cfg(test)] +mod tests { + use super::{check_drop_location, table_location}; + + fn id(parts: &[&str]) -> Vec { + 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()); + } +}