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()); + } +}