vector: weed-vector registers to master; s3 discovers it there (#12146)

* vector: weed-vector registers with the weed master

weed-vector-server --masters holds a KeepConnected stream to the weed
master, the same mechanism filer and s3 use via wdclient — the server
shows up in ListClusterNodes(client_type=vector). Resends registration
and rediscovers on failure.

* vector: s3 discovers weed-vector through the master

With -vectors.grpc unset the vectors gateway resolves one registered
weed-vector via ListClusterNodes on the masters it already derives
from filer config — discovery replaces the static dial for the
single-worker OSS path. The conformance harness registers the engine
to mini's master and drops -vectors.grpc.

* vector: default-fill registration fields

..Default::default() keeps the source compiling against proto variants
that carry extra fields, so the same file works in either tree.

* vector: serve the Vectors API on the main s3 port

The vectors router sits on the main -port listener behind a dispatch
that checks the SigV4 credential scope for /s3vectors/ — no separate
port needed. -port.vectors stays for explicit dedicated-port
deployments; the conformance harness now hits vectors through :8334.

* vector: drop -port.vectors entirely

It was new — no compat to preserve. The main -port listener serves the
API whenever a weed-vector backend is reachable.

* vector: drop -vectors.grpc; discovery is the only locator

weed-vector always registers to master, the gateway always resolves it
through ListClusterNodes — a static address just forks membership.

* vector: binary is weed-vector

Matches the enterprise binary; package and lib names are unchanged.

* vector: review fixes — cluster type, grpc port, lazy resolve, leader hint, advertise

- cluster: register 'vector' as a node type so ListClusterNodes returns
  weed-vector members
- s3: dial masters via ToGrpcAddress (HTTP port + 10000)
- s3: backend resolves lazily through vectorDiscovery — workers joining
  after gateway startup work on first query, and address changes re-dial
- master_client: convert master HTTP addr to grpc +10000, honor http(s)
  scheme, follow the leader hint in KeepConnectedResponse, and
  --advertise covers wildcard/unreachable listen addresses

* vector: no --advertise flag; derive the outbound address

A wildcard listen resolves to the source IP the master sees via a
route lookup, so registration always advertises something reachable
without another flag.

* vector: review fixes — plaintext dispatch, conn leaks, discovery backoff

The two plaintext http servers now serve the vectors dispatch too —
signed s3vectors calls on the main port reach the right router.
discoverVectorWorker closes each master conn and bounds lookups to 5s;
a dead or absent worker no longer re-hits master on every RPC — the
resolver backs off for 5s between rediscovery attempts.

* vector: 1 GiB grpc cap, dead-worker invalidation, CLI rename

Valid PutVectors batches exceed the tonic 4 MiB default — match the
gateway's 1 GiB cap. A replaced worker no longer sticks: any RPC error
resets the cached address and the next call rediscovers through the
master. The core CLI bin renames to weed-vector-cli so the two crates
do not collide on the same output path.

* vector: retry-only-transport reset, periodic re-resolve, cli rename

- reset() takes the failed client and runs only on transport errors —
  an InvalidArgument on a healthy conn no longer interrupts unrelated
  requests, and a concurrent re-resolve wins the race.
- cli() re-resolves once the interval passes even with a live client,
  so a worker that moves addresses is picked up.
- verify.rs spike docs use weed-vector-cli after the bin rename.

* vector: only Unavailable counts as a transport failure

A request deadline or an application error can fire on a healthy
connection — resetting the shared conn on those churned reconnects.

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

@@ -2,6 +2,17 @@
 
 pub mod service;
 
+pub mod remote_pb {
+    tonic::include_proto!("remote_pb");
+}
+pub mod volume_server_pb {
+    tonic::include_proto!("volume_server_pb");
+}
+pub mod master_pb {
+    tonic::include_proto!("master_pb");
+}
+pub mod master_client;
+
 pub mod pb {
     pub mod vector {
         tonic::include_proto!("vector_pb");

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

@@ -23,6 +23,10 @@ struct Args {
     /// key=value, repeatable.
     #[arg(long = "storage-option", value_parser = parse_option)]
     storage_options: Vec<(String, String)>,
+    /// Comma-separated master addresses to register this server with; the
+    /// gateway discovers it via ListClusterNodes(client_type=vector).
+    #[arg(long, env = "WEED_VECTOR_MASTERS")]
+    masters: Option<String>,
 }
 
 fn parse_option(s: &str) -> Result<(String, String), String> {
@@ -31,16 +35,52 @@ fn parse_option(s: &str) -> Result<(String, String), String> {
         .ok_or_else(|| "expected key=value".to_string())
 }
 
+/// A wildcard listen address is not reachable for peers; derive the outbound
+/// IP toward the first master instead — the address the master sees anyway.
+fn advertised_addr(listen: &str, masters: &[String]) -> String {
+    let Some((host, _)) = listen.rsplit_once(':') else {
+        return listen.to_string();
+    };
+    if !matches!(host, "0.0.0.0" | "::" | "" | "[::]") {
+        return listen.to_string();
+    }
+    let port = listen.rsplit_once(':').map(|(_, p)| p).unwrap_or_default();
+    if let Some(master) = masters.first() {
+        if let Ok(sock) = std::net::UdpSocket::bind((host, 0u16)) {
+            if sock.connect(master).is_ok() {
+                if let Ok(local) = sock.local_addr() {
+                    return format!("{}:{}", local.ip(), port);
+                }
+            }
+        }
+    }
+    listen.to_string()
+}
+
 #[tokio::main]
 async fn main() -> Result<()> {
     tracing_subscriber::fmt::init();
     let args = Args::parse();
+    if let Some(masters) = &args.masters {
+        let masters: Vec<String> = masters.split(',').map(|s| s.trim().to_string()).collect();
+        let advertise = advertised_addr(&args.listen, &masters);
+        tokio::spawn(weed_vector_server::master_client::keep_connected(
+            masters,
+            advertise,
+        ));
+    }
     let fallback: FallbackOptions = args.storage_options.into_iter().collect();
     let client = NamespaceClient::new(args.namespace.clone()).with_fallback(&fallback);
     let service = VectorServer::new(client, fallback);
-    tracing::info!(namespace = %args.namespace, listen = %args.listen, "weed-vector-server");
+    tracing::info!(namespace = %args.namespace, listen = %args.listen, "weed-vector");
     Server::builder()
-        .add_service(VectorServiceServer::new(service))
+        // Match the gateway's 1 GiB cap — put batches over the tonic 4 MiB
+        // default otherwise get rejected before they reach the handler.
+        .add_service(
+            VectorServiceServer::new(service)
+                .max_decoding_message_size(1 << 30)
+                .max_encoding_message_size(1 << 30),
+        )
         .serve(args.listen.parse()?)
         .await?;
     Ok(())

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

@@ -0,0 +1,91 @@
+//! Self-registration with the weed master, mirroring how filer/s3 keep
+//! connected: hold a KeepConnected stream open to the leader and the node
+//! shows up in ListClusterNodes(client_type="vector").
+
+use std::time::Duration;
+
+use anyhow::Result;
+use tokio::time::sleep;
+use tonic::transport::Endpoint;
+
+use crate::master_pb::{seaweed_client::SeaweedClient, KeepConnectedRequest};
+
+fn registration(client_addr: &str) -> KeepConnectedRequest {
+    KeepConnectedRequest {
+        client_type: "vector".into(),
+        client_address: client_addr.into(),
+        version: env!("CARGO_PKG_VERSION").into(),
+        filer_group: String::new(),
+        data_center: String::new(),
+        rack: String::new(),
+        leave_lock_ring: false,
+        ..Default::default()
+    }
+}
+
+/// Masters are configured by their HTTP address; gRPC lives +10000 like the
+/// weed conventions (server_address.ToGrpcAddress).
+fn grpc_addr(master: &str) -> String {
+    let (scheme, rest) = master.split_once("://").unwrap_or(("http", master));
+    if let Some((host, port)) = rest.rsplit_once(':') {
+        if let Ok(port) = port.parse::<u16>() {
+            return format!("{scheme}://{host}:{}", port + 10000);
+        }
+    }
+    format!("{scheme}://{rest}")
+}
+
+/// Retry forever: try each master until a stream opens, hold it until it
+/// errors, then rediscover — same shape as wdclient.KeepConnectedToMaster.
+/// A follower response's leader hint takes priority on the next pass.
+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 {
+            match stream_to(master, &client_addr).await {
+                Ok(Some(l)) => {
+                    leader = Some(l);
+                    break;
+                }
+                Ok(None) => break,
+                Err(e) => tracing::warn!("master {master}: {e:#}"),
+            }
+        }
+        sleep(Duration::from_secs(3)).await;
+    }
+}
+
+async fn stream_to(master: &str, client_addr: &str) -> Result<Option<String>> {
+    let channel = Endpoint::new(grpc_addr(master))?.connect().await?;
+    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();
+    // 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) => {
+                    if let Some(loc) = resp.volume_location {
+                        if !loc.leader.is_empty() {
+                            return Ok(Some(loc.leader));
+                        }
+                    }
+                }
+                None => anyhow::bail!("master closed the keep-connected stream"),
+            },
+            _ = sleep(Duration::from_secs(5)) => {
+                tx.send(registration(client_addr)).await?;
+            }
+        }
+    }
+}