diff --git a/CHANGELOG.md b/CHANGELOG.md index 2a3819c0c..c10244087 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -10,6 +10,10 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ### Changed - A listener with SASL configured closes a connection that sends a frame larger than 512KiB before the client authenticates, matching the Apache Kafka default for `sasl.server.max.receive.size`. The same limit applies while a client re-authenticates. The broker logs this rejection as `PreAuthenticationFrameTooBig`, and counts it in `nisshi_frames_rejected`. +- The broker stops at startup when its storage URL has a query option that the storage engine does not read, such as a misspelt option or `vacuum_into` on a `postgres://` URL. The error names the option and the engine. Previously the broker ignored the option. +- The broker stops at startup when `maintenance_interval` or `transaction_maintenance_interval` has an invalid value: unparsable, zero, longer than 365 days, or a bare number without a unit. Previously the broker ignored an invalid value and used the default interval. Give a bare number its unit, for example `600s` or `10m` instead of `600`. Compound values such as `1h30m` and `5min` still work. +- A `parquet`, `iceberg` or `delta` broker without `--schema-registry` stops at startup with an error that names the missing option, instead of panicking. `--schema-registry` is accepted before or after the subcommand. +- When the broker closes the connection of a client that sends a request other than ApiVersions, SaslHandshake or SaslAuthenticate before it authenticates, it logs an ERROR line that names the client's address. ### Security diff --git a/Cargo.lock b/Cargo.lock index 62a875bc2..ecace69ac 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -5750,6 +5750,7 @@ dependencies = [ "futures", "glob", "http-body-util", + "human-units", "humantime", "hyper", "hyper-util", diff --git a/nisshi-broker/Cargo.toml b/nisshi-broker/Cargo.toml index 17a80d204..04358f1e0 100644 --- a/nisshi-broker/Cargo.toml +++ b/nisshi-broker/Cargo.toml @@ -29,6 +29,7 @@ console.workspace = true futures.workspace = true glob.workspace = true http-body-util.workspace = true +human-units.workspace = true humantime.workspace = true hyper-util.workspace = true hyper.workspace = true diff --git a/nisshi-broker/src/broker.rs b/nisshi-broker/src/broker.rs index a41f8a34f..f3dbba346 100644 --- a/nisshi-broker/src/broker.rs +++ b/nisshi-broker/src/broker.rs @@ -13,6 +13,7 @@ // limitations under the License. pub mod group; +mod maintenance; use crate::{ CancelKind, Error, Result, @@ -455,6 +456,18 @@ where // not an anomaly worth an `error!` on every occurrence. || io.kind() == ErrorKind::TimedOut => {} + // The broker closes the connection of a client that sends a + // request other than ApiVersions or SASL before it + // authenticates. This line names the peer at ERROR, the level + // of the default filter, because that filter disables the + // INFO `peer` span that carries the address. + Err(Error::KafkaProtocol(nisshi_sans_io::Error::NotAuthenticated)) => { + error!( + %addr, + "closed connection: client sent a request before it authenticated" + ); + } + Err(error) => { error!(?error); }, @@ -556,8 +569,6 @@ pub struct Builder { authentication: bool, tls_server_config: Option, silent: bool, - maintenance_interval: Option, - transaction_maintenance_interval: Option, cancellation: CancellationToken, } @@ -572,9 +583,6 @@ type PhantomBuilder = Builder< >; impl Builder { - const MAINTENANCE_INTERVAL: &str = "maintenance_interval"; - const TRANSACTION_MAINTENANCE_INTERVAL: &str = "transaction_maintenance_interval"; - pub fn node_id(self, node_id: i32) -> Builder { Builder { node_id, @@ -589,8 +597,6 @@ impl Builder { authentication: self.authentication, tls_server_config: self.tls_server_config, silent: self.silent, - maintenance_interval: self.maintenance_interval, - transaction_maintenance_interval: self.transaction_maintenance_interval, cancellation: self.cancellation, } } @@ -609,8 +615,6 @@ impl Builder { authentication: self.authentication, tls_server_config: self.tls_server_config, silent: self.silent, - maintenance_interval: self.maintenance_interval, - transaction_maintenance_interval: self.transaction_maintenance_interval, cancellation: self.cancellation, } @@ -630,8 +634,6 @@ impl Builder { authentication: self.authentication, tls_server_config: self.tls_server_config, silent: self.silent, - maintenance_interval: self.maintenance_interval, - transaction_maintenance_interval: self.transaction_maintenance_interval, cancellation: self.cancellation, } @@ -654,52 +656,13 @@ impl Builder { authentication: self.authentication, tls_server_config: self.tls_server_config, silent: self.silent, - maintenance_interval: self.maintenance_interval, - transaction_maintenance_interval: self.transaction_maintenance_interval, cancellation: self.cancellation, } } - pub fn storage(self, mut storage: Url) -> Builder { - let maintenance_interval = storage.query_pairs().find_map(|(k, v)| { - if k == Self::MAINTENANCE_INTERVAL { - v.parse::().map(Into::into).ok() - } else { - None - } - }); - - let transaction_maintenance_interval = storage.query_pairs().find_map(|(k, v)| { - if k == Self::TRANSACTION_MAINTENANCE_INTERVAL { - v.parse::().map(Into::into).ok() - } else { - None - } - }); - - let pairs = storage - .query_pairs() - .filter_map(|(k, v)| { - if k == Self::MAINTENANCE_INTERVAL || k == Self::TRANSACTION_MAINTENANCE_INTERVAL { - None - } else { - Some((k.to_string(), v.to_string())) - } - }) - .collect::>(); - - if pairs.is_empty() { - storage.set_query(None); - } else { - _ = storage.query_pairs_mut().clear().extend_pairs(pairs); - } - - debug!( - ?maintenance_interval, - ?transaction_maintenance_interval, - storage = %redact_url(&storage) - ); + pub fn storage(self, storage: Url) -> Builder { + debug!(storage = %redact_url(&storage)); Builder { node_id: self.node_id, @@ -714,8 +677,6 @@ impl Builder { authentication: self.authentication, tls_server_config: self.tls_server_config, silent: self.silent, - maintenance_interval, - transaction_maintenance_interval, cancellation: self.cancellation, } @@ -737,8 +698,6 @@ impl Builder { authentication: self.authentication, tls_server_config: self.tls_server_config, silent: self.silent, - maintenance_interval: self.maintenance_interval, - transaction_maintenance_interval: self.transaction_maintenance_interval, cancellation: self.cancellation, } @@ -793,6 +752,9 @@ impl Builder { .map(otel::metric_exporter) .transpose()?; + let (storage, intervals) = maintenance::take_intervals(self.storage.clone())?; + debug!(?intervals); + let builder = { let mut builder = StorageContainer::builder(); @@ -832,7 +794,7 @@ impl Builder { .advertised_listener(self.advertised_listener.clone()) .schema_registry(self.schema_registry.clone()) .lake_house(self.lake_house.clone()) - .storage(self.storage.clone()) + .storage(storage) .cancellation(self.cancellation.clone()) .silent(self.silent) .build() @@ -859,8 +821,8 @@ impl Builder { tls_server_config: self.tls_server_config.map(Arc::new), silent: self.silent, - maintenance_interval: self.maintenance_interval, - transaction_maintenance_interval: self.transaction_maintenance_interval, + maintenance_interval: intervals.maintenance, + transaction_maintenance_interval: intervals.transaction_maintenance, cancellation: self.cancellation, meter_provider, }) diff --git a/nisshi-broker/src/broker/maintenance.rs b/nisshi-broker/src/broker/maintenance.rs new file mode 100644 index 000000000..ba99b4100 --- /dev/null +++ b/nisshi-broker/src/broker/maintenance.rs @@ -0,0 +1,199 @@ +// Copyright ⓒ 2024-2026 Peter Morgan +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +//! The broker's sweep interval options, read from the storage URL's query string. + +use crate::{Error, Result}; +use std::{str::FromStr, time::Duration}; +use url::Url; + +pub(crate) const MAINTENANCE_INTERVAL: &str = "maintenance_interval"; +pub(crate) const TRANSACTION_MAINTENANCE_INTERVAL: &str = "transaction_maintenance_interval"; + +/// The longest sweep interval that the broker accepts. +/// +/// A late tick makes `tokio::time::Interval` add the period to `Instant::now()` +/// without an overflow check, so a very large period panics the broker. One year is +/// far above any useful sweep interval and far below that overflow. +const MAXIMUM_INTERVAL: Duration = Duration::from_secs(365 * 24 * 60 * 60); + +/// The sweep intervals that the broker reads from a storage URL. +#[derive(Clone, Debug, Default, PartialEq, Eq)] +pub(crate) struct Intervals { + pub(crate) maintenance: Option, + pub(crate) transaction_maintenance: Option, +} + +/// Removes the sweep interval options from `storage`, and returns them parsed. +/// +/// The broker owns these options, and no storage engine reads them, so the URL that +/// this returns is the one that the broker gives to the storage engine. +/// +/// # Errors +/// +/// [`Error::InvalidStorageOptionValue`] when an interval option is present and +/// [`parse_interval`] rejects its value. +pub(crate) fn take_intervals(mut storage: Url) -> Result<(Url, Intervals)> { + let mut intervals = Intervals::default(); + let mut retained = Vec::new(); + + for (key, value) in storage.query_pairs() { + match key.as_ref() { + MAINTENANCE_INTERVAL => { + intervals.maintenance = Some(parse_interval(&key, &value)?); + } + TRANSACTION_MAINTENANCE_INTERVAL => { + intervals.transaction_maintenance = Some(parse_interval(&key, &value)?); + } + _ => retained.push((key.into_owned(), value.into_owned())), + } + } + + if intervals == Intervals::default() { + return Ok((storage, intervals)); + } + + if retained.is_empty() { + storage.set_query(None); + } else { + _ = storage.query_pairs_mut().clear().extend_pairs(retained); + } + + Ok((storage, intervals)) +} + +/// Parses the value of the sweep interval option `option`. +/// +/// The value takes a `human_units` duration (`90s`, `10m`), or a `humantime` duration +/// (`1h30m`, `5min`). A value must name its unit: a bare number is rejected instead +/// of read as seconds, because a number of milliseconds then becomes a period of days. +/// +/// # Errors +/// +/// [`Error::InvalidStorageOptionValue`] when the value is a bare number, does not +/// parse, is zero (`tokio::time::interval` panics on a zero period), or is longer +/// than one year. +pub(crate) fn parse_interval(option: &str, value: &str) -> Result { + let invalid = || Error::InvalidStorageOptionValue { + option: option.to_owned(), + value: value.to_owned(), + }; + + if value.trim().chars().all(|c| c.is_ascii_digit()) { + return Err(invalid()); + } + + human_units::Duration::from_str(value) + .map(|duration| duration.0) + .or_else(|_| value.parse::().map(Into::into)) + .ok() + .filter(|duration| !duration.is_zero() && *duration <= MAXIMUM_INTERVAL) + .ok_or_else(invalid) +} + +#[cfg(test)] +mod tests { + use super::*; + + fn invalid(error: Error, expected_option: &str, expected_value: &str) -> bool { + matches!( + error, + Error::InvalidStorageOptionValue { option, value } + if option == expected_option && value == expected_value + ) + } + + #[test] + fn parses_human_units_and_humantime_durations() -> Result<()> { + for (value, expected) in [ + ("90s", Duration::from_secs(90)), + ("10m", Duration::from_mins(10)), + ("500ms", Duration::from_millis(500)), + ("1h30m", Duration::from_mins(90)), + ("5min", Duration::from_mins(5)), + ("365d", MAXIMUM_INTERVAL), + ] { + assert_eq!( + expected, + parse_interval(MAINTENANCE_INTERVAL, value)?, + "{value}" + ); + } + + Ok(()) + } + + #[test] + fn rejects_bare_zero_unparsable_and_too_long_values() { + for value in [ + "", + "600000", + "0s", + "0m", + "ten minutes", + "10x", + "366d", + "10000000000000000000s", + ] { + let error = parse_interval(MAINTENANCE_INTERVAL, value).unwrap_err(); + assert!(invalid(error, MAINTENANCE_INTERVAL, value), "{value}"); + } + } + + #[test] + fn take_intervals_without_options_leaves_url_unchanged() -> Result<()> { + let storage = Url::parse("sqlite://nisshi.db?vacuum_into=/tmp/x")?; + + let (retained, intervals) = take_intervals(storage.clone())?; + + assert_eq!(storage, retained); + assert_eq!(Intervals::default(), intervals); + + Ok(()) + } + + #[test] + fn take_intervals_removes_interval_options_and_keeps_the_rest() -> Result<()> { + let (retained, intervals) = take_intervals(Url::parse( + "sqlite://nisshi.db?maintenance_interval=1m&vacuum_into=/tmp/x&transaction_maintenance_interval=5s", + )?)?; + + assert_eq!( + "sqlite://nisshi.db?vacuum_into=%2Ftmp%2Fx", + retained.as_str() + ); + assert_eq!( + Intervals { + maintenance: Some(Duration::from_mins(1)), + transaction_maintenance: Some(Duration::from_secs(5)), + }, + intervals + ); + + let (retained, _) = take_intervals(Url::parse("memory://?maintenance_interval=1m")?)?; + assert_eq!(None, retained.query()); + + Ok(()) + } + + #[test] + fn take_intervals_rejects_an_invalid_interval() -> Result<()> { + let error = take_intervals(Url::parse("memory://?transaction_maintenance_interval=0s")?) + .unwrap_err(); + + assert!(invalid(error, TRANSACTION_MAINTENANCE_INTERVAL, "0s")); + + Ok(()) + } +} diff --git a/nisshi-broker/src/lib.rs b/nisshi-broker/src/lib.rs index 7e7cdf1dd..b7de3b530 100644 --- a/nisshi-broker/src/lib.rs +++ b/nisshi-broker/src/lib.rs @@ -34,7 +34,6 @@ use regex::{Regex, Replacer}; use thiserror::Error; use tokio::{sync::broadcast::error::SendError, task::JoinError}; use tracing_subscriber::filter::ParseError; -use url::Url; pub mod broker; pub mod coordinator; @@ -79,6 +78,14 @@ pub enum Error { ExpectedJoinGroupRequestProtocol(&'static str), Hyper(Arc), + + /// A sweep interval option in the storage URL has a value that the broker + /// rejects. + InvalidStorageOptionValue { + option: String, + value: String, + }, + Io(Arc), Join(Arc), Json(Arc), @@ -105,7 +112,6 @@ pub enum Error { TryFromInt(#[from] TryFromIntError), UnsupportedApiService(i16), - UnsupportedStorageUrl(Url), UnsupportedTracingFormat(String), Url(#[from] url::ParseError), Utf8(#[from] Utf8Error), diff --git a/nisshi-broker/tests/it/main.rs b/nisshi-broker/tests/it/main.rs index 5eb774d18..8631cc348 100644 --- a/nisshi-broker/tests/it/main.rs +++ b/nisshi-broker/tests/it/main.rs @@ -55,10 +55,12 @@ pub mod storage_describe_configs; pub mod storage_fetch; pub mod storage_list_offsets; pub mod storage_metadata; +pub mod storage_options; pub mod tls; pub mod topic; pub mod topic_lifecycle; pub mod txn; +pub mod unauthenticated; pub mod update_group_conditional; // Cargo only builds the modules declared above, so a file in `tests/it/` without a diff --git a/nisshi-broker/tests/it/storage_options.rs b/nisshi-broker/tests/it/storage_options.rs new file mode 100644 index 000000000..c598fbb75 --- /dev/null +++ b/nisshi-broker/tests/it/storage_options.rs @@ -0,0 +1,143 @@ +// Copyright ⓒ 2024-2026 Peter Morgan +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +//! Storage URL query options at broker startup: the broker accepts its sweep +//! interval options on every engine, and rejects an invalid interval or an option +//! that the engine does not read. + +use nisshi_broker::{ + Error, NODE_ID, Result, broker::Broker, coordinator::group::administrator::Controller, +}; +use nisshi_storage::ArcDynStorage; +use url::Url; +use uuid::Uuid; + +use crate::common::init_tracing; + +async fn build(storage: &str) -> Result<()> { + let listener = Url::parse("tcp://127.0.0.1:9092")?; + + Broker::, ArcDynStorage>::builder() + .node_id(NODE_ID) + .cluster_id(format!("storage-options-{}", Uuid::now_v7())) + .incarnation_id(Uuid::now_v7()) + .advertised_listener(listener.clone()) + .storage(Url::parse(storage)?) + .listener(listener) + .silent(true) + .build() + .await + .map(|_| ()) +} + +fn is_invalid_interval(error: &Error, expected_option: &str, expected_value: &str) -> bool { + matches!( + error, + Error::InvalidStorageOptionValue { option, value } + if option == expected_option && value == expected_value + ) +} + +fn is_unrecognized(error: &Error, expected_scheme: &str, expected_option: &str) -> bool { + matches!( + error, + Error::Storage(nisshi_storage::Error::UnrecognizedStorageOption { scheme, option }) + if scheme == expected_scheme && option == expected_option + ) +} + +#[tokio::test] +async fn memory_accepts_interval_options() -> Result<()> { + let _guard = init_tracing()?; + + build("memory://?maintenance_interval=1m&transaction_maintenance_interval=5s").await +} + +#[tokio::test] +async fn memory_rejects_zero_interval() -> Result<()> { + let _guard = init_tracing()?; + + let error = build("memory://?maintenance_interval=0s") + .await + .unwrap_err(); + assert!( + is_invalid_interval(&error, "maintenance_interval", "0s"), + "{error:?}" + ); + + Ok(()) +} + +#[tokio::test] +async fn memory_rejects_misspelt_option() -> Result<()> { + let _guard = init_tracing()?; + + let error = build("memory://?maintenance_intervl=1m").await.unwrap_err(); + assert!( + is_unrecognized(&error, "memory", "maintenance_intervl"), + "{error:?}" + ); + + Ok(()) +} + +#[cfg(feature = "libsql")] +mod sqlite { + use super::*; + + fn url(query: &str) -> String { + format!( + "sqlite://../logs/{}/storage-options-{}.db?{query}", + env!("CARGO_PKG_NAME"), + Uuid::now_v7() + ) + } + + #[tokio::test] + async fn accepts_interval_and_engine_options() -> Result<()> { + let _guard = init_tracing()?; + + build(&url(concat!( + "maintenance_interval=1m&busy_timeout=5s", + "&vacuum_into=../logs/nisshi-broker/storage-options.vacuum.db" + ))) + .await + } + + #[tokio::test] + async fn rejects_zero_interval() -> Result<()> { + let _guard = init_tracing()?; + + let error = build(&url("maintenance_interval=0s")).await.unwrap_err(); + assert!( + is_invalid_interval(&error, "maintenance_interval", "0s"), + "{error:?}" + ); + + Ok(()) + } + + #[tokio::test] + async fn rejects_misspelt_option() -> Result<()> { + let _guard = init_tracing()?; + + let error = build(&url("vacuum_onto=/tmp/x")).await.unwrap_err(); + assert!( + is_unrecognized(&error, "sqlite", "vacuum_onto"), + "{error:?}" + ); + + Ok(()) + } +} diff --git a/nisshi-broker/tests/it/unauthenticated.rs b/nisshi-broker/tests/it/unauthenticated.rs new file mode 100644 index 000000000..c18077eb6 --- /dev/null +++ b/nisshi-broker/tests/it/unauthenticated.rs @@ -0,0 +1,235 @@ +// Copyright ⓒ 2024-2026 Peter Morgan +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +//! Transport-level tests: on a broker with `--authentication`, +//! a client that sends a request before completing SASL must have its +//! connection closed promptly, the same way `nisshi-broker/tests/it/auth.rs` +//! proves the error type without ever opening a socket. These tests open a +//! real socket, so they can show the connection actually closes. + +use std::{net::SocketAddr, time::Duration}; + +use anyhow::{Context as _, Result, anyhow}; +use bytes::Bytes; +use nisshi_broker::{NODE_ID, broker::Broker, coordinator::group::administrator::Controller}; +use nisshi_sans_io::{ + ApiKey as _, ApiVersionsRequest, ApiVersionsResponse, Body, Frame, Header, MetadataRequest, + MetadataResponse, +}; +use nisshi_storage::ArcDynStorage; +use tokio::{ + io::{AsyncRead, AsyncReadExt, AsyncWrite, AsyncWriteExt}, + net::{TcpListener, TcpStream}, + task::JoinHandle, + time::{Instant, sleep, timeout}, +}; +use tracing::debug; +use url::Url; +use uuid::Uuid; + +use crate::common::init_tracing; + +const TEST_TIMEOUT: Duration = Duration::from_secs(30); + +/// Largest response the test client will read; anything bigger is not a +/// Kafka frame. +const MAXIMUM_RESPONSE: usize = 16 * 1024 * 1024; + +struct RunningBroker { + handle: JoinHandle<()>, + addr: SocketAddr, +} + +impl Drop for RunningBroker { + fn drop(&mut self) { + self.handle.abort(); + } +} + +async fn free_port() -> Result { + let listener = TcpListener::bind("127.0.0.1:0").await?; + let port = listener.local_addr()?.port(); + drop(listener); + Ok(port) +} + +/// A plaintext, memory-backed broker with `--authentication` enabled and no +/// credentials configured: every client is unauthenticated. +async fn spawn_authenticating_broker() -> Result { + let port = free_port().await?; + let addr = SocketAddr::from(([127, 0, 0, 1], port)); + let listener = Url::parse(&format!("tcp://{addr}"))?; + + let mut broker = Broker::, ArcDynStorage>::builder() + .node_id(NODE_ID) + .cluster_id(format!("unauthenticated-{}", Uuid::now_v7())) + .incarnation_id(Uuid::now_v7()) + .advertised_listener(listener.clone()) + .storage(Url::parse("memory://")?) + .listener(listener) + .authentication(true) + .silent(true) + .build() + .await?; + + let handle = tokio::spawn(async move { + if let Err(err) = broker.serve(Instant::now()).await { + debug!(?err); + } + }); + + let deadline = Instant::now() + Duration::from_secs(10); + + loop { + match TcpStream::connect(addr).await { + Ok(_) => break, + Err(_) if Instant::now() < deadline => sleep(Duration::from_millis(50)).await, + Err(err) => return Err(anyhow!("broker did not start listening on {addr}: {err}")), + } + } + + Ok(RunningBroker { handle, addr }) +} + +/// Send one Kafka request frame and read back the response frame. +async fn round_trip( + stream: &mut S, + api_key: i16, + api_version: i16, + correlation_id: i32, + body: Body, +) -> Result +where + S: AsyncRead + AsyncWrite + Unpin, +{ + let request = Frame::request( + Header::Request { + api_key, + api_version, + correlation_id, + client_id: Some(env!("CARGO_PKG_NAME").into()), + }, + body, + )?; + + stream.write_all(&request).await?; + stream.flush().await?; + + let mut size = [0u8; 4]; + _ = stream.read_exact(&mut size).await?; + + let length = usize::try_from(i32::from_be_bytes(size)) + .ok() + .filter(|length| *length <= MAXIMUM_RESPONSE) + .ok_or_else(|| anyhow!("not a kafka frame length prefix: {size:02x?}"))?; + + let mut buffer = vec![0u8; length + size.len()]; + buffer[..size.len()].copy_from_slice(&size); + _ = stream.read_exact(&mut buffer[size.len()..]).await?; + + Frame::response_from_bytes(Bytes::from(buffer), api_key, api_version).map_err(Into::into) +} + +async fn api_versions(stream: &mut S) -> Result +where + S: AsyncRead + AsyncWrite + Unpin, +{ + let frame = round_trip( + stream, + ApiVersionsRequest::KEY, + 3, + 1, + ApiVersionsRequest::default() + .client_software_name(Some(env!("CARGO_PKG_NAME").into())) + .client_software_version(Some(env!("CARGO_PKG_VERSION").into())) + .into(), + ) + .await?; + + ApiVersionsResponse::try_from(frame.body).map_err(Into::into) +} + +async fn metadata(stream: &mut S) -> Result +where + S: AsyncRead + AsyncWrite + Unpin, +{ + let frame = round_trip( + stream, + MetadataRequest::KEY, + 12, + 2, + MetadataRequest::default() + .topics(Some([].into())) + .allow_auto_topic_creation(Some(false)) + .include_cluster_authorized_operations(Some(false)) + .include_topic_authorized_operations(Some(false)) + .into(), + ) + .await?; + + MetadataResponse::try_from(frame.body).map_err(Into::into) +} + +/// `ApiVersions` is exempt from authentication (a client always needs it to +/// negotiate versions before it can even attempt SASL), so it must still +/// succeed on an otherwise-unauthenticated connection. +#[tokio::test] +async fn api_versions_exempt_before_authentication() -> Result<()> { + let _guard = init_tracing()?; + + timeout(TEST_TIMEOUT, async { + let broker = spawn_authenticating_broker().await?; + + let mut stream = TcpStream::connect(broker.addr).await?; + let response = api_versions(&mut stream).await?; + + assert!( + !response.api_keys.unwrap_or_default().is_empty(), + "ApiVersions must succeed before authentication" + ); + + Ok(()) + }) + .await? +} + +/// A request other than `ApiVersions`/`SaslHandshake`/`SaslAuthenticate`, +/// sent before any SASL exchange, must close the connection promptly rather +/// than hang. +#[tokio::test] +async fn request_before_authentication_closes_connection() -> Result<()> { + let _guard = init_tracing()?; + + timeout(TEST_TIMEOUT, async { + let broker = spawn_authenticating_broker().await?; + + let mut stream = TcpStream::connect(broker.addr).await?; + let outcome = metadata(&mut stream).await; + + assert!( + outcome.is_err(), + "a request before authentication must not get a response, got {outcome:?}" + ); + + // The rejected connection must not have taken the listener with it. + let mut next = TcpStream::connect(broker.addr).await?; + let response = api_versions(&mut next) + .await + .context("a later connection must still be served")?; + assert!(!response.api_keys.unwrap_or_default().is_empty()); + + Ok(()) + }) + .await? +} diff --git a/nisshi-cli/src/cli/broker.rs b/nisshi-cli/src/cli/broker.rs index 6ccad9a5a..172950cd2 100644 --- a/nisshi-cli/src/cli/broker.rs +++ b/nisshi-cli/src/cli/broker.rs @@ -82,7 +82,7 @@ pub(super) struct Arg { storage_engine: EnvVarExp, /// Schema registry examples are: file://./etc/schema or s3://nisshi/, containing: topic.json, topic.proto or topic.avsc - #[arg(long, env = "SCHEMA_REGISTRY")] + #[arg(long, env = "SCHEMA_REGISTRY", global = true)] schema_registry: Option>, /// Schema registry cache expiry duration @@ -332,6 +332,40 @@ pub(super) enum Lake { }, } +#[cfg(any(feature = "parquet", feature = "iceberg", feature = "delta"))] +impl Lake { + fn name(&self) -> &'static str { + match self { + #[cfg(feature = "iceberg")] + Self::Iceberg { .. } => "iceberg", + + #[cfg(feature = "delta")] + Self::Delta { .. } => "delta", + + #[cfg(feature = "parquet")] + Self::Parquet { .. } => "parquet", + } + } +} + +/// Requires a schema registry whenever a data lake subcommand is selected. +/// +/// clap cannot enforce this itself without making `--schema-registry` +/// required on each lake subcommand individually, which rejects the flag +/// given before the subcommand (`broker --schema-registry y iceberg ...`): +/// the only ordering this repo's own `justfile`, `compose.yaml` and +/// `docs/schema-registry.md` use. This check runs instead, after clap has +/// already resolved the value regardless of where the flag appeared. +#[cfg(any(feature = "parquet", feature = "iceberg", feature = "delta"))] +fn require_schema_registry(command: &Option, has_schema_registry: bool) -> Result<()> { + match command { + Some(lake) if !has_schema_registry => { + Err(Error::LakeRequiresSchemaRegistry { lake: lake.name() }) + } + _ => Ok(()), + } +} + impl Arg { pub(super) async fn main(self) -> Result { let started = Instant::now(); @@ -387,48 +421,58 @@ impl Arg { .transpose()?; #[cfg(any(feature = "parquet", feature = "iceberg", feature = "delta"))] - let lake_house = match self.command { - #[cfg(feature = "iceberg")] - Some(Lake::Iceberg { - location, - catalog, - namespace, - warehouse, - }) => Some( - nisshi_schema::lake::House::iceberg() - .location(location.into_inner()) - .catalog(catalog.into_inner()) - .schema_registry(schema_registry.clone().unwrap()) - .namespace(namespace) - .warehouse(warehouse) - .build() - .await?, - ), - - #[cfg(feature = "delta")] - Some(Lake::Delta { - location, - database, - records_per_second, - }) => Some( - nisshi_schema::lake::House::delta() - .location(location.into_inner()) - .schema_registry(schema_registry.clone().unwrap()) - .database(database) - .records_per_second(records_per_second) - .build()?, - ), - - #[cfg(feature = "parquet")] - Some(Lake::Parquet { location }) => Some( - nisshi_schema::lake::House::parquet() - .location(location.into_inner()) - .schema_registry(schema_registry.clone().unwrap()) - .build()?, - ), + require_schema_registry(&self.command, schema_registry.is_some())?; - None => None, - }; + #[cfg(any(feature = "parquet", feature = "iceberg", feature = "delta"))] + let lake_house = + match self.command { + #[cfg(feature = "iceberg")] + Some(Lake::Iceberg { + location, + catalog, + namespace, + warehouse, + }) => Some( + nisshi_schema::lake::House::iceberg() + .location(location.into_inner()) + .catalog(catalog.into_inner()) + .schema_registry(schema_registry.clone().expect( + "require_schema_registry already checked a lake command has one", + )) + .namespace(namespace) + .warehouse(warehouse) + .build() + .await?, + ), + + #[cfg(feature = "delta")] + Some(Lake::Delta { + location, + database, + records_per_second, + }) => Some( + nisshi_schema::lake::House::delta() + .location(location.into_inner()) + .schema_registry(schema_registry.clone().expect( + "require_schema_registry already checked a lake command has one", + )) + .database(database) + .records_per_second(records_per_second) + .build()?, + ), + + #[cfg(feature = "parquet")] + Some(Lake::Parquet { location }) => Some( + nisshi_schema::lake::House::parquet() + .location(location.into_inner()) + .schema_registry(schema_registry.clone().expect( + "require_schema_registry already checked a lake command has one", + )) + .build()?, + ), + + None => None, + }; let broker = Broker::, ArcDynStorage>::builder() .node_id(NODE_ID) @@ -1048,4 +1092,127 @@ mod tests { .await .expect("build with an encrypted cert and key bundle"); } + + #[cfg(any(feature = "parquet", feature = "iceberg", feature = "delta"))] + mod lake_schema_registry { + use super::*; + + /// Whichever lake feature this build enables, so the test runs + /// regardless of which combination `--all-features` turns on. + fn lake() -> Lake { + #[cfg(feature = "parquet")] + { + Lake::Parquet { + location: "file://./lake".parse().expect("valid url"), + } + } + + #[cfg(all(not(feature = "parquet"), feature = "delta"))] + { + Lake::Delta { + location: "file://./lake".parse().expect("valid url"), + database: None, + records_per_second: None, + } + } + + #[cfg(all(not(feature = "parquet"), not(feature = "delta"), feature = "iceberg"))] + { + Lake::Iceberg { + location: "file://./lake".parse().expect("valid url"), + catalog: "http://localhost:8181/".parse().expect("valid url"), + namespace: None, + warehouse: None, + } + } + } + + #[test] + fn no_lake_no_registry_is_ok() { + require_schema_registry(&None, false).expect("no lake command needs no registry"); + } + + #[test] + fn no_lake_with_registry_is_ok() { + require_schema_registry(&None, true) + .expect("a registry present without a lake command is fine"); + } + + #[test] + fn lake_without_registry_is_rejected() { + let err = require_schema_registry(&Some(lake()), false) + .expect_err("a lake command without a schema registry must be rejected"); + + assert!( + matches!(err, Error::LakeRequiresSchemaRegistry { .. }), + "{err:?}" + ); + } + + #[test] + fn lake_with_registry_is_ok() { + require_schema_registry(&Some(lake()), true) + .expect("a lake command with a schema registry must be accepted"); + } + } + + #[cfg(feature = "parquet")] + #[tokio::test] + async fn parquet_lake_without_schema_registry_fails_build() { + let arg = parse(&[ + "--storage-engine", + "memory://nisshi/", + "--silent", + "parquet", + "--location", + "file://./lake", + ]) + .expect("arguments parse"); + + // Never depend on a SCHEMA_REGISTRY the environment happens to export. + let arg = Arg { + schema_registry: None, + ..arg + }; + + let err = arg + .build() + .await + .expect_err("a lake command without a schema registry must fail build"); + + assert!( + matches!(err, Error::LakeRequiresSchemaRegistry { lake: "parquet" }), + "{err:?}" + ); + } + + #[cfg(feature = "parquet")] + #[test] + fn schema_registry_before_subcommand_parses() { + let arg = parse(&[ + "--schema-registry", + "file://./etc/schema", + "parquet", + "--location", + "file://./lake", + ]) + .expect("--schema-registry given before the subcommand must still parse"); + + assert!(arg.schema_registry.is_some()); + } + + #[cfg(feature = "parquet")] + #[test] + fn schema_registry_after_subcommand_parses() { + let arg = parse(&[ + "parquet", + "--location", + "file://./lake", + "--schema-registry", + "file://./etc/schema", + ]) + .expect("--schema-registry given after the subcommand must parse"); + + assert!(arg.schema_registry.is_some()); + } } diff --git a/nisshi-cli/src/lib.rs b/nisshi-cli/src/lib.rs index 06ce36801..7554bc094 100644 --- a/nisshi-cli/src/lib.rs +++ b/nisshi-cli/src/lib.rs @@ -27,6 +27,11 @@ pub enum Error { DotEnv(#[from] dotenv::Error), Generate(#[from] nisshi_generator::Error), InvalidLength(#[from] sha2::digest::InvalidLength), + /// A data lake subcommand (`iceberg`, `delta`, `parquet`) was given + /// without `--schema-registry`, which every lake writer requires. + LakeRequiresSchemaRegistry { + lake: &'static str, + }, Perf(#[from] nisshi_perf::Error), Proxy(#[from] nisshi_proxy::Error), Regex(#[from] regex::Error), diff --git a/nisshi-service/src/frame.rs b/nisshi-service/src/frame.rs index b9ea61de2..3aea9b0cc 100644 --- a/nisshi-service/src/frame.rs +++ b/nisshi-service/src/frame.rs @@ -29,7 +29,7 @@ use opentelemetry::KeyValue; use rama::{Layer, Service, extensions::Extensions, matcher::Matcher, service::BoxService}; use rsasl::config::SASLConfig; use tokio::task::spawn_blocking; -use tracing::{debug, error, instrument, warn}; +use tracing::{debug, error, instrument}; use crate::{API_ERRORS, API_REQUESTS, ProgressBarExtension}; @@ -323,7 +323,7 @@ where .map(|api_key| i16::from_be_bytes([api_key[0], api_key[1]])) && !self.is_authenticated(peeked_api_key) { - warn!(api_key = peeked_api_key, "request before authentication"); + debug!(api_key = peeked_api_key, "request before authentication"); return Err(Into::into(nisshi_sans_io::Error::NotAuthenticated)); } @@ -335,6 +335,12 @@ where let api_key = req.api_key()?; if !self.is_authenticated(api_key) { + debug!( + api_key, + api_name = %req.api_name(), + "request before authentication" + ); + return Err(Into::into(nisshi_sans_io::Error::NotAuthenticated)); } diff --git a/nisshi-storage-dynostore/src/factory.rs b/nisshi-storage-dynostore/src/factory.rs index 25cc080c1..09fe5b756 100644 --- a/nisshi-storage-dynostore/src/factory.rs +++ b/nisshi-storage-dynostore/src/factory.rs @@ -18,6 +18,7 @@ use async_trait::async_trait; use nisshi_schema::redact_url; use nisshi_storage::{ ArcDynStorage, ProduceRequestBatcher, Result, StorageFactory, StorageFactoryConfiguration, + reject_unrecognized_options, }; use object_store::{ aws::{AmazonS3Builder, S3ConditionalPut}, @@ -39,6 +40,8 @@ impl StorageFactory for MemoryEngineFactory { } async fn build(&self, configuration: StorageFactoryConfiguration) -> Result { + reject_unrecognized_options(&configuration.storage, &[])?; + Ok(Arc::new(Box::new( DynoStore::new( configuration.cluster.as_str(), @@ -62,6 +65,11 @@ impl StorageFactory for S3OptimisticConcurrencyEngineFactory { } async fn build(&self, configuration: StorageFactoryConfiguration) -> Result { + reject_unrecognized_options( + &configuration.storage, + &["batch_min_size", "batch_max_delay"], + )?; + let bucket_name = configuration.storage.host_str().unwrap_or("nisshi"); let minimum_size = configuration.storage.query_pairs().find_map(|(k, v)| { @@ -128,6 +136,11 @@ impl StorageFactory for GoogleCloudStorageEngineFactory { } async fn build(&self, configuration: StorageFactoryConfiguration) -> Result { + reject_unrecognized_options( + &configuration.storage, + &["batch_min_size", "batch_max_delay"], + )?; + let bucket_name = configuration.storage.host_str().unwrap_or("nisshi"); let minimum_size = configuration.storage.query_pairs().find_map(|(k, v)| { diff --git a/nisshi-storage-null/src/factory.rs b/nisshi-storage-null/src/factory.rs index a76c4f008..da4e3557e 100644 --- a/nisshi-storage-null/src/factory.rs +++ b/nisshi-storage-null/src/factory.rs @@ -14,7 +14,9 @@ use crate::Engine; use async_trait::async_trait; -use nisshi_storage::{ArcDynStorage, Result, StorageFactory, StorageFactoryConfiguration}; +use nisshi_storage::{ + ArcDynStorage, Result, StorageFactory, StorageFactoryConfiguration, reject_unrecognized_options, +}; use regex::Regex; use std::sync::Arc; @@ -28,6 +30,8 @@ impl StorageFactory for EngineFactory { } async fn build(&self, configuration: StorageFactoryConfiguration) -> Result { + reject_unrecognized_options(&configuration.storage, &[])?; + Ok(Arc::new(Box::new(Engine::new( configuration.cluster, configuration.node_id, diff --git a/nisshi-storage-slatedb/src/factory.rs b/nisshi-storage-slatedb/src/factory.rs index 67dd9567e..34c78ef31 100644 --- a/nisshi-storage-slatedb/src/factory.rs +++ b/nisshi-storage-slatedb/src/factory.rs @@ -15,7 +15,10 @@ use std::sync::Arc; use async_trait::async_trait; -use nisshi_storage::{ArcDynStorage, Error, Result, StorageFactory, StorageFactoryConfiguration}; +use nisshi_storage::{ + ArcDynStorage, Error, Result, StorageFactory, StorageFactoryConfiguration, + reject_unrecognized_options, +}; use regex::Regex; use slatedb::{ Db, Settings, @@ -48,6 +51,8 @@ impl StorageFactory for EngineFactory { } async fn build(&self, configuration: StorageFactoryConfiguration) -> Result { + reject_unrecognized_options(&configuration.storage, &[])?; + let host = configuration.storage.host_str().unwrap_or("nisshi"); let db_path = format!("nisshi-{}.slatedb", configuration.cluster); diff --git a/nisshi-storage-sql/src/limbo/factory.rs b/nisshi-storage-sql/src/limbo/factory.rs index 9b238cadc..5430bf6c3 100644 --- a/nisshi-storage-sql/src/limbo/factory.rs +++ b/nisshi-storage-sql/src/limbo/factory.rs @@ -15,7 +15,9 @@ use std::sync::Arc; use async_trait::async_trait; -use nisshi_storage::{ArcDynStorage, Result, StorageFactory, StorageFactoryConfiguration}; +use nisshi_storage::{ + ArcDynStorage, Result, StorageFactory, StorageFactoryConfiguration, reject_unrecognized_options, +}; use regex::Regex; use super::Engine; @@ -30,6 +32,8 @@ impl StorageFactory for LimboFactory { } async fn build(&self, configuration: StorageFactoryConfiguration) -> Result { + reject_unrecognized_options(&configuration.storage, &[])?; + Engine::builder() .storage(configuration.storage) .node(configuration.node_id) diff --git a/nisshi-storage-sql/src/lite/factory.rs b/nisshi-storage-sql/src/lite/factory.rs index 468833f20..ea253323f 100644 --- a/nisshi-storage-sql/src/lite/factory.rs +++ b/nisshi-storage-sql/src/lite/factory.rs @@ -13,7 +13,9 @@ // limitations under the License. use async_trait::async_trait; -use nisshi_storage::{ArcDynStorage, Result, StorageFactory, StorageFactoryConfiguration}; +use nisshi_storage::{ + ArcDynStorage, Result, StorageFactory, StorageFactoryConfiguration, reject_unrecognized_options, +}; use regex::Regex; use super::Engine; @@ -28,6 +30,11 @@ impl StorageFactory for LiteFactory { } async fn build(&self, configuration: StorageFactoryConfiguration) -> Result { + reject_unrecognized_options( + &configuration.storage, + &["vacuum_into", "busy_timeout", "compaction", "mode"], + )?; + Engine::builder() .storage(configuration.storage) .node(configuration.node_id) diff --git a/nisshi-storage-sql/src/pg.rs b/nisshi-storage-sql/src/pg.rs index ebecac730..4b751a29a 100644 --- a/nisshi-storage-sql/src/pg.rs +++ b/nisshi-storage-sql/src/pg.rs @@ -4189,6 +4189,17 @@ mod tests { Ok(()) } + /// A storage URL option that tokio-postgres does not read (`vacuum_into` is an + /// option of the sqlite engine) stops the build before any connection, and the + /// error names the option. + #[test] + fn unrecognized_query_option_rejected_without_connecting() { + let error = Postgres::builder(&format!("{CONNECTION}?vacuum_into=/tmp/does-not-matter")) + .expect_err("an option that tokio-postgres does not read must be rejected"); + + assert!(error.to_string().contains("vacuum_into"), "{error}"); + } + #[test] fn pool_error_timeout_maps_to_retriable_error_code() { let error = Error::from(PoolError::Timeout(TimeoutType::Wait)); diff --git a/nisshi-storage/src/lib.rs b/nisshi-storage/src/lib.rs index eea46dc07..54ea3c284 100644 --- a/nisshi-storage/src/lib.rs +++ b/nisshi-storage/src/lib.rs @@ -322,7 +322,14 @@ pub enum Error { UnknownCacheKey(String), - UnsupportedStorageUrl(Url), + /// A storage URL named a query option that scheme's engine does not + /// recognise (for example a typo, or an option that belongs to a + /// different engine). + UnrecognizedStorageOption { + scheme: String, + option: String, + }, + UnexpectedAddPartitionsToTxnRequest(Box), Url(#[from] url::ParseError), UnknownTxnState(String), @@ -2194,6 +2201,23 @@ pub struct Builder { type PhantomBuilder = Builder, PhantomData, PhantomData, PhantomData>; +/// Returns [`Error::UnrecognizedStorageOption`] for the first of `storage`'s query keys +/// that is not in `recognized`. +/// +/// A [`StorageFactory::build`] passes the options that its engine reads, so a misspelt +/// option, or an option of a different engine, stops startup instead of being ignored. +pub fn reject_unrecognized_options(storage: &Url, recognized: &[&str]) -> Result<()> { + storage + .query_pairs() + .find(|(k, _)| !recognized.contains(&k.as_ref())) + .map_or(Ok(()), |(k, _)| { + Err(Error::UnrecognizedStorageOption { + scheme: storage.scheme().to_owned(), + option: k.into_owned(), + }) + }) +} + impl Builder { pub fn node_id(self, node_id: i32) -> Builder { Builder { @@ -2477,4 +2501,33 @@ mod tests { Ok(()) } + + #[test] + fn reject_unrecognized_options_accepts_listed_keys() -> Result<()> { + reject_unrecognized_options(&Url::parse("sqlite://nisshi.db")?, &[])?; + reject_unrecognized_options( + &Url::parse("sqlite://nisshi.db?vacuum_into=/tmp/x&mode=memory")?, + &["vacuum_into", "mode"], + ) + } + + #[test] + fn reject_unrecognized_options_names_unknown_key_and_scheme() -> Result<()> { + let error = reject_unrecognized_options( + &Url::parse("sqlite://nisshi.db?vacuum_onto=/tmp/x")?, + &["vacuum_into"], + ) + .unwrap_err(); + + assert!( + matches!( + &error, + Error::UnrecognizedStorageOption { scheme, option } + if scheme == "sqlite" && option == "vacuum_onto" + ), + "{error:?}" + ); + + Ok(()) + } } diff --git a/nisshi/src/bin/nisshi.rs b/nisshi/src/bin/nisshi.rs index 0fc82ed4e..0991fbc5a 100644 --- a/nisshi/src/bin/nisshi.rs +++ b/nisshi/src/bin/nisshi.rs @@ -94,6 +94,19 @@ async fn main() -> Result { nisshi_cli::Error::TlsRequiresCertAndKey => { error!("TLS requires both --cert and --key.") } + nisshi_cli::Error::LakeRequiresSchemaRegistry { lake } => error!( + "{lake} requires --schema-registry: a data lake writer validates every record against a schema." + ), + nisshi_cli::Error::Server(e) => match &**e { + nisshi_broker::Error::InvalidStorageOptionValue { option, value } => error!( + "storage option {option}={value} is not a valid interval: give a non-zero duration with a unit, up to 365d (e.g. 500ms, 90s, 10m, 1h30m)" + ), + nisshi_broker::Error::Storage(nisshi_storage::Error::UnrecognizedStorageOption { + scheme, + option, + }) => error!("storage option {option} is not recognised by the {scheme} engine"), + _ => error!("Unknown error occurred during command: {}", err), + }, _ => error!("Unknown error occurred during command: {}", err), }) }