spetz commented on code in PR #4615:
URL: https://github.com/apache/iggy/pull/4615#discussion_r4337691818
##########
.github/actions/utils/server-start/action.yml:
##########
@@ -53,6 +53,14 @@ inputs:
description: "--replica-id <u8> value (leave empty to omit)"
required: false
default: ""
+ encryption-enabled:
+ description: "Start the server with encryption enabled, which the SDK
stream key tests need"
+ required: false
+ default: "true"
Review Comment:
Fixed. `encryption-enabled` now defaults to false
(`.github/actions/utils/server-start/action.yml:56`) and only the jobs that run
server-generated key cases opt in. The Node, Java and PHP TLS runs use the
default and cover the plaintext path again. BDD stays encrypted, because
`stream_options.feature` asserts the generated key
(`bdd/docker-compose.server.yml:73`).
##########
core/binary_protocol/src/codes.rs:
##########
@@ -28,6 +28,7 @@ pub const GET_CLUSTER_METADATA_CODE: u32 = 12;
pub const DESCRIBE_OPTIONS_CODE: u32 = 13;
// Code 14 is retired.
pub const BIND_SESSION_CODE: u32 = 15;
+pub const UPDATE_RESOURCE_CODE: u32 = 16;
Review Comment:
Fixed. Added the `breaking:client-to-replica-protocol` and
`breaking:replica-to-replica-protocol` labels. Both apply: command code 16 is
new (`core/binary_protocol/src/codes.rs:31`), 35, 204 and 304 are retired, the
protocol moves to 0.12.0 and operations 72 to 75 and 150 change the replica
wire.
##########
core/binary_protocol/src/requests/system/update_resource.rs:
##########
@@ -0,0 +1,399 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you 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.
+
+use super::describe_options::{OPTIONS_SCOPE_STREAM, OPTIONS_SCOPE_TOPIC,
OPTIONS_SCOPE_USER};
+use crate::WireError;
+use crate::WireIdentifier;
+use crate::codec::{WireDecode, WireEncode, read_u8};
+use crate::primitives::identifier::WireName;
+use crate::primitives::options::{
+ WireOptions, decode_options_prefixed, encode_options_prefixed,
options_prefixed_size,
+};
+use bytes::{BufMut, BytesMut};
+use std::borrow::Cow;
+use std::collections::BTreeSet;
+
+/// Most reset names one patch carries, the reach of its count byte.
+pub const MAX_RESET_KEYS: usize = u8::MAX as usize;
+
+/// The resource an `UpdateResource` patches, tagged by its options scope code.
+#[derive(Debug, Clone, PartialEq, Eq)]
+pub enum ResourceTarget {
+ Topic {
+ stream_id: WireIdentifier,
+ topic_id: WireIdentifier,
+ },
+ Stream {
+ stream_id: WireIdentifier,
+ },
+ User {
+ user_id: WireIdentifier,
+ },
+}
+
+/// Patch one resource's options: set the keys in `options`, reset the keys in
+/// `reset` to their defaults, leave every other key alone.
+///
+/// Wire format:
+/// ```text
+/// [scope:u8][identifiers][reset_count:u8][reset_key:WireName]*[options to
end of body]
+/// ```
+/// Identifiers are `[stream_id][topic_id]` for a topic, `[stream_id]` for a
+/// stream and `[user_id]` for a user. A patch with nothing to set or reset, a
+/// key named twice, or a key both set and reset is refused at decode.
+#[derive(Debug, Clone, PartialEq, Eq)]
+pub struct UpdateResourceRequest {
+ pub target: ResourceTarget,
+ /// At most [`MAX_RESET_KEYS`] names. Callers bound the list, the encoder
+ /// cannot refuse a longer one.
+ pub reset: Vec<WireName>,
+ pub options: WireOptions,
+}
+
+impl ResourceTarget {
+ /// The options scope code the target is tagged with on the wire.
+ #[must_use]
+ pub const fn scope_code(&self) -> u8 {
+ match self {
+ Self::Topic { .. } => OPTIONS_SCOPE_TOPIC,
+ Self::Stream { .. } => OPTIONS_SCOPE_STREAM,
+ Self::User { .. } => OPTIONS_SCOPE_USER,
+ }
+ }
+}
+
+impl WireEncode for ResourceTarget {
+ fn encoded_size(&self) -> usize {
+ 1 + match self {
+ Self::Topic {
+ stream_id,
+ topic_id,
+ } => stream_id.encoded_size() + topic_id.encoded_size(),
+ Self::Stream { stream_id } => stream_id.encoded_size(),
+ Self::User { user_id } => user_id.encoded_size(),
+ }
+ }
+
+ fn encode(&self, buf: &mut BytesMut) {
+ buf.put_u8(self.scope_code());
+ match self {
+ Self::Topic {
+ stream_id,
+ topic_id,
+ } => {
+ stream_id.encode(buf);
+ topic_id.encode(buf);
+ }
+ Self::Stream { stream_id } => stream_id.encode(buf),
+ Self::User { user_id } => user_id.encode(buf),
+ }
+ }
+}
+
+impl WireDecode for ResourceTarget {
+ fn decode(buf: &[u8]) -> Result<(Self, usize), WireError> {
+ let scope = read_u8(buf, 0)?;
+ let mut position = 1;
+ let target = match scope {
+ OPTIONS_SCOPE_TOPIC => {
+ let (stream_id, consumed) =
WireIdentifier::decode(&buf[position..])?;
+ position += consumed;
+ let (topic_id, consumed) =
WireIdentifier::decode(&buf[position..])?;
+ position += consumed;
+ Self::Topic {
+ stream_id,
+ topic_id,
+ }
+ }
+ OPTIONS_SCOPE_STREAM => {
+ let (stream_id, consumed) =
WireIdentifier::decode(&buf[position..])?;
+ position += consumed;
+ Self::Stream { stream_id }
+ }
+ OPTIONS_SCOPE_USER => {
+ let (user_id, consumed) =
WireIdentifier::decode(&buf[position..])?;
+ position += consumed;
+ Self::User { user_id }
+ }
+ unknown => {
+ return Err(WireError::Validation(Cow::Owned(format!(
+ "unknown resource scope: {unknown}"
+ ))));
+ }
+ };
+ Ok((target, position))
+ }
+}
+
+impl WireEncode for UpdateResourceRequest {
+ fn encoded_size(&self) -> usize {
+ self.target.encoded_size()
+ + 1
+ + self
+ .reset
+ .iter()
+ .map(WireEncode::encoded_size)
+ .sum::<usize>()
+ + self.options.encoded_size()
+ }
+
+ fn encode(&self, buf: &mut BytesMut) {
+ self.target.encode(buf);
+ debug_assert!(self.reset.len() <= MAX_RESET_KEYS);
+ #[allow(clippy::cast_possible_truncation)]
+ buf.put_u8(self.reset.len() as u8);
Review Comment:
Fixed. The reset list is now the bounded `WireResetKeys` newtype
(`core/binary_protocol/src/requests/system/update_resource.rs:72`), whose
`TryFrom` refuses more than `MAX_RESET_KEYS` names, so the count byte always
matches the names and the `debug_assert!` is gone.
`given_more_reset_names_than_the_count_byte_holds_when_building_should_refuse`
pins the bound.
##########
core/binary_protocol/src/version.rs:
##########
@@ -102,17 +102,23 @@ const PATCH_MASK: u32 = COMPONENT_MAX;
/// Current development protocol version, independent of package versions.
///
/// Version 0.11.1 requires the session-binding secret in login requests.
+/// Version 0.12.0 replaces the per-resource update commands with the generic
Review Comment:
Fixed. The 0.12.0 note (`core/binary_protocol/src/version.rs:105`) and the
MIN rationale (`core/binary_protocol/src/version.rs:117`) now name the topic
header dropping message expiry, compression and the size cap, 50 to 33 fixed
bytes. The note also names the flags byte each `DescribeOptions` descriptor
gained.
##########
core/cli/src/args/stream.rs:
##########
@@ -72,6 +75,15 @@ pub(crate) struct StreamCreateArgs {
/// Legacy stream ID flag (ignored)
#[clap(short, long)]
pub(crate) stream_id: Option<u32>,
+ /// File holding the stream's own at-rest encryption key
Review Comment:
Fixed. The flag is now `--stream-key-file` and its help asks for 32 random
bytes per stream (`core/cli/src/args/stream.rs:78`). The
`IGGY_STREAM_ENCRYPTION_KEY` fallback is gone, so the key comes only from the
file named on that call
(`core/cli/src/commands/binary_streams/create_stream.rs:46`).
`given_environment_key_when_creating_stream_should_ignore_it` pins it.
##########
core/common/examples/resource_options_fixtures.rs:
##########
@@ -0,0 +1,937 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you 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.
+
+#[path = "resource_options_fixtures/schema.rs"]
+mod schema;
+
+use base64::Engine;
+use base64::engine::general_purpose::STANDARD;
+use iggy_binary_protocol::requests::system::DescribeOptionsRequest;
+use iggy_binary_protocol::responses::streams::{GetStreamResponse,
StreamResponse, TopicHeader};
+use iggy_binary_protocol::responses::system::DescribeOptionsResponse;
+use iggy_binary_protocol::responses::topics::GetTopicResponse;
+use iggy_binary_protocol::responses::users::{UserDetailsResponse,
UserResponse};
+use iggy_binary_protocol::{WireDecode, WireEncode, WireName};
+use iggy_common::wire_conversions::{resource_options_from_wire,
resource_options_to_wire_split};
+use iggy_common::{
+ CATALOG, Durability, HeaderKey, HeaderValue, OptionColumns, OptionKey,
OptionStorage,
+ OptionValue, OptionVisibility, OptionsScope, ResourceOptions,
StreamDetails,
+ StreamEncryptionKey, StreamEncryptionSource, TopicDetails,
UserInfoDetails, UserStatus,
+ descriptors_for, public_options, stream_key_fingerprint,
+};
+use schema::{
+ CANONICAL_PATH, CatalogRow, CreateStreamInput, CreateStreamVector,
CreateTopicInput,
+ CreateTopicVector, CreateUserInput, CreateUserVector,
DescribeOptionsVector, DescribedOption,
+ ErrorExpectation, GENERATOR, HttpUpdateVector, IdentifierInput,
InvalidKeyVector, KeyInput,
+ LANGUAGE_COPIES, PatchInput, ResourceOptionsFixture, SCHEMA_VERSION,
SCOPES, StreamKeyState,
+ StreamKeyVectors, StreamProjection, StreamResponseVector, TargetInput,
TopicProjection,
+ TopicResponseVector, UpdateResourceErrorVector, UpdateResourceVector,
UserProjection,
+ UserResponseVector, WireConstants, admit_create, admit_update,
contains_bytes,
+ describe_options_response, evaluate_http_update, update_resource_request,
+};
+use std::collections::BTreeMap;
+use std::error::Error;
+use std::fs;
+use std::iter;
+use std::path::Path;
+
+const FINGERPRINT_LABEL: &str = "iggy/stream-key-fingerprint/v1";
+const FIRST_KEK_CONFIG_KEY: &str =
"/rvT1xP4V8u1EAhk4xDdqzqM2UOPXyy9XYkl4uRShgE=";
+const SECOND_KEK_CONFIG_KEY: &str =
"AQIDBAUGBwgJCgsMDQ4PEBESExQVFhcYGRobHB0eHyA=";
+const CREATED_AT_MICROS: u64 = 1_710_000_000_000_000;
+const SEVEN_DAYS_MICROS: &str = "604800000000";
+const ONE_DAY_MICROS: &str = "86400000000";
+const ABOVE_2_POW_53: u64 = (1 << 53) + 1;
+
+type GeneratorResult<T> = Result<T, Box<dyn Error>>;
+
+fn main() -> GeneratorResult<()> {
+ let fixture = ResourceOptionsFixture {
+ schema_version: SCHEMA_VERSION,
+ generated_by: GENERATOR.to_owned(),
+ conventions: BTreeMap::from(
+ [
+ (
+ "byte_strings",
+ "Every *_hex field is lowercase hex without separators.",
+ ),
+ (
+ "u64_values",
+ "u64 option values and typed u64 inputs are decimal
strings, so values above 2^53 keep their precision.",
+ ),
+ (
+ "message_expiry",
+ "Microseconds. 18446744073709551615 means never expire.",
+ ),
+ (
+ "max_topic_size",
+ "Bytes. 18446744073709551615 means unlimited.",
+ ),
+ (
+ "option_maps",
+ "Option maps use the HTTP JSON form: value as reversible
text, kind, and explicit provenance. Request blocks mark every entry explicit.",
+ ),
+ (
+ "options_block_order",
+ "Options block entries are encoded in ascending key byte
order.",
+ ),
+ (
+ "reset_order",
Review Comment:
Fixed. The convention now says reset keys are written in ascending key byte
order (`core/common/examples/resource_options_fixtures.rs:94`), as every
encoder and `topic_reset_both_defaults` do. The fixture and its eight SDK
copies regenerate byte-identical, and
`foreign/csharp/Iggy_SDK/Contracts/OptionsPatch.cs:25` says the same.
##########
core/metadata/src/stm/snapshot.rs:
##########
@@ -59,7 +59,11 @@ use crate::stm::user::UsersSnapshot;
///
/// Version 12: pending consumer group owner generations equal their metadata
/// operation, replacing independently incremented generations.
-pub const SNAPSHOT_FORMAT_VERSION: u32 = 12;
+///
+/// Version 13: `TopicSnapshot` dropped its typed expiry, compression and size
+/// fields, which the options map already holds, and `StreamSnapshot` gained
+/// `created_op`, the identity of a stream incarnation.
+pub const SNAPSHOT_FORMAT_VERSION: u32 = 13;
Review Comment:
Fixed. `STORAGE_FORMAT` moves to `IGGY-RESOURCE-OPTIONS-1`
(`core/server/src/boot/mod.rs:90`), so a pre-PR data dir is refused unchanged
before the journal opens and a pre-PR replica cannot join.
`incompatible_storage_is_refused_without_changing_existing_files` now refuses
the `IGGY-BOUND-GRANTS-4` marker too.
##########
core/common/src/utils/stream_key.rs:
##########
@@ -0,0 +1,1042 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you 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.
+
+use crate::IggyError;
+use crate::utils::crypto::{Aes256GcmEncryptor, EncryptorKind};
+use crate::utils::text;
+use aes_gcm::aead::{Aead, Generate, Payload};
+use aes_gcm::{Aes256Gcm, KeyInit, Nonce};
+use base64::Engine;
+use base64::engine::general_purpose::STANDARD;
+use ring::rand::{SecureRandom, SystemRandom};
+use ring::{digest, hkdf};
+use secrecy::zeroize::Zeroizing;
+use secrecy::{ExposeSecret, ExposeSecretMut, SecretBox};
+use std::fmt::{Debug, Formatter};
+use std::str::FromStr;
+use std::sync::Arc;
+
+pub use iggy_binary_protocol::requests::streams::{
+ MAX_WRAP_SET_ENVELOPES, WRAPPED_STREAM_KEY_LENGTH,
+};
+
+pub const STREAM_KEY_LENGTH: usize = 32;
+/// Most retired config keys a node keeps for unwrapping during a rotation.
+pub const MAX_PREVIOUS_KEYS: usize = MAX_WRAP_SET_ENVELOPES - 1;
+
+const WRAPPED_STREAM_KEY_VERSION: u8 = 1;
+const FINGERPRINT_LENGTH: usize = 8;
+const NONCE_LENGTH: usize = 12;
+const WRAPPED_KEY_CIPHERTEXT_LENGTH: usize = STREAM_KEY_LENGTH + 16;
+const ORIGIN_OFFSET: usize = 1;
+const KEK_FINGERPRINT_OFFSET: usize = ORIGIN_OFFSET + 1;
+const NONCE_OFFSET: usize = KEK_FINGERPRINT_OFFSET + FINGERPRINT_LENGTH;
+const CIPHERTEXT_OFFSET: usize = NONCE_OFFSET + NONCE_LENGTH;
+const KEY_FINGERPRINT_OFFSET: usize = CIPHERTEXT_OFFSET +
WRAPPED_KEY_CIPHERTEXT_LENGTH;
+const STREAM_KEY_WRAP_INFO: &[u8] = b"iggy/stream-key-wrap/v1";
+const STREAM_KEY_AAD_LABEL: &[u8] = b"iggy/stream-key/v1";
+const STREAM_KEY_FINGERPRINT_LABEL: &[u8] = b"iggy/stream-key-fingerprint/v1";
+const KEK_FINGERPRINT_LABEL: &[u8] = b"iggy/stream-kek-fingerprint/v1";
+
+/// A stream's own data encryption key: 32 random bytes.
+///
+/// Never printed and never serialized. The bytes leave this type only to be
+/// wrapped, encoded into a create request, or turned into an encryptor.
+pub struct StreamEncryptionKey(SecretBox<[u8; STREAM_KEY_LENGTH]>);
+
+/// Who chose a stream's key, recorded and authenticated in its envelope.
+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+#[repr(u8)]
+pub enum StreamKeyOrigin {
+ /// The client sent the key with the create.
+ Client = 1,
+ /// The server generated the key because the client sent none.
+ Server = 2,
+}
+
+/// A stream key wrapped under one key-encryption key.
+///
+/// Layout:
`[version:1][origin:1][kek_fingerprint:8][nonce:12][ciphertext:32][tag:16][key_fingerprint:8]`.
+/// The version, the origin and both fingerprints are authenticated as
+/// associated data.
+#[derive(Clone, PartialEq, Eq)]
+pub struct WrappedStreamKey([u8; WRAPPED_STREAM_KEY_LENGTH]);
+
+/// One stream key wrapped under one or more key-encryption keys: one envelope
+/// outside a rotation, one per configured key during it. All envelopes hold
+/// the same key from the same origin, sorted by KEK fingerprint, none twice.
+#[derive(Clone, PartialEq, Eq)]
+pub struct WrapSet {
+ envelopes: Vec<WrappedStreamKey>,
+ bytes: Vec<u8>,
+}
+
+/// The key-encryption keys derived from the server config: the current one,
+/// which wraps, and the previous ones a rotation still unwraps with.
+#[derive(Clone)]
+pub struct ServerKeyContext {
+ current: Option<KeyEncryptionKey>,
+ previous: Vec<KeyEncryptionKey>,
+ generates: bool,
+}
+
+#[derive(Clone)]
+struct KeyEncryptionKey {
+ cipher: Aes256Gcm,
+ fingerprint: [u8; FINGERPRINT_LENGTH],
+}
+
+struct DerivedKeyLength;
+
+impl StreamEncryptionKey {
+ /// A new random key from the operating system's secure generator.
+ ///
+ /// # Errors
+ ///
+ /// `CannotEncryptData` when the generator fails.
+ pub fn generate() -> Result<Self, IggyError> {
+ let mut key = SecretBox::new(Box::new([0u8; STREAM_KEY_LENGTH]));
+ SystemRandom::new()
+ .fill(key.expose_secret_mut())
+ .map_err(|_| IggyError::CannotEncryptData)?;
+ Ok(Self(key))
+ }
+
+ /// 16 lowercase hex characters identifying this key without revealing it.
+ #[must_use]
+ pub fn fingerprint(&self) -> String {
Review Comment:
Fixed. The key type, `fingerprint` and `stream_key_fingerprint` now say the
fingerprint hides only a uniformly random key
(`core/common/src/utils/stream_key.rs:55`,
`core/common/src/utils/stream_key.rs:126`), and the CLI help asks for 32 random
bytes (`core/cli/src/args/stream.rs:80`). Admission still accepts any 32 bytes,
because the server cannot tell a random key from a derived one.
##########
core/common/src/traits/binary_impls/users.rs:
##########
@@ -108,28 +111,17 @@ impl<B: BinaryClient> UserClient for B {
async fn update_user(
&self,
user_id: &Identifier,
- username: Option<&str>,
- status: Option<UserStatus>,
options: &UserUpdateOptions,
) -> Result<(), IggyError> {
fail_if_not_authenticated(self).await?;
- let wire_id = identifier_to_wire(user_id)?;
- let wire_username = username
- .map(WireName::new)
- .transpose()
- .map_err(|_| IggyError::InvalidFormat)?;
- self.send_raw_with_response(
- UPDATE_USER_CODE,
- UpdateUserRequest {
- user_id: wire_id,
- username: wire_username,
- status: status.map(|s| s.as_code()),
- options: options.to_wire()?,
- }
- .to_bytes(),
- )
- .await?;
- Ok(())
+ if let Some(username) = &options.username {
Review Comment:
Fixed. Binary `update_user` no longer pre-checks the username
(`core/common/src/traits/binary_impls/users.rs:117`), so TCP answers
`OptionNotResettable("status")` like HTTP.
`given_short_username_and_status_reset_when_updating_user_should_report_admission_category`
and the new `user_short_username_and_status_reset` fixture case pin it in
every SDK.
##########
core/common/src/utils/stream_key.rs:
##########
@@ -0,0 +1,1042 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you 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.
+
+use crate::IggyError;
+use crate::utils::crypto::{Aes256GcmEncryptor, EncryptorKind};
+use crate::utils::text;
+use aes_gcm::aead::{Aead, Generate, Payload};
+use aes_gcm::{Aes256Gcm, KeyInit, Nonce};
+use base64::Engine;
+use base64::engine::general_purpose::STANDARD;
+use ring::rand::{SecureRandom, SystemRandom};
+use ring::{digest, hkdf};
+use secrecy::zeroize::Zeroizing;
+use secrecy::{ExposeSecret, ExposeSecretMut, SecretBox};
+use std::fmt::{Debug, Formatter};
+use std::str::FromStr;
+use std::sync::Arc;
+
+pub use iggy_binary_protocol::requests::streams::{
+ MAX_WRAP_SET_ENVELOPES, WRAPPED_STREAM_KEY_LENGTH,
+};
+
+pub const STREAM_KEY_LENGTH: usize = 32;
+/// Most retired config keys a node keeps for unwrapping during a rotation.
+pub const MAX_PREVIOUS_KEYS: usize = MAX_WRAP_SET_ENVELOPES - 1;
+
+const WRAPPED_STREAM_KEY_VERSION: u8 = 1;
+const FINGERPRINT_LENGTH: usize = 8;
+const NONCE_LENGTH: usize = 12;
+const WRAPPED_KEY_CIPHERTEXT_LENGTH: usize = STREAM_KEY_LENGTH + 16;
+const ORIGIN_OFFSET: usize = 1;
+const KEK_FINGERPRINT_OFFSET: usize = ORIGIN_OFFSET + 1;
+const NONCE_OFFSET: usize = KEK_FINGERPRINT_OFFSET + FINGERPRINT_LENGTH;
+const CIPHERTEXT_OFFSET: usize = NONCE_OFFSET + NONCE_LENGTH;
+const KEY_FINGERPRINT_OFFSET: usize = CIPHERTEXT_OFFSET +
WRAPPED_KEY_CIPHERTEXT_LENGTH;
+const STREAM_KEY_WRAP_INFO: &[u8] = b"iggy/stream-key-wrap/v1";
+const STREAM_KEY_AAD_LABEL: &[u8] = b"iggy/stream-key/v1";
+const STREAM_KEY_FINGERPRINT_LABEL: &[u8] = b"iggy/stream-key-fingerprint/v1";
+const KEK_FINGERPRINT_LABEL: &[u8] = b"iggy/stream-kek-fingerprint/v1";
+
+/// A stream's own data encryption key: 32 random bytes.
+///
+/// Never printed and never serialized. The bytes leave this type only to be
+/// wrapped, encoded into a create request, or turned into an encryptor.
+pub struct StreamEncryptionKey(SecretBox<[u8; STREAM_KEY_LENGTH]>);
+
+/// Who chose a stream's key, recorded and authenticated in its envelope.
+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+#[repr(u8)]
+pub enum StreamKeyOrigin {
+ /// The client sent the key with the create.
+ Client = 1,
+ /// The server generated the key because the client sent none.
+ Server = 2,
+}
+
+/// A stream key wrapped under one key-encryption key.
+///
+/// Layout:
`[version:1][origin:1][kek_fingerprint:8][nonce:12][ciphertext:32][tag:16][key_fingerprint:8]`.
+/// The version, the origin and both fingerprints are authenticated as
+/// associated data.
+#[derive(Clone, PartialEq, Eq)]
+pub struct WrappedStreamKey([u8; WRAPPED_STREAM_KEY_LENGTH]);
+
+/// One stream key wrapped under one or more key-encryption keys: one envelope
+/// outside a rotation, one per configured key during it. All envelopes hold
+/// the same key from the same origin, sorted by KEK fingerprint, none twice.
+#[derive(Clone, PartialEq, Eq)]
+pub struct WrapSet {
+ envelopes: Vec<WrappedStreamKey>,
+ bytes: Vec<u8>,
+}
+
+/// The key-encryption keys derived from the server config: the current one,
+/// which wraps, and the previous ones a rotation still unwraps with.
+#[derive(Clone)]
+pub struct ServerKeyContext {
+ current: Option<KeyEncryptionKey>,
+ previous: Vec<KeyEncryptionKey>,
+ generates: bool,
+}
+
+#[derive(Clone)]
+struct KeyEncryptionKey {
+ cipher: Aes256Gcm,
+ fingerprint: [u8; FINGERPRINT_LENGTH],
+}
+
+struct DerivedKeyLength;
+
+impl StreamEncryptionKey {
+ /// A new random key from the operating system's secure generator.
+ ///
+ /// # Errors
+ ///
+ /// `CannotEncryptData` when the generator fails.
+ pub fn generate() -> Result<Self, IggyError> {
+ let mut key = SecretBox::new(Box::new([0u8; STREAM_KEY_LENGTH]));
+ SystemRandom::new()
+ .fill(key.expose_secret_mut())
+ .map_err(|_| IggyError::CannotEncryptData)?;
+ Ok(Self(key))
+ }
+
+ /// 16 lowercase hex characters identifying this key without revealing it.
+ #[must_use]
+ pub fn fingerprint(&self) -> String {
+ hex(&self.fingerprint_bytes())
+ }
+
+ /// The raw key, for the request encoder and the stream encryptor.
+ #[must_use]
+ pub fn as_bytes(&self) -> &[u8; STREAM_KEY_LENGTH] {
+ self.0.expose_secret()
+ }
+
+ /// The encryptor for this stream's message payloads and user headers.
+ ///
+ /// # Errors
+ ///
+ /// `InvalidEncryptionKey` when the cipher refuses the key.
+ pub fn encryptor(&self) -> Result<Arc<EncryptorKind>, IggyError> {
+ Ok(Arc::new(EncryptorKind::Aes256Gcm(Aes256GcmEncryptor::new(
+ self.as_bytes(),
+ )?)))
+ }
+
+ fn fingerprint_bytes(&self) -> [u8; FINGERPRINT_LENGTH] {
+ labelled_fingerprint(STREAM_KEY_FINGERPRINT_LABEL, self.as_bytes())
+ }
+}
+
+impl WrappedStreamKey {
+ /// Who chose the key inside this envelope.
+ #[must_use]
+ pub fn origin(&self) -> StreamKeyOrigin {
+ // Construction refuses any other origin byte.
+ if self.0[ORIGIN_OFFSET] == StreamKeyOrigin::Server as u8 {
+ StreamKeyOrigin::Server
+ } else {
+ StreamKeyOrigin::Client
+ }
+ }
+
+ /// Fingerprint of the stream key inside this envelope.
+ #[must_use]
+ pub fn key_fingerprint(&self) -> String {
+ hex(self.key_fingerprint_bytes())
+ }
+
+ /// Fingerprint of the key-encryption key that wrapped this envelope.
+ #[must_use]
+ pub fn kek_fingerprint(&self) -> String {
+ hex(self.kek_fingerprint_bytes())
+ }
+
+ fn key_fingerprint_bytes(&self) -> &[u8] {
+ &self.0[KEY_FINGERPRINT_OFFSET..]
+ }
+
+ fn kek_fingerprint_bytes(&self) -> &[u8] {
+ &self.0[KEK_FINGERPRINT_OFFSET..NONCE_OFFSET]
+ }
+
+ fn associated_data(&self) -> Vec<u8> {
+ associated_data(
+ self.0[ORIGIN_OFFSET],
+ self.kek_fingerprint_bytes(),
+ self.key_fingerprint_bytes(),
+ )
+ }
+}
+
+impl WrapSet {
+ /// A set of `envelopes` in canonical order.
+ ///
+ /// # Errors
+ ///
+ /// `UnsupportedCriticalOption` for an empty set, more than three
+ /// envelopes, a KEK fingerprint twice, or envelopes of different keys or
+ /// origins.
+ pub fn new(mut envelopes: Vec<WrappedStreamKey>) -> Result<Self,
IggyError> {
+ envelopes.sort_by(|left, right| {
+ left.kek_fingerprint_bytes()
+ .cmp(right.kek_fingerprint_bytes())
+ });
+ let Some(first) = envelopes.first() else {
+ return Err(unsupported_wrapped_key());
+ };
+ let consistent = envelopes.len() <= MAX_WRAP_SET_ENVELOPES
+ && envelopes.iter().all(|envelope| {
+ envelope.origin() == first.origin()
+ && envelope.key_fingerprint_bytes() ==
first.key_fingerprint_bytes()
+ })
+ && envelopes
+ .windows(2)
+ .all(|pair| pair[0].kek_fingerprint_bytes() <
pair[1].kek_fingerprint_bytes());
+ if !consistent {
+ return Err(unsupported_wrapped_key());
+ }
+ let bytes = envelopes.iter().flat_map(|envelope| envelope.0).collect();
+ Ok(Self { envelopes, bytes })
+ }
+
+ #[must_use]
+ pub fn envelopes(&self) -> &[WrappedStreamKey] {
+ &self.envelopes
+ }
+
+ /// Who chose the key every envelope holds.
+ #[must_use]
+ pub fn origin(&self) -> StreamKeyOrigin {
+ self.envelopes[0].origin()
+ }
+
+ /// Fingerprint of the key every envelope holds.
+ #[must_use]
+ pub fn key_fingerprint(&self) -> String {
+ self.envelopes[0].key_fingerprint()
+ }
+
+ /// The raw fingerprint of the key every envelope holds.
+ #[must_use]
+ pub fn key_fingerprint_bytes(&self) -> &[u8] {
+ self.envelopes[0].key_fingerprint_bytes()
+ }
+
+ /// Fingerprints of the key-encryption keys, in canonical order.
+ #[must_use]
+ pub fn kek_fingerprints(&self) -> Vec<String> {
+ self.envelopes
+ .iter()
+ .map(WrappedStreamKey::kek_fingerprint)
+ .collect()
+ }
+}
+
+impl ServerKeyContext {
+ /// No config key: streams stay plaintext and stream keys are refused.
+ #[must_use]
+ pub const fn disabled() -> Self {
+ Self {
+ current: None,
+ previous: Vec::new(),
+ generates: false,
+ }
+ }
+
+ /// Derive the key-encryption keys from the base64 config keys.
+ ///
+ /// An empty `key` is valid only when encryption is disabled and no
+ /// previous key is set, and then nothing is derived. A supplied `key`
+ /// always derives the key-encryption key, so a client may bring its own
+ /// stream key while keyless streams stay plaintext. `enabled` decides
+ /// whether a keyless stream gets a generated key.
+ ///
+ /// # Errors
+ ///
+ /// `InvalidEncryptionKey` when a key is not 32 bytes of base64, when
+ /// `key` is empty while `enabled` is true or previous keys are set, for
+ /// more than two previous keys, or for a key configured twice.
+ pub fn from_config(
+ key: &str,
+ previous_keys: &[String],
+ enabled: bool,
+ ) -> Result<Self, IggyError> {
+ if key.is_empty() {
+ return if enabled || !previous_keys.is_empty() {
+ Err(IggyError::InvalidEncryptionKey)
+ } else {
+ Ok(Self::disabled())
+ };
+ }
+ if previous_keys.len() > MAX_PREVIOUS_KEYS {
+ return Err(IggyError::InvalidEncryptionKey);
+ }
+ let current = KeyEncryptionKey::from_config_key(key)?;
+ let previous = previous_keys
+ .iter()
+ .map(|previous| KeyEncryptionKey::from_config_key(previous))
+ .collect::<Result<Vec<_>, _>>()?;
+ let mut fingerprints: Vec<_> = previous.iter().map(|kek|
kek.fingerprint).collect();
+ fingerprints.push(current.fingerprint);
+ fingerprints.sort_unstable();
+ if fingerprints.windows(2).any(|pair| pair[0] == pair[1]) {
+ return Err(IggyError::InvalidEncryptionKey);
+ }
+ Ok(Self {
+ current: Some(current),
+ previous,
+ generates: enabled,
+ })
+ }
+
+ /// Whether a stream may bring its own key on this node.
+ #[must_use]
+ pub const fn accepts_stream_keys(&self) -> bool {
+ self.current.is_some()
+ }
+
+ /// Whether this node gives a keyless stream a generated key.
+ #[must_use]
+ pub const fn generates_stream_keys(&self) -> bool {
+ self.current.is_some() && self.generates
+ }
+
+ /// Fingerprint of the key-encryption key that wraps, when a key is set.
+ #[must_use]
+ pub fn kek_fingerprint(&self) -> Option<String> {
+ self.current.as_ref().map(|kek| hex(&kek.fingerprint))
+ }
+
+ /// Fingerprints of every configured key-encryption key, current first.
+ #[must_use]
+ pub fn kek_fingerprints(&self) -> Vec<String> {
+ self.current
+ .iter()
+ .chain(&self.previous)
+ .map(|kek| hex(&kek.fingerprint))
+ .collect()
+ }
+
+ /// Wrap a stream key under the current key-encryption key.
+ ///
+ /// # Errors
+ ///
+ /// `StreamEncryptionUnavailable` without a config key, `CannotEncryptData`
+ /// when the cipher fails.
+ pub fn wrap(
+ &self,
+ key: &StreamEncryptionKey,
+ origin: StreamKeyOrigin,
+ ) -> Result<WrappedStreamKey, IggyError> {
+ let kek = self
+ .current
+ .as_ref()
+ .ok_or(IggyError::StreamEncryptionUnavailable)?;
+ let key_fingerprint = key.fingerprint_bytes();
+ let aad = associated_data(origin as u8, &kek.fingerprint,
&key_fingerprint);
+ let nonce = Nonce::generate();
+ let sealed = kek
+ .cipher
+ .encrypt(
+ &nonce,
+ Payload {
+ msg: key.as_bytes(),
+ aad: &aad,
+ },
+ )
+ .map_err(|_| IggyError::CannotEncryptData)?;
+ if sealed.len() != WRAPPED_KEY_CIPHERTEXT_LENGTH {
+ return Err(IggyError::CannotEncryptData);
+ }
+ let mut envelope = [0u8; WRAPPED_STREAM_KEY_LENGTH];
+ envelope[0] = WRAPPED_STREAM_KEY_VERSION;
+ envelope[ORIGIN_OFFSET] = origin as u8;
+
envelope[KEK_FINGERPRINT_OFFSET..NONCE_OFFSET].copy_from_slice(&kek.fingerprint);
+ envelope[NONCE_OFFSET..CIPHERTEXT_OFFSET].copy_from_slice(&nonce);
+
envelope[CIPHERTEXT_OFFSET..KEY_FINGERPRINT_OFFSET].copy_from_slice(&sealed);
+ envelope[KEY_FINGERPRINT_OFFSET..].copy_from_slice(&key_fingerprint);
+ Ok(WrappedStreamKey(envelope))
+ }
+
+ /// Authenticate and unwrap one envelope with the configured key-encryption
+ /// key it names.
+ ///
+ /// # Errors
+ ///
+ /// `StreamEncryptionUnavailable` without a config key, `CannotDecryptData`
+ /// when no configured key matches, the envelope does not authenticate, or
+ /// its key fingerprint does not match the key inside.
+ pub fn unwrap(&self, wrapped: &WrappedStreamKey) ->
Result<StreamEncryptionKey, IggyError> {
+ if self.current.is_none() {
+ return Err(IggyError::StreamEncryptionUnavailable);
+ }
+ let kek = self
+ .current
+ .iter()
+ .chain(&self.previous)
+ .find(|kek| kek.fingerprint == wrapped.kek_fingerprint_bytes())
+ .ok_or(IggyError::CannotDecryptData)?;
+ let nonce =
Nonce::try_from(&wrapped.0[NONCE_OFFSET..CIPHERTEXT_OFFSET])
+ .map_err(|_| IggyError::CannotDecryptData)?;
+ let plaintext = Zeroizing::new(
+ kek.cipher
+ .decrypt(
+ &nonce,
+ Payload {
+ msg:
&wrapped.0[CIPHERTEXT_OFFSET..KEY_FINGERPRINT_OFFSET],
+ aad: &wrapped.associated_data(),
+ },
+ )
+ .map_err(|_| IggyError::CannotDecryptData)?,
+ );
+ let key = StreamEncryptionKey::try_from(plaintext.as_slice())
+ .map_err(|_| IggyError::CannotDecryptData)?;
+ if key.fingerprint_bytes() != wrapped.key_fingerprint_bytes() {
+ return Err(IggyError::CannotDecryptData);
+ }
+ Ok(key)
+ }
+
+ /// The stream key of a wrap set, from the first envelope that opens.
+ ///
+ /// # Errors
+ ///
+ /// As [`Self::unwrap`], from the last envelope tried.
+ pub fn open(&self, set: &WrapSet) -> Result<StreamEncryptionKey,
IggyError> {
+ let mut last = IggyError::CannotDecryptData;
+ for envelope in set.envelopes() {
+ match self.unwrap(envelope) {
+ Ok(key) => return Ok(key),
+ Err(error) => last = error,
+ }
+ }
+ Err(last)
+ }
+
+ /// The set a rotation should store instead of `stored`: the envelopes
+ /// under configured keys, plus one under the current key when missing.
+ /// `None` when nothing changes, or when no envelope opens, because a set
+ /// this node cannot open is never touched.
+ ///
+ /// # Errors
+ ///
+ /// `CannotEncryptData` when wrapping fails.
+ pub fn target_set(&self, stored: &WrapSet) -> Result<Option<WrapSet>,
IggyError> {
+ let Some(current) = &self.current else {
+ return Ok(None);
+ };
+ let configured = |envelope: &&WrappedStreamKey| {
+ self.current
+ .iter()
+ .chain(&self.previous)
+ .any(|kek| kek.fingerprint == envelope.kek_fingerprint_bytes())
+ };
+ let kept: Vec<WrappedStreamKey> = stored
Review Comment:
Fixed. `target_set` keeps every wrap that is not retired and never re-adds
wraps under previous keys, so phase 1 and phase 2 primaries never undo each
other (`core/common/src/utils/stream_key.rs:510`). A wrap is dropped only when
its KEK is listed in `retired_kek_fingerprints`
(`core/configs/src/common/system.rs:88`).
`given_old_config_primary_when_sets_hold_new_wraps_should_keep_them` pins it.
##########
core/common/src/types/options/mod.rs:
##########
@@ -525,74 +600,467 @@ fn raw_options_map(raw: &BTreeMap<String, String>) ->
Result<ResourceOptions, Ig
Ok(options)
}
-/// Encode string key-values into an options block, all marked explicit.
-///
-/// # Errors
-///
-/// See [`raw_options_map`].
-fn raw_options_to_wire(raw: &BTreeMap<String, String>) -> Result<WireOptions,
IggyError> {
- crate::wire_conversions::resource_options_to_wire(
- &raw_options_map(raw)?,
- OptionsProvenance::All,
- )
+/// An update's options in wire form: the keys to set and the keys to reset
+/// to their defaults. Every other key keeps its stored value.
+#[derive(Debug, Clone, PartialEq, Eq)]
+pub struct OptionsPatch {
+ pub set: WireOptions,
+ pub reset: Vec<WireName>,
}
-/// The options an `UpdateStream` may carry.
-///
-/// Absent keys leave the stream's existing values alone -- an update patches
-/// the option map, it does not replace it. See [`TopicUpdateOptions`] for why.
+/// The options a `CreateStream` may carry.
+#[derive(Clone, Default, PartialEq)]
+pub struct StreamCreateOptions {
+ /// The stream's own data encryption key. `None` follows the server
+ /// encryption setting. The server stores the key wrapped and never
+ /// returns it, reads report only its source and fingerprint.
+ pub encryption_key: Option<StreamEncryptionKey>,
+ /// Keys sent as `String` values, for keys newer than this build. A typed
+ /// field wins on collision.
+ pub raw: BTreeMap<String, String>,
+}
+
+/// The options an update may change on a stream. Absent keys keep their
+/// stored values.
#[derive(Debug, Clone, Default, PartialEq)]
pub struct StreamUpdateOptions {
- /// Keys sent as `String` values, checked against
- /// [`UPDATABLE_STREAM_OPTION_KEYS`] server-side. That list is empty, so
- /// every key is currently refused by name; the field exists so the first
- /// updatable stream key costs a catalog entry, not a wire change.
+ /// `None` keeps the stream's name.
+ pub name: Option<String>,
+ /// Keys to reset to their defaults. No stream key is resettable yet, so a
+ /// non-empty set is refused before anything is sent.
+ pub reset: BTreeSet<OptionKey>,
+ /// Keys to reset by name, for keys newer than this build. The server
+ /// decides whether each one may be reset.
+ pub raw_reset: BTreeSet<String>,
+ /// Keys sent as `String` values, for updatable keys newer than this build.
pub raw: BTreeMap<String, String>,
}
-impl StreamUpdateOptions {
- /// Encode the present keys into an options block.
+/// The options an update may change on a topic. Absent keys keep their stored
+/// values. Create-only settings (compression, `segment_size`, durability,
+/// flush thresholds, preallocation, resize policy) have no field here, so a
+/// caller cannot send one only to learn at the server that it was refused.
+#[derive(Debug, Clone, Default, PartialEq)]
+pub struct TopicUpdateOptions {
+ /// `None` keeps the topic's name.
+ pub name: Option<String>,
+ /// `None` keeps the topic's current expiry. `ServerDefault` changes
+ /// nothing, use `reset` to return to the default.
+ pub message_expiry: Option<IggyExpiry>,
+ /// `None` keeps the topic's current cap. `ServerDefault` changes nothing,
+ /// use `reset` to return to the default.
+ pub max_topic_size: Option<MaxTopicSize>,
+ /// Keys to reset to their defaults: `message_expiry` or `max_topic_size`.
+ pub reset: BTreeSet<OptionKey>,
+ /// Keys to reset by name, for keys newer than this build. The server
+ /// decides whether each one may be reset.
+ pub raw_reset: BTreeSet<String>,
+ /// Keys sent as `String` values, for updatable keys newer than this build.
+ pub raw: BTreeMap<String, String>,
+}
+
+/// The options a `CreateUser` may carry. No user key exists yet, the type
+/// exists so the first one costs a field, not a signature change.
+#[derive(Debug, Clone, Default, PartialEq)]
+pub struct UserCreateOptions {
+ /// Keys sent as `String` values, for keys newer than this build.
+ pub raw: BTreeMap<String, String>,
+}
+
+/// The options an update may change on a user. Absent keys keep their stored
+/// values.
+#[derive(Debug, Clone, Default, PartialEq)]
+pub struct UserUpdateOptions {
+ /// `None` keeps the username.
+ pub username: Option<String>,
+ /// `None` keeps the status. Deactivating a user ends its sessions.
+ pub status: Option<UserStatus>,
+ /// Keys to reset to their defaults. No user key is resettable yet, so a
+ /// non-empty set is refused before anything is sent.
+ pub reset: BTreeSet<OptionKey>,
+ /// Keys to reset by name, for keys newer than this build. The server
+ /// decides whether each one may be reset.
+ pub raw_reset: BTreeSet<String>,
+ /// Keys sent as `String` values, for updatable keys newer than this build.
+ pub raw: BTreeMap<String, String>,
+}
+
+/// How a stream's messages are stored, read from its committed options.
+#[derive(Debug, Clone, PartialEq)]
+pub enum StreamEncryption {
+ /// No wrap set: the stream is plaintext for life.
+ Plaintext,
+ /// Encrypted under the key the wrap set holds.
+ Wrapped(WrapSet),
+ /// Critical state this build cannot read. The stream is never served as
+ /// plaintext then.
+ Unavailable(IggyError),
+}
+
+/// A topic's effective settings, every one resolved: the stored value, or the
+/// catalog default for a key the map does not hold.
+#[derive(Debug, Clone, Copy, PartialEq)]
+pub struct TopicResolvedOptions {
+ pub compression_algorithm: CompressionAlgorithm,
+ pub message_expiry: IggyExpiry,
+ pub max_topic_size: MaxTopicSize,
+ pub partition_resize_policy: PartitionResizePolicy,
+ /// The partition-facing knobs, with the shard-default semantics of
+ /// [`TopicRuntimeOptions`].
+ pub runtime: TopicRuntimeOptions,
+}
+
+impl StreamCreateOptions {
+ /// Encode the present keys into a create options block.
///
/// # Errors
///
- /// See `raw_options_to_wire`.
+ /// See `raw_options_map`.
pub fn to_wire(&self) -> Result<WireOptions, IggyError> {
- raw_options_to_wire(&self.raw)
+ crate::wire_conversions::resource_options_to_wire(
+ &self.to_option_map()?,
+ OptionsProvenance::All,
+ )
+ }
+
+ /// The present keys as string values for an HTTP body. The key travels as
+ /// canonical standard base64.
+ ///
+ /// # Errors
+ ///
+ /// See `raw_options_map`.
+ pub fn to_input_map(&self) -> Result<BTreeMap<String, OptionInputValue>,
IggyError> {
+ option_input_map(&self.to_option_map()?)
+ }
+
+ fn to_option_map(&self) -> Result<ResourceOptions, IggyError> {
+ let mut options = raw_options_map(&self.raw)?;
+ if let Some(key) = &self.encryption_key {
+ options.insert(
+ HeaderKey::from(OptionKey::EncryptionKey),
+
OptionValue::explicit(HeaderValue::try_from(key.as_bytes().as_slice())?),
+ );
+ }
+ Ok(options)
}
}
-/// The options an `UpdateUser` may carry. Same patch semantics as
-/// [`StreamUpdateOptions`].
-#[derive(Debug, Clone, Default, PartialEq)]
-pub struct UserUpdateOptions {
- /// Keys sent as `String` values, checked against
- /// [`UPDATABLE_USER_OPTION_KEYS`] server-side. That list is empty, so
every
- /// key is currently refused by name; the field exists so the first
- /// updatable user key costs a catalog entry, not a wire change.
- pub raw: BTreeMap<String, String>,
+impl StreamUpdateOptions {
+ /// Encode the patch for `UpdateResource`.
+ ///
+ /// # Errors
+ ///
+ /// `InvalidCommand` when the patch changes nothing or names a key both
+ /// set and reset, the server's admission error for a key that cannot be
+ /// reset, `InvalidStreamName` for a name that does not fit.
+ pub fn to_patch(&self) -> Result<OptionsPatch, IggyError> {
+ options_patch(
+ OptionsScope::Stream,
+ self.to_option_map()?,
+ &self.reset,
+ &self.raw_reset,
+ !self.raw.is_empty(),
+ )
+ }
+
+ /// The patch as an HTTP update body: set keys with their values, reset
keys
+ /// in the `reset` array.
+ ///
+ /// # Errors
+ ///
+ /// See [`Self::to_patch`].
+ pub fn to_update_body(&self) -> Result<OptionsUpdate, IggyError> {
+ update_body(
+ OptionsScope::Stream,
+ self.to_option_map()?,
+ &self.reset,
+ &self.raw_reset,
+ !self.raw.is_empty(),
+ )
+ }
+
+ fn to_option_map(&self) -> Result<ResourceOptions, IggyError> {
+ let mut options = raw_options_map(&self.raw)?;
+ if let Some(name) = &self.name {
+ insert_text(&mut options, OptionKey::Name, name);
+ }
+ Ok(options)
+ }
}
-impl UserUpdateOptions {
- /// Encode the present keys into an options block.
+impl TopicUpdateOptions {
+ /// Encode the patch for `UpdateResource`, in canonical kinds.
///
/// # Errors
///
- /// See `raw_options_to_wire`.
+ /// `InvalidCommand` when the patch changes nothing or names a key both
+ /// set and reset, `OptionNotUpdatable` for a create-only key,
+ /// `OptionNotResettable` for a key without a default, `InvalidTopicName`
+ /// for a name that does not fit.
+ pub fn to_patch(&self) -> Result<OptionsPatch, IggyError> {
+ options_patch(
+ OptionsScope::Topic,
+ self.to_option_map()?,
+ &self.reset,
+ &self.raw_reset,
+ !self.raw.is_empty(),
+ )
+ }
+
+ /// The patch as an HTTP update body.
+ ///
+ /// # Errors
+ ///
+ /// See [`Self::to_patch`].
+ pub fn to_update_body(&self) -> Result<OptionsUpdate, IggyError> {
+ update_body(
+ OptionsScope::Topic,
+ self.to_option_map()?,
+ &self.reset,
+ &self.raw_reset,
+ !self.raw.is_empty(),
+ )
+ }
+
+ fn to_option_map(&self) -> Result<ResourceOptions, IggyError> {
+ let mut options = raw_options_map(&self.raw)?;
+ if let Some(name) = &self.name {
+ insert_text(&mut options, OptionKey::Name, name);
+ }
+ if let Some(message_expiry) = self.message_expiry
+ && message_expiry != IggyExpiry::ServerDefault
+ {
+ options.insert(
+ HeaderKey::from(OptionKey::MessageExpiry),
+ OptionValue::explicit(message_expiry.encode()?),
+ );
+ }
+ if let Some(max_topic_size) = self.max_topic_size
+ && max_topic_size != MaxTopicSize::ServerDefault
+ {
+ options.insert(
+ HeaderKey::from(OptionKey::MaxTopicSize),
+ OptionValue::explicit(max_topic_size.encode()?),
+ );
+ }
+ Ok(options)
+ }
+}
+
+impl UserCreateOptions {
+ /// Encode the present keys into a create options block.
+ ///
+ /// # Errors
+ ///
+ /// See `raw_options_map`.
pub fn to_wire(&self) -> Result<WireOptions, IggyError> {
- raw_options_to_wire(&self.raw)
+ crate::wire_conversions::resource_options_to_wire(
+ &raw_options_map(&self.raw)?,
+ OptionsProvenance::All,
+ )
+ }
+
+ /// The present keys as values for an HTTP body.
+ ///
+ /// # Errors
+ ///
+ /// See `raw_options_map`.
+ pub fn to_input_map(&self) -> Result<BTreeMap<String, OptionInputValue>,
IggyError> {
+ option_input_map(&raw_options_map(&self.raw)?)
}
}
-/// This node's configured fallbacks for the runtime knobs, used to fill the
-/// derived-options block for keys a client did not send.
-#[derive(Debug, Clone, Copy, PartialEq)]
-pub struct TopicRuntimeDefaults {
- pub segment_size: IggyByteSize,
- pub durability: Durability,
- pub consumer_offset_durability: Durability,
- pub messages_required_to_save: u32,
- pub size_of_messages_required_to_save: IggyByteSize,
- pub preallocate_segments: bool,
+impl UserUpdateOptions {
+ /// Encode the patch for `UpdateResource`.
+ ///
+ /// # Errors
+ ///
+ /// `InvalidCommand` when the patch changes nothing or names a key both
+ /// set and reset, the server's admission error for a key that cannot be
+ /// reset, `InvalidUsername` for a username that does not fit.
+ pub fn to_patch(&self) -> Result<OptionsPatch, IggyError> {
+ options_patch(
+ OptionsScope::User,
+ self.to_option_map()?,
+ &self.reset,
+ &self.raw_reset,
+ !self.raw.is_empty(),
+ )
+ }
+
+ /// The patch as an HTTP update body.
+ ///
+ /// # Errors
+ ///
+ /// See [`Self::to_patch`].
+ pub fn to_update_body(&self) -> Result<OptionsUpdate, IggyError> {
+ update_body(
+ OptionsScope::User,
+ self.to_option_map()?,
+ &self.reset,
+ &self.raw_reset,
+ !self.raw.is_empty(),
+ )
+ }
+
+ fn to_option_map(&self) -> Result<ResourceOptions, IggyError> {
+ let mut options = raw_options_map(&self.raw)?;
+ if let Some(username) = &self.username {
+ insert_text(&mut options, OptionKey::Username, username);
+ }
+ if let Some(status) = self.status {
+ options.insert(
+ HeaderKey::from(OptionKey::Status),
+ OptionValue::explicit(status.encode()?),
+ );
+ }
+ Ok(options)
+ }
+}
+
+impl From<&ResourceOptions> for StreamEncryption {
+ fn from(options: &ResourceOptions) -> Self {
+ match check_critical_options(options).and_then(|()|
stream_wrap_set(options)) {
+ Ok(Some(set)) => Self::Wrapped(set),
+ Ok(None) => Self::Plaintext,
+ Err(error) => Self::Unavailable(error),
+ }
+ }
+}
+
+impl From<&ResourceOptions> for TopicResolvedOptions {
Review Comment:
Fixed. `TopicRuntimeOptions` now comes from the one parse through
`From<&TopicCreateOptions>` (`core/common/src/types/options/mod.rs:1149`). The
retention and segment size readers and the partition resize gates read
`topic.resolved` instead of parsing the map
(`core/metadata/src/stm/stream.rs:1234`,
`core/metadata/src/stm/stream.rs:2688`).
##########
core/common/src/types/options/mod.rs:
##########
@@ -53,21 +53,38 @@
//! Java each assert independently, so a new encoder has a fixture to match
Review Comment:
Fixed. The module docs (`core/common/src/types/options/mod.rs:18`) now cover
creation and `UpdateResource` patches, list all nine SDKs with their typed
options and raw escape hatch (`core/common/src/types/options/mod.rs:34`), and
name the `resource_options.json` fixture every SDK asserts.
##########
core/common/src/types/options/mod.rs:
##########
@@ -525,74 +600,467 @@ fn raw_options_map(raw: &BTreeMap<String, String>) ->
Result<ResourceOptions, Ig
Ok(options)
}
-/// Encode string key-values into an options block, all marked explicit.
-///
-/// # Errors
-///
-/// See [`raw_options_map`].
-fn raw_options_to_wire(raw: &BTreeMap<String, String>) -> Result<WireOptions,
IggyError> {
- crate::wire_conversions::resource_options_to_wire(
- &raw_options_map(raw)?,
- OptionsProvenance::All,
- )
+/// An update's options in wire form: the keys to set and the keys to reset
+/// to their defaults. Every other key keeps its stored value.
+#[derive(Debug, Clone, PartialEq, Eq)]
+pub struct OptionsPatch {
+ pub set: WireOptions,
+ pub reset: Vec<WireName>,
}
-/// The options an `UpdateStream` may carry.
-///
-/// Absent keys leave the stream's existing values alone -- an update patches
-/// the option map, it does not replace it. See [`TopicUpdateOptions`] for why.
+/// The options a `CreateStream` may carry.
+#[derive(Clone, Default, PartialEq)]
+pub struct StreamCreateOptions {
+ /// The stream's own data encryption key. `None` follows the server
+ /// encryption setting. The server stores the key wrapped and never
+ /// returns it, reads report only its source and fingerprint.
+ pub encryption_key: Option<StreamEncryptionKey>,
+ /// Keys sent as `String` values, for keys newer than this build. A typed
+ /// field wins on collision.
+ pub raw: BTreeMap<String, String>,
+}
+
+/// The options an update may change on a stream. Absent keys keep their
+/// stored values.
#[derive(Debug, Clone, Default, PartialEq)]
pub struct StreamUpdateOptions {
- /// Keys sent as `String` values, checked against
- /// [`UPDATABLE_STREAM_OPTION_KEYS`] server-side. That list is empty, so
- /// every key is currently refused by name; the field exists so the first
- /// updatable stream key costs a catalog entry, not a wire change.
+ /// `None` keeps the stream's name.
+ pub name: Option<String>,
+ /// Keys to reset to their defaults. No stream key is resettable yet, so a
+ /// non-empty set is refused before anything is sent.
+ pub reset: BTreeSet<OptionKey>,
+ /// Keys to reset by name, for keys newer than this build. The server
+ /// decides whether each one may be reset.
+ pub raw_reset: BTreeSet<String>,
+ /// Keys sent as `String` values, for updatable keys newer than this build.
pub raw: BTreeMap<String, String>,
}
-impl StreamUpdateOptions {
- /// Encode the present keys into an options block.
+/// The options an update may change on a topic. Absent keys keep their stored
+/// values. Create-only settings (compression, `segment_size`, durability,
+/// flush thresholds, preallocation, resize policy) have no field here, so a
+/// caller cannot send one only to learn at the server that it was refused.
+#[derive(Debug, Clone, Default, PartialEq)]
+pub struct TopicUpdateOptions {
+ /// `None` keeps the topic's name.
+ pub name: Option<String>,
+ /// `None` keeps the topic's current expiry. `ServerDefault` changes
+ /// nothing, use `reset` to return to the default.
+ pub message_expiry: Option<IggyExpiry>,
+ /// `None` keeps the topic's current cap. `ServerDefault` changes nothing,
+ /// use `reset` to return to the default.
+ pub max_topic_size: Option<MaxTopicSize>,
+ /// Keys to reset to their defaults: `message_expiry` or `max_topic_size`.
+ pub reset: BTreeSet<OptionKey>,
+ /// Keys to reset by name, for keys newer than this build. The server
+ /// decides whether each one may be reset.
+ pub raw_reset: BTreeSet<String>,
+ /// Keys sent as `String` values, for updatable keys newer than this build.
+ pub raw: BTreeMap<String, String>,
+}
+
+/// The options a `CreateUser` may carry. No user key exists yet, the type
+/// exists so the first one costs a field, not a signature change.
+#[derive(Debug, Clone, Default, PartialEq)]
+pub struct UserCreateOptions {
+ /// Keys sent as `String` values, for keys newer than this build.
+ pub raw: BTreeMap<String, String>,
+}
+
+/// The options an update may change on a user. Absent keys keep their stored
+/// values.
+#[derive(Debug, Clone, Default, PartialEq)]
+pub struct UserUpdateOptions {
+ /// `None` keeps the username.
+ pub username: Option<String>,
+ /// `None` keeps the status. Deactivating a user ends its sessions.
+ pub status: Option<UserStatus>,
+ /// Keys to reset to their defaults. No user key is resettable yet, so a
+ /// non-empty set is refused before anything is sent.
+ pub reset: BTreeSet<OptionKey>,
+ /// Keys to reset by name, for keys newer than this build. The server
+ /// decides whether each one may be reset.
+ pub raw_reset: BTreeSet<String>,
+ /// Keys sent as `String` values, for updatable keys newer than this build.
+ pub raw: BTreeMap<String, String>,
+}
+
+/// How a stream's messages are stored, read from its committed options.
+#[derive(Debug, Clone, PartialEq)]
+pub enum StreamEncryption {
+ /// No wrap set: the stream is plaintext for life.
+ Plaintext,
+ /// Encrypted under the key the wrap set holds.
+ Wrapped(WrapSet),
+ /// Critical state this build cannot read. The stream is never served as
+ /// plaintext then.
+ Unavailable(IggyError),
+}
+
+/// A topic's effective settings, every one resolved: the stored value, or the
+/// catalog default for a key the map does not hold.
+#[derive(Debug, Clone, Copy, PartialEq)]
+pub struct TopicResolvedOptions {
+ pub compression_algorithm: CompressionAlgorithm,
+ pub message_expiry: IggyExpiry,
+ pub max_topic_size: MaxTopicSize,
+ pub partition_resize_policy: PartitionResizePolicy,
+ /// The partition-facing knobs, with the shard-default semantics of
+ /// [`TopicRuntimeOptions`].
+ pub runtime: TopicRuntimeOptions,
+}
+
+impl StreamCreateOptions {
+ /// Encode the present keys into a create options block.
///
/// # Errors
///
- /// See `raw_options_to_wire`.
+ /// See `raw_options_map`.
pub fn to_wire(&self) -> Result<WireOptions, IggyError> {
- raw_options_to_wire(&self.raw)
+ crate::wire_conversions::resource_options_to_wire(
+ &self.to_option_map()?,
+ OptionsProvenance::All,
+ )
+ }
+
+ /// The present keys as string values for an HTTP body. The key travels as
+ /// canonical standard base64.
+ ///
+ /// # Errors
+ ///
+ /// See `raw_options_map`.
+ pub fn to_input_map(&self) -> Result<BTreeMap<String, OptionInputValue>,
IggyError> {
+ option_input_map(&self.to_option_map()?)
+ }
+
+ fn to_option_map(&self) -> Result<ResourceOptions, IggyError> {
+ let mut options = raw_options_map(&self.raw)?;
+ if let Some(key) = &self.encryption_key {
+ options.insert(
+ HeaderKey::from(OptionKey::EncryptionKey),
+
OptionValue::explicit(HeaderValue::try_from(key.as_bytes().as_slice())?),
+ );
+ }
+ Ok(options)
}
}
-/// The options an `UpdateUser` may carry. Same patch semantics as
-/// [`StreamUpdateOptions`].
-#[derive(Debug, Clone, Default, PartialEq)]
-pub struct UserUpdateOptions {
- /// Keys sent as `String` values, checked against
- /// [`UPDATABLE_USER_OPTION_KEYS`] server-side. That list is empty, so
every
- /// key is currently refused by name; the field exists so the first
- /// updatable user key costs a catalog entry, not a wire change.
- pub raw: BTreeMap<String, String>,
+impl StreamUpdateOptions {
+ /// Encode the patch for `UpdateResource`.
+ ///
+ /// # Errors
+ ///
+ /// `InvalidCommand` when the patch changes nothing or names a key both
+ /// set and reset, the server's admission error for a key that cannot be
+ /// reset, `InvalidStreamName` for a name that does not fit.
+ pub fn to_patch(&self) -> Result<OptionsPatch, IggyError> {
+ options_patch(
+ OptionsScope::Stream,
+ self.to_option_map()?,
+ &self.reset,
+ &self.raw_reset,
+ !self.raw.is_empty(),
+ )
+ }
+
+ /// The patch as an HTTP update body: set keys with their values, reset
keys
+ /// in the `reset` array.
+ ///
+ /// # Errors
+ ///
+ /// See [`Self::to_patch`].
+ pub fn to_update_body(&self) -> Result<OptionsUpdate, IggyError> {
+ update_body(
+ OptionsScope::Stream,
+ self.to_option_map()?,
+ &self.reset,
+ &self.raw_reset,
+ !self.raw.is_empty(),
+ )
+ }
+
+ fn to_option_map(&self) -> Result<ResourceOptions, IggyError> {
+ let mut options = raw_options_map(&self.raw)?;
+ if let Some(name) = &self.name {
+ insert_text(&mut options, OptionKey::Name, name);
+ }
+ Ok(options)
+ }
}
-impl UserUpdateOptions {
- /// Encode the present keys into an options block.
+impl TopicUpdateOptions {
+ /// Encode the patch for `UpdateResource`, in canonical kinds.
///
/// # Errors
///
- /// See `raw_options_to_wire`.
+ /// `InvalidCommand` when the patch changes nothing or names a key both
+ /// set and reset, `OptionNotUpdatable` for a create-only key,
+ /// `OptionNotResettable` for a key without a default, `InvalidTopicName`
+ /// for a name that does not fit.
+ pub fn to_patch(&self) -> Result<OptionsPatch, IggyError> {
+ options_patch(
+ OptionsScope::Topic,
+ self.to_option_map()?,
+ &self.reset,
+ &self.raw_reset,
+ !self.raw.is_empty(),
+ )
+ }
+
+ /// The patch as an HTTP update body.
+ ///
+ /// # Errors
+ ///
+ /// See [`Self::to_patch`].
+ pub fn to_update_body(&self) -> Result<OptionsUpdate, IggyError> {
+ update_body(
+ OptionsScope::Topic,
+ self.to_option_map()?,
+ &self.reset,
+ &self.raw_reset,
+ !self.raw.is_empty(),
+ )
+ }
+
+ fn to_option_map(&self) -> Result<ResourceOptions, IggyError> {
+ let mut options = raw_options_map(&self.raw)?;
+ if let Some(name) = &self.name {
+ insert_text(&mut options, OptionKey::Name, name);
+ }
+ if let Some(message_expiry) = self.message_expiry
Review Comment:
Fixed. Admission now refuses with `InvalidCommand` an update whose set,
columns and reset are all empty once zero sentinels are dropped
(`core/common/src/types/options/catalog.rs:730`), so HTTP
`{"message_expiry":0}` gets a 400 and nothing is replicated.
`given_update_of_only_zero_sentinels_when_admitting_should_refuse_it_as_empty`
and the `topic_zero_sentinel_only` fixture pin it.
##########
core/common/src/types/options/catalog.rs:
##########
@@ -0,0 +1,1659 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you 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.
+
+use super::codec::{OptionValueCodec, invalid};
+use super::key::OptionKey;
+use super::{
+ DEFAULT_MAX_TOPIC_SIZE, DEFAULT_MESSAGE_EXPIRY,
DEFAULT_MESSAGES_REQUIRED_TO_SAVE,
+ DEFAULT_PREALLOCATE_SEGMENTS, DEFAULT_SEGMENT_SIZE,
DEFAULT_SIZE_OF_MESSAGES_REQUIRED_TO_SAVE,
+ Durability, MAX_MESSAGES_REQUIRED_TO_SAVE, MAX_PREALLOCATED_TOPIC_BYTES,
+ MAX_SIZE_OF_MESSAGES_REQUIRED_TO_SAVE, MAX_TOPIC_SEGMENT_SIZE,
MIN_TOPIC_SEGMENT_SIZE,
+ OptionValue, OptionsProvenance, OptionsScope, PartitionResizePolicy,
ResourceOptions,
+};
+use crate::http::users::defaults::{MAX_USERNAME_LENGTH, MIN_USERNAME_LENGTH};
+use crate::types::compression::compression_algorithm::CompressionAlgorithm;
+use crate::types::message::{HeaderKey, HeaderKind, HeaderValue};
+use crate::utils::stream_key::{StreamEncryptionKey, StreamKeyOrigin, WrapSet};
+use crate::{IggyByteSize, IggyError, IggyExpiry, MaxTopicSize, UserStatus};
+use iggy_binary_protocol::{WireHeaderKind, WireOptions, WireUserHeaderEntry};
+use serde::{Deserialize, Serialize};
+use std::collections::{BTreeMap, BTreeSet};
+use std::fmt::{Debug, Formatter};
+use strum::{Display, EnumString, IntoStaticStr};
+
+/// Bit set in a `DescribeOptions` entry's flags when an update may set the
key.
+pub const OPTION_FLAG_UPDATABLE: u8 = 1;
+/// Bit set when an update may reset the key to its default.
+pub const OPTION_FLAG_RESETTABLE: u8 = 1 << 1;
+/// Bit set when the key is write-only: accepted, never returned.
+pub const OPTION_FLAG_SECRET: u8 = 1 << 2;
+/// Bit set when the server computes the key and refuses it from clients.
+pub const OPTION_FLAG_READ_ONLY: u8 = 1 << 3;
+/// Bit set when a node that does not understand the key must not serve the
+/// resource.
+pub const OPTION_FLAG_CRITICAL: u8 = 1 << 4;
+
+/// Longest marker a resource may carry. Every critical key name fits many
+/// times over, and the bound keeps the marker a single header field.
+const CRITICAL_KEYS_MAX_LENGTH: usize = 255;
+/// Longest key echoed back in an option error. A rejected key is
+/// attacker-sized, and over HTTP the error text rides the response body.
+const ERROR_KEY_PREVIEW_LENGTH: usize = 64;
+const MAX_RESOURCE_NAME_LENGTH: usize = 255;
+
+const ALL_SCOPES: &[OptionsScope] = &[
+ OptionsScope::Stream,
+ OptionsScope::Topic,
+ OptionsScope::User,
+];
+const STREAM_AND_TOPIC: &[OptionsScope] = &[OptionsScope::Stream,
OptionsScope::Topic];
+const STREAM_ONLY: &[OptionsScope] = &[OptionsScope::Stream];
+const TOPIC_ONLY: &[OptionsScope] = &[OptionsScope::Topic];
+const USER_ONLY: &[OptionsScope] = &[OptionsScope::User];
+
+/// Every option key, indexed by its `OptionKey` discriminant.
+pub static CATALOG: [OptionDescriptor; 19] = [
+ OptionDescriptor {
+ key: OptionKey::Name,
+ scopes: STREAM_AND_TOPIC,
+ kind: HeaderKind::String,
+ storage: OptionStorage::Column,
+ mutability: OptionMutability::Updatable,
+ default: OptionDefault::None,
+ visibility: OptionVisibility::Public,
+ critical: false,
+ description: || {
+ "Resource name, 1 to 255 bytes, unique within its parent. Set by
the create \
+ command's name field and changed through an update."
+ .to_owned()
+ },
+ },
+ OptionDescriptor {
+ key: OptionKey::Username,
+ scopes: USER_ONLY,
+ kind: HeaderKind::String,
+ storage: OptionStorage::Column,
+ mutability: OptionMutability::Updatable,
+ default: OptionDefault::None,
+ visibility: OptionVisibility::Public,
+ critical: false,
+ description: || {
+ format!(
+ "User name, {MIN_USERNAME_LENGTH} to {MAX_USERNAME_LENGTH}
bytes, unique. Set by \
+ the create command's username field and changed through an
update."
+ )
+ },
+ },
+ OptionDescriptor {
+ key: OptionKey::Status,
+ scopes: USER_ONLY,
+ kind: HeaderKind::String,
+ storage: OptionStorage::Column,
+ mutability: OptionMutability::Updatable,
+ default: OptionDefault::None,
+ visibility: OptionVisibility::Public,
+ critical: false,
+ description: || {
+ "User status, active or inactive. Set by the create command's
status field and \
+ changed through an update. Deactivating a user ends its sessions."
+ .to_owned()
+ },
+ },
+ OptionDescriptor {
+ key: OptionKey::CompressionAlgorithm,
+ scopes: TOPIC_ONLY,
+ kind: HeaderKind::String,
+ storage: OptionStorage::Map,
+ mutability: OptionMutability::CreateOnly,
+ default: OptionDefault::Const(||
CompressionAlgorithm::default().encode()),
+ visibility: OptionVisibility::Public,
+ critical: false,
+ description: || {
+ "Compression algorithm (none, gzip). Stored and reported, not
applied to messages"
+ .to_owned()
+ },
+ },
+ OptionDescriptor {
+ key: OptionKey::MessageExpiry,
+ scopes: TOPIC_ONLY,
+ kind: HeaderKind::Uint64,
+ storage: OptionStorage::Map,
+ mutability: OptionMutability::Updatable,
+ default: OptionDefault::Const(||
Ok(HeaderValue::from(DEFAULT_MESSAGE_EXPIRY))),
+ visibility: OptionVisibility::Public,
+ critical: false,
+ description: || {
+ "Message expiry in microseconds, or a humantime string (e.g. 7
days)".to_owned()
+ },
+ },
+ OptionDescriptor {
+ key: OptionKey::MaxTopicSize,
+ scopes: TOPIC_ONLY,
+ kind: HeaderKind::Uint64,
+ storage: OptionStorage::Map,
+ mutability: OptionMutability::Updatable,
+ default: OptionDefault::Const(||
Ok(HeaderValue::from(DEFAULT_MAX_TOPIC_SIZE))),
+ visibility: OptionVisibility::Public,
+ critical: false,
+ description: || {
+ "Topic-wide sealed-segment size cap, split across all partitions,
in bytes or a \
+ byte-size string (e.g. 1 GiB); finite values must be at least the
segment size"
+ .to_owned()
+ },
+ },
+ OptionDescriptor {
+ key: OptionKey::SegmentSize,
+ scopes: TOPIC_ONLY,
+ kind: HeaderKind::Uint64,
+ storage: OptionStorage::Map,
+ mutability: OptionMutability::CreateOnly,
+ default: OptionDefault::Const(||
Ok(HeaderValue::from(DEFAULT_SEGMENT_SIZE))),
+ visibility: OptionVisibility::Public,
+ critical: false,
+ description: || {
+ format!(
+ "Segment size in bytes, or a byte-size string (e.g. 128 MiB);
a 512-byte \
+ multiple within
{MIN_TOPIC_SEGMENT_SIZE}..={MAX_TOPIC_SEGMENT_SIZE}"
+ )
+ },
+ },
+ OptionDescriptor {
+ key: OptionKey::Durability,
+ scopes: TOPIC_ONLY,
+ kind: HeaderKind::String,
+ storage: OptionStorage::Map,
+ mutability: OptionMutability::CreateOnly,
+ default: OptionDefault::Const(|| Durability::default().encode()),
+ visibility: OptionVisibility::Public,
+ critical: false,
+ description: || {
+ "Message completion: replicated or persisted. Independently
defaults to \
+ replicated. A singleton quorum has one copy. In replicated
groups, persisted \
+ messages use WAL references to segment bodies, retaining their
inodes by hard \
+ link until WAL reclamation. The full body size still counts
against \
+ partition.wal_bytes_max."
+ .to_owned()
+ },
+ },
+ OptionDescriptor {
+ key: OptionKey::ConsumerOffsetDurability,
+ scopes: TOPIC_ONLY,
+ kind: HeaderKind::String,
+ storage: OptionStorage::Map,
+ mutability: OptionMutability::CreateOnly,
+ default: OptionDefault::Const(|| Durability::default().encode()),
+ visibility: OptionVisibility::Public,
+ critical: false,
+ description: || {
+ "Explicit offset completion: replicated or persisted.
Independently defaults to \
+ replicated. Poll auto-commit remains asynchronous. In replicated
groups, persisted \
+ offsets also enable WAL references to segment bodies, retaining
their inodes by \
+ hard link until reclamation, even with replicated message
durability. Full body \
+ sizes count against partition.wal_bytes_max."
+ .to_owned()
+ },
+ },
+ OptionDescriptor {
+ key: OptionKey::PartitionResizePolicy,
+ scopes: TOPIC_ONLY,
+ kind: HeaderKind::String,
+ storage: OptionStorage::Map,
+ mutability: OptionMutability::CreateOnly,
+ default: OptionDefault::Const(||
PartitionResizePolicy::default().encode()),
+ visibility: OptionVisibility::Public,
+ critical: false,
+ description: || {
+ "Immutable after creation: mutable permits partition
addition/deletion without a \
+ bounded cutover or per-key ordering across layouts; fixed rejects
both. Retention \
+ and topic deletion remain destructive."
+ .to_owned()
+ },
+ },
+ OptionDescriptor {
+ key: OptionKey::MessagesRequiredToSave,
+ scopes: TOPIC_ONLY,
+ kind: HeaderKind::Uint32,
+ storage: OptionStorage::Map,
+ mutability: OptionMutability::CreateOnly,
+ default: OptionDefault::Const(||
Ok(HeaderValue::from(DEFAULT_MESSAGES_REQUIRED_TO_SAVE))),
+ visibility: OptionVisibility::Public,
+ critical: false,
+ description: || {
+ format!(
+ "Ordinary message-count flush trigger;
1..={MAX_MESSAGES_REQUIRED_TO_SAVE}. \
+ Required persistence, capacity pressure, and lifecycle
operations can flush \
+ earlier"
+ )
+ },
+ },
+ OptionDescriptor {
+ key: OptionKey::SizeOfMessagesRequiredToSave,
+ scopes: TOPIC_ONLY,
+ kind: HeaderKind::Uint64,
+ storage: OptionStorage::Map,
+ mutability: OptionMutability::CreateOnly,
+ default: OptionDefault::Const(|| {
+ Ok(HeaderValue::from(DEFAULT_SIZE_OF_MESSAGES_REQUIRED_TO_SAVE))
+ }),
+ visibility: OptionVisibility::Public,
+ critical: false,
+ description: || {
+ format!(
+ "Flush the journal once it holds this many bytes, or a
byte-size string; \
+ whichever threshold trips first flushes. At most \
+ {MAX_SIZE_OF_MESSAGES_REQUIRED_TO_SAVE}"
+ )
+ },
+ },
+ OptionDescriptor {
+ key: OptionKey::PreallocateSegments,
+ scopes: TOPIC_ONLY,
+ kind: HeaderKind::Bool,
+ storage: OptionStorage::Map,
+ mutability: OptionMutability::CreateOnly,
+ default: OptionDefault::Const(||
Ok(HeaderValue::from(DEFAULT_PREALLOCATE_SEGMENTS))),
+ visibility: OptionVisibility::Public,
+ critical: false,
+ description: || {
+ format!(
+ "Request segment_size bytes of disk reservation when each
segment opens. \
+ Unsupported or failed reservations fall back to ordinary
allocation with a \
+ warning. Reservation runs inline on the owning shard.
segment_size * \
+ partitions_count is capped at {MAX_PREALLOCATED_TOPIC_BYTES}
bytes per create"
+ )
+ },
+ },
+ OptionDescriptor {
+ key: OptionKey::EncryptionKey,
+ scopes: STREAM_ONLY,
+ kind: HeaderKind::Raw,
+ storage: OptionStorage::Map,
+ mutability: OptionMutability::CreateOnly,
+ default: OptionDefault::None,
+ visibility: OptionVisibility::Secret,
+ critical: true,
+ description: || {
+ "The stream's own data encryption key: 32 random bytes, raw or
standard padded \
+ base64. Create-only and write-only: the server stores it wrapped
under a key \
+ derived from the server encryption key and never returns it.
Requires the server \
+ encryption key to be configured. Without it, a server with
encryption enabled \
+ generates a key for the stream. The key travels in the create
request over the \
+ transport the client uses."
+ .to_owned()
+ },
+ },
+ OptionDescriptor {
+ key: OptionKey::EncryptionKeyWrapped,
+ scopes: STREAM_ONLY,
+ kind: HeaderKind::Raw,
+ storage: OptionStorage::Map,
+ mutability: OptionMutability::CreateOnly,
+ default: OptionDefault::None,
+ visibility: OptionVisibility::Internal,
+ critical: true,
+ description: || {
+ "The stream key wrapped by the server under one or more
key-encryption keys, \
+ never returned"
+ .to_owned()
+ },
+ },
+ OptionDescriptor {
+ key: OptionKey::EncryptionKeySource,
+ scopes: STREAM_ONLY,
+ kind: HeaderKind::String,
+ storage: OptionStorage::Computed,
+ mutability: OptionMutability::ReadOnly,
+ default: OptionDefault::None,
+ visibility: OptionVisibility::ReadOnly,
+ critical: false,
+ description: || {
+ "Who chose the stream's at-rest key: stream when the client sent
it, server when \
+ the server generated it, none when the stream's messages are
stored unencrypted"
+ .to_owned()
+ },
+ },
+ OptionDescriptor {
+ key: OptionKey::EncryptionKeyFingerprint,
+ scopes: STREAM_ONLY,
+ kind: HeaderKind::String,
+ storage: OptionStorage::Computed,
+ mutability: OptionMutability::ReadOnly,
+ default: OptionDefault::None,
+ visibility: OptionVisibility::ReadOnly,
+ critical: false,
+ description: || {
+ "The first 8 bytes of SHA-256 over the label
iggy/stream-key-fingerprint/v1 and \
+ the stream's key, as 16 lowercase hex characters. Absent when the
source is none"
+ .to_owned()
+ },
+ },
+ OptionDescriptor {
+ key: OptionKey::EncryptionKekFingerprint,
+ scopes: STREAM_ONLY,
+ kind: HeaderKind::String,
+ storage: OptionStorage::Computed,
+ mutability: OptionMutability::ReadOnly,
+ default: OptionDefault::None,
+ visibility: OptionVisibility::ReadOnly,
+ critical: false,
+ description: || {
+ "Fingerprints of the key-encryption keys that wrap the stream's
key, \
+ comma-separated: one outside a key rotation, one per configured
server key \
+ during it. Absent when the source is none"
+ .to_owned()
+ },
+ },
+ OptionDescriptor {
+ key: OptionKey::CriticalKeys,
+ scopes: ALL_SCOPES,
+ kind: HeaderKind::String,
+ storage: OptionStorage::Map,
+ mutability: OptionMutability::CreateOnly,
+ default: OptionDefault::None,
+ visibility: OptionVisibility::Internal,
+ critical: false,
+ description: || {
+ "Sorted comma-separated names of stored keys a node must
understand before \
+ serving the resource"
+ .to_owned()
+ },
+ },
+];
+
+/// One option key's catalog entry.
+pub struct OptionDescriptor {
+ pub key: OptionKey,
+ pub scopes: &'static [OptionsScope],
+ /// Canonical kind: what admission stores whatever kind the client sent.
+ pub kind: HeaderKind,
+ pub storage: OptionStorage,
+ pub mutability: OptionMutability,
+ pub default: OptionDefault,
+ pub visibility: OptionVisibility,
+ /// A node that does not understand the key must not serve the resource.
+ pub critical: bool,
+ pub description: fn() -> String,
+}
+
+/// Where a key's value lives.
+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+pub enum OptionStorage {
+ /// In the resource's options map.
+ Map,
+ /// In a typed, indexed field of the resource, such as its name.
+ Column,
+ /// Nowhere: computed when the resource is read.
+ Computed,
+}
+
+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+pub enum OptionMutability {
+ CreateOnly,
+ Updatable,
+ ReadOnly,
+}
+
+#[derive(Clone, Copy)]
+pub enum OptionDefault {
+ None,
+ Const(fn() -> Result<HeaderValue, IggyError>),
+}
+
+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+pub enum OptionVisibility {
+ /// Accepted from clients and returned on reads.
+ Public,
+ /// Accepted from clients, never returned.
+ Secret,
+ /// Written by the server, never accepted from clients, never returned.
+ Internal,
+ /// Computed by the server on reads, never accepted from clients.
+ ReadOnly,
+}
+
+/// Which command an options block arrived on.
+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+pub enum OptionOperation {
+ Create,
+ Update,
+}
+
+/// Who chose a stream's at-rest key, as reported on reads.
+#[derive(
+ Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize, Display,
EnumString, IntoStaticStr,
+)]
+#[serde(rename_all = "snake_case")]
+#[strum(serialize_all = "snake_case")]
+pub enum StreamEncryptionSource {
+ /// The client sent the key.
+ Stream,
+ /// The server generated the key.
+ Server,
+ /// Stored unencrypted.
+ None,
+}
+
+/// The fixed columns a create carries or an update patches.
+#[derive(Debug, Clone, Default, PartialEq)]
+pub struct OptionColumns {
+ pub name: Option<String>,
+ pub username: Option<String>,
+ pub status: Option<UserStatus>,
+}
+
+/// What admission may know about the resource beyond the request itself.
+#[derive(Debug, Clone, Default)]
+pub struct OptionAdmissionContext {
+ /// The resource's stored options, for an update's capacity check.
+ pub current: Option<ResourceOptions>,
+ /// The create command's own fixed fields.
+ pub create_columns: OptionColumns,
+}
+
+/// An options block that passed admission, in canonical kinds.
+#[derive(Clone, Default, PartialEq)]
+pub struct AdmittedOptions {
+ /// Map keys to set, explicit and canonical. Zero sentinels that mean "use
+ /// the default" are dropped, so they leave the stored value alone.
+ pub explicit: ResourceOptions,
+ /// Map keys to reset to their defaults.
+ pub reset: Vec<OptionKey>,
+ /// Column values the update sets, or the create's validated columns.
+ pub columns: OptionColumns,
+}
+
+impl Debug for AdmittedOptions {
+ /// Prints secret values as `<redacted>`.
+ fn fmt(&self, formatter: &mut Formatter<'_>) -> std::fmt::Result {
+ let explicit: BTreeMap<String, &dyn Debug> = self
+ .explicit
+ .iter()
+ .map(|(key, option)| {
+ let secret = OptionKey::try_from(key)
+ .is_ok_and(|known| known.descriptor().visibility !=
OptionVisibility::Public);
+ let shown: &dyn Debug = if secret { &"<redacted>" } else {
&option.value };
+ (String::from_utf8_lossy(key.as_bytes()).into_owned(), shown)
+ })
+ .collect();
+ formatter
+ .debug_struct("AdmittedOptions")
+ .field("explicit", &explicit)
+ .field("reset", &self.reset)
+ .field("columns", &self.columns)
+ .finish()
+ }
+}
+
+impl From<StreamKeyOrigin> for StreamEncryptionSource {
+ fn from(origin: StreamKeyOrigin) -> Self {
+ match origin {
+ StreamKeyOrigin::Client => Self::Stream,
+ StreamKeyOrigin::Server => Self::Server,
+ }
+ }
+}
+
+impl OptionKey {
+ /// This key's catalog entry.
+ #[must_use]
+ pub fn descriptor(self) -> &'static OptionDescriptor {
+ &CATALOG[self as usize]
+ }
+}
+
+impl OptionDescriptor {
+ #[must_use]
+ pub fn applies_to(&self, scope: OptionsScope) -> bool {
+ self.scopes.contains(&scope)
+ }
+
+ #[must_use]
+ pub fn is_updatable(&self) -> bool {
+ self.mutability == OptionMutability::Updatable
+ }
+
+ /// An update may reset only a mutable map key with a default. A default
+ /// alone does not make a create-only setting resettable.
+ #[must_use]
+ pub fn is_resettable(&self) -> bool {
+ self.is_updatable()
+ && self.storage == OptionStorage::Map
+ && matches!(self.default, OptionDefault::Const(_))
+ }
+
+ /// Whether `DescribeOptions` lists this key. Internal keys are the
+ /// server's bookkeeping and stay out of the catalog clients see.
+ #[must_use]
+ pub fn is_described(&self) -> bool {
+ self.visibility != OptionVisibility::Internal
+ }
+
+ /// The `DescribeOptions` flag byte.
+ #[must_use]
+ pub fn flags(&self) -> u8 {
+ let mut flags = 0;
+ if self.is_updatable() {
+ flags |= OPTION_FLAG_UPDATABLE;
+ }
+ if self.is_resettable() {
+ flags |= OPTION_FLAG_RESETTABLE;
+ }
+ if self.visibility == OptionVisibility::Secret {
+ flags |= OPTION_FLAG_SECRET;
+ }
+ if self.visibility == OptionVisibility::ReadOnly {
+ flags |= OPTION_FLAG_READ_ONLY;
+ }
+ if self.critical {
+ flags |= OPTION_FLAG_CRITICAL;
+ }
+ flags
+ }
+
+ /// The key's default in its canonical kind, or `None` without one.
+ ///
+ /// # Errors
+ ///
+ /// Propagates a default that does not fit a header value, which the
+ /// catalog tests rule out.
+ pub fn default_value(&self) -> Result<Option<HeaderValue>, IggyError> {
+ match self.default {
+ OptionDefault::None => Ok(None),
+ OptionDefault::Const(default) => default().map(Some),
+ }
+ }
+}
+
+/// Catalog entries valid in `scope`, in catalog order.
+pub fn descriptors_for(scope: OptionsScope) -> impl Iterator<Item = &'static
OptionDescriptor> {
+ CATALOG
+ .iter()
+ .filter(move |descriptor| descriptor.applies_to(scope))
+}
+
+/// The catalog default of every map key in `scope` that `explicit` does not
+/// set, as derived entries, so a create stores a concrete value for each.
+///
+/// # Errors
+///
+/// Propagates a default that does not fit a header value, which the catalog
+/// tests rule out.
+pub fn derived_defaults(
+ scope: OptionsScope,
+ explicit: &ResourceOptions,
+) -> Result<ResourceOptions, IggyError> {
+ let mut derived = ResourceOptions::new();
+ for descriptor in
+ descriptors_for(scope).filter(|descriptor| descriptor.storage ==
OptionStorage::Map)
+ {
+ let key = HeaderKey::from(descriptor.key);
+ if explicit.contains_key(&key) {
+ continue;
+ }
+ if let Some(default) = descriptor.default_value()? {
+ derived.insert(key, OptionValue::derived(default));
+ }
+ }
+ Ok(derived)
+}
+
+/// Validate and canonicalize a create or update options block.
+///
+/// Checks run in a fixed order so every transport and SDK observes the same
+/// error category for the same request:
+///
+/// 1. `InvalidCommand`: a name repeated across set and reset, a repeated reset
+/// name, an update with nothing to change, a reset on a create.
+/// 2. `UnsupportedOptionKey`: a key outside the catalog or outside `scope`.
+/// 3. `ReadOnlyOption`: an internal or read-only key.
+/// 4. `OptionNotUpdatable`: a create-only key in an update's set or reset.
+/// 5. `OptionNotResettable`: a reset of a column or of a key without default.
+/// 6. Value errors: a column inside a create block is `InvalidOptionValue`,
+/// column values keep their own errors, other keys `InvalidOptionValue`.
+/// 7. `OptionsBlockTooLarge`: the resulting map exceeds the block limits.
+///
+/// # Errors
+///
+/// As listed above.
+pub fn admit(
+ scope: OptionsScope,
+ operation: OptionOperation,
+ block: &WireOptions,
+ reset: &[String],
+ context: &OptionAdmissionContext,
+) -> Result<AdmittedOptions, IggyError> {
+ let mut reset_names = BTreeSet::new();
+ for name in reset {
+ if !reset_names.insert(name.as_str()) {
+ return Err(IggyError::InvalidCommand);
+ }
+ }
+ let entries: Vec<WireUserHeaderEntry<'_>> = block.into_iter().collect();
+ let mut set_names = Vec::with_capacity(entries.len());
+ for entry in &entries {
+ // Wire validation already enforced UTF-8 string keys.
+ let name = std::str::from_utf8(entry.key).map_err(|_|
IggyError::InvalidCommand)?;
+ if reset_names.contains(name) {
+ return Err(IggyError::InvalidCommand);
+ }
+ set_names.push(name);
+ }
+ match operation {
+ OptionOperation::Create if !reset.is_empty() => return
Err(IggyError::InvalidCommand),
+ OptionOperation::Update if entries.is_empty() && reset.is_empty() => {
+ return Err(IggyError::InvalidCommand);
+ }
+ _ => {}
+ }
+
+ let set_keys = set_names
+ .iter()
+ .map(|name| scoped_key(scope, name))
+ .collect::<Result<Vec<_>, _>>()?;
+ let reset_keys = reset
+ .iter()
+ .map(|name| scoped_key(scope, name))
+ .collect::<Result<Vec<_>, _>>()?;
+ let all_keys = || set_keys.iter().chain(reset_keys.iter()).copied();
+
+ if let Some(key) = all_keys().find(|key| {
+ matches!(
+ key.descriptor().visibility,
+ OptionVisibility::Internal | OptionVisibility::ReadOnly
+ )
+ }) {
+ return Err(IggyError::ReadOnlyOption(key.to_string()));
+ }
+ if operation == OptionOperation::Update
+ && let Some(key) = all_keys().find(|key|
!key.descriptor().is_updatable())
+ {
+ return Err(IggyError::OptionNotUpdatable(key.to_string()));
+ }
+ if let Some(key) = reset_keys
+ .iter()
+ .find(|key| !key.descriptor().is_resettable())
+ {
+ return Err(IggyError::OptionNotResettable(key.to_string()));
+ }
+ if operation == OptionOperation::Create
+ && let Some(key) = set_keys
+ .iter()
+ .find(|key| key.descriptor().storage == OptionStorage::Column)
+ {
+ return Err(invalid(*key));
+ }
+
+ let mut columns = match operation {
+ OptionOperation::Create => validated_create_columns(scope,
&context.create_columns)?,
+ OptionOperation::Update => OptionColumns::default(),
+ };
+ let mut explicit = ResourceOptions::new();
+ for (key, entry) in set_keys.iter().zip(&entries) {
+ match key.descriptor().storage {
+ OptionStorage::Column => decode_column(scope, *key, entry, &mut
columns)?,
+ OptionStorage::Map => {
+ if let Some(value) = canonical_value(*key, entry)? {
+ explicit.insert(HeaderKey::from(*key),
OptionValue::explicit(value));
+ }
+ }
+ OptionStorage::Computed => return
Err(IggyError::ReadOnlyOption(key.to_string())),
+ }
+ }
+
+ let mut candidate = context.current.clone().unwrap_or_default();
+ for key in &reset_keys {
+ if let Some(default) = key.descriptor().default_value()? {
+ candidate.insert(HeaderKey::from(*key),
OptionValue::derived(default));
+ }
+ }
+ candidate.extend(explicit.clone());
+ crate::wire_conversions::resource_options_to_wire(&candidate,
OptionsProvenance::All)?;
+
+ Ok(AdmittedOptions {
+ explicit,
+ reset: reset_keys,
+ columns,
+ })
+}
+
+/// The canonical value of one map key, or `None` for a zero sentinel that
+/// means "use the default" and therefore sets nothing.
+///
+/// # Errors
+///
+/// `InvalidOptionValue` naming `key` when the value does not decode or
+/// breaks a per-key bound. `InvalidEncryptionKey` for a malformed stream key.
+pub fn canonical_value(
+ key: OptionKey,
+ entry: &WireUserHeaderEntry<'_>,
+) -> Result<Option<HeaderValue>, IggyError> {
+ let value = match key {
+ OptionKey::CompressionAlgorithm => CompressionAlgorithm::decode(key,
entry)?.encode()?,
+ OptionKey::MessageExpiry => {
+ let expiry = IggyExpiry::decode(key, entry)?;
+ if expiry == IggyExpiry::ServerDefault {
+ return Ok(None);
+ }
+ expiry.encode()?
+ }
+ OptionKey::MaxTopicSize => {
+ let size = MaxTopicSize::decode(key, entry)?;
+ if size == MaxTopicSize::ServerDefault {
+ return Ok(None);
+ }
+ size.encode()?
+ }
+ OptionKey::SegmentSize | OptionKey::SizeOfMessagesRequiredToSave => {
+ let size = IggyByteSize::decode(key, entry)?;
+ if size.as_bytes_u64() == 0 {
+ return Ok(None);
+ }
+ size.encode()?
+ }
+ OptionKey::Durability | OptionKey::ConsumerOffsetDurability => {
+ Durability::decode(key, entry)?.encode()?
+ }
+ OptionKey::PartitionResizePolicy => PartitionResizePolicy::decode(key,
entry)?.encode()?,
+ OptionKey::MessagesRequiredToSave => u32::decode(key,
entry)?.encode()?,
+ OptionKey::PreallocateSegments => bool::decode(key, entry)?.encode()?,
+ OptionKey::EncryptionKey => {
+
HeaderValue::try_from(stream_key_from_entry(entry)?.as_bytes().as_slice())?
+ }
+ OptionKey::Name
+ | OptionKey::Username
+ | OptionKey::Status
+ | OptionKey::EncryptionKeyWrapped
+ | OptionKey::EncryptionKeySource
+ | OptionKey::EncryptionKeyFingerprint
+ | OptionKey::EncryptionKekFingerprint
+ | OptionKey::CriticalKeys => return Err(invalid(key)),
+ };
+ Ok(Some(value))
+}
+
+/// The stream key a create block carries: 32 raw bytes, or their canonical
+/// standard padded base64 as a string.
+///
+/// # Errors
+///
+/// `InvalidEncryptionKey` for any other kind, length or encoding.
+pub fn stream_key_from_entry(
+ entry: &WireUserHeaderEntry<'_>,
+) -> Result<StreamEncryptionKey, IggyError> {
+ match HeaderKind::from_code(entry.value_kind.0) {
+ Ok(HeaderKind::Raw) => StreamEncryptionKey::try_from(entry.value),
+ Ok(HeaderKind::String) => std::str::from_utf8(entry.value)
+ .map_err(|_| IggyError::InvalidEncryptionKey)?
+ .parse(),
+ _ => Err(IggyError::InvalidEncryptionKey),
+ }
+}
+
+/// The options a read returns: public map keys by provenance, unknown keys
+/// this build stores without understanding, and the stream's encryption
+/// projections as derived entries.
+///
+/// The projections come from the stored wrap set alone, so every node answers
+/// the same for the same stream, and a cached create reply stays true.
Review Comment:
Fixed. Dropped the claim. The doc
(`core/common/src/types/options/catalog.rs:822`) and the `CreateStream` apply
comment (`core/metadata/src/stm/stream.rs:2120`) now say a cached create reply
keeps the create-time projections, so its KEK fingerprints predate any later
rewrap.
##########
core/common/src/types/options/catalog.rs:
##########
@@ -0,0 +1,1659 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you 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.
+
+use super::codec::{OptionValueCodec, invalid};
+use super::key::OptionKey;
+use super::{
+ DEFAULT_MAX_TOPIC_SIZE, DEFAULT_MESSAGE_EXPIRY,
DEFAULT_MESSAGES_REQUIRED_TO_SAVE,
+ DEFAULT_PREALLOCATE_SEGMENTS, DEFAULT_SEGMENT_SIZE,
DEFAULT_SIZE_OF_MESSAGES_REQUIRED_TO_SAVE,
+ Durability, MAX_MESSAGES_REQUIRED_TO_SAVE, MAX_PREALLOCATED_TOPIC_BYTES,
+ MAX_SIZE_OF_MESSAGES_REQUIRED_TO_SAVE, MAX_TOPIC_SEGMENT_SIZE,
MIN_TOPIC_SEGMENT_SIZE,
+ OptionValue, OptionsProvenance, OptionsScope, PartitionResizePolicy,
ResourceOptions,
+};
+use crate::http::users::defaults::{MAX_USERNAME_LENGTH, MIN_USERNAME_LENGTH};
+use crate::types::compression::compression_algorithm::CompressionAlgorithm;
+use crate::types::message::{HeaderKey, HeaderKind, HeaderValue};
+use crate::utils::stream_key::{StreamEncryptionKey, StreamKeyOrigin, WrapSet};
+use crate::{IggyByteSize, IggyError, IggyExpiry, MaxTopicSize, UserStatus};
+use iggy_binary_protocol::{WireHeaderKind, WireOptions, WireUserHeaderEntry};
+use serde::{Deserialize, Serialize};
+use std::collections::{BTreeMap, BTreeSet};
+use std::fmt::{Debug, Formatter};
+use strum::{Display, EnumString, IntoStaticStr};
+
+/// Bit set in a `DescribeOptions` entry's flags when an update may set the
key.
+pub const OPTION_FLAG_UPDATABLE: u8 = 1;
+/// Bit set when an update may reset the key to its default.
+pub const OPTION_FLAG_RESETTABLE: u8 = 1 << 1;
+/// Bit set when the key is write-only: accepted, never returned.
+pub const OPTION_FLAG_SECRET: u8 = 1 << 2;
+/// Bit set when the server computes the key and refuses it from clients.
+pub const OPTION_FLAG_READ_ONLY: u8 = 1 << 3;
+/// Bit set when a node that does not understand the key must not serve the
+/// resource.
+pub const OPTION_FLAG_CRITICAL: u8 = 1 << 4;
+
+/// Longest marker a resource may carry. Every critical key name fits many
+/// times over, and the bound keeps the marker a single header field.
+const CRITICAL_KEYS_MAX_LENGTH: usize = 255;
+/// Longest key echoed back in an option error. A rejected key is
+/// attacker-sized, and over HTTP the error text rides the response body.
+const ERROR_KEY_PREVIEW_LENGTH: usize = 64;
+const MAX_RESOURCE_NAME_LENGTH: usize = 255;
+
+const ALL_SCOPES: &[OptionsScope] = &[
+ OptionsScope::Stream,
+ OptionsScope::Topic,
+ OptionsScope::User,
+];
+const STREAM_AND_TOPIC: &[OptionsScope] = &[OptionsScope::Stream,
OptionsScope::Topic];
+const STREAM_ONLY: &[OptionsScope] = &[OptionsScope::Stream];
+const TOPIC_ONLY: &[OptionsScope] = &[OptionsScope::Topic];
+const USER_ONLY: &[OptionsScope] = &[OptionsScope::User];
+
+/// Every option key, indexed by its `OptionKey` discriminant.
+pub static CATALOG: [OptionDescriptor; 19] = [
+ OptionDescriptor {
+ key: OptionKey::Name,
+ scopes: STREAM_AND_TOPIC,
+ kind: HeaderKind::String,
+ storage: OptionStorage::Column,
+ mutability: OptionMutability::Updatable,
+ default: OptionDefault::None,
+ visibility: OptionVisibility::Public,
+ critical: false,
+ description: || {
+ "Resource name, 1 to 255 bytes, unique within its parent. Set by
the create \
+ command's name field and changed through an update."
+ .to_owned()
+ },
+ },
+ OptionDescriptor {
+ key: OptionKey::Username,
+ scopes: USER_ONLY,
+ kind: HeaderKind::String,
+ storage: OptionStorage::Column,
+ mutability: OptionMutability::Updatable,
+ default: OptionDefault::None,
+ visibility: OptionVisibility::Public,
+ critical: false,
+ description: || {
+ format!(
+ "User name, {MIN_USERNAME_LENGTH} to {MAX_USERNAME_LENGTH}
bytes, unique. Set by \
+ the create command's username field and changed through an
update."
+ )
+ },
+ },
+ OptionDescriptor {
+ key: OptionKey::Status,
+ scopes: USER_ONLY,
+ kind: HeaderKind::String,
+ storage: OptionStorage::Column,
+ mutability: OptionMutability::Updatable,
+ default: OptionDefault::None,
+ visibility: OptionVisibility::Public,
+ critical: false,
+ description: || {
+ "User status, active or inactive. Set by the create command's
status field and \
+ changed through an update. Deactivating a user ends its sessions."
+ .to_owned()
+ },
+ },
+ OptionDescriptor {
+ key: OptionKey::CompressionAlgorithm,
+ scopes: TOPIC_ONLY,
+ kind: HeaderKind::String,
+ storage: OptionStorage::Map,
+ mutability: OptionMutability::CreateOnly,
+ default: OptionDefault::Const(||
CompressionAlgorithm::default().encode()),
+ visibility: OptionVisibility::Public,
+ critical: false,
+ description: || {
+ "Compression algorithm (none, gzip). Stored and reported, not
applied to messages"
+ .to_owned()
+ },
+ },
+ OptionDescriptor {
+ key: OptionKey::MessageExpiry,
+ scopes: TOPIC_ONLY,
+ kind: HeaderKind::Uint64,
+ storage: OptionStorage::Map,
+ mutability: OptionMutability::Updatable,
+ default: OptionDefault::Const(||
Ok(HeaderValue::from(DEFAULT_MESSAGE_EXPIRY))),
+ visibility: OptionVisibility::Public,
+ critical: false,
+ description: || {
+ "Message expiry in microseconds, or a humantime string (e.g. 7
days)".to_owned()
+ },
+ },
+ OptionDescriptor {
+ key: OptionKey::MaxTopicSize,
+ scopes: TOPIC_ONLY,
+ kind: HeaderKind::Uint64,
+ storage: OptionStorage::Map,
+ mutability: OptionMutability::Updatable,
+ default: OptionDefault::Const(||
Ok(HeaderValue::from(DEFAULT_MAX_TOPIC_SIZE))),
+ visibility: OptionVisibility::Public,
+ critical: false,
+ description: || {
+ "Topic-wide sealed-segment size cap, split across all partitions,
in bytes or a \
+ byte-size string (e.g. 1 GiB); finite values must be at least the
segment size"
+ .to_owned()
+ },
+ },
+ OptionDescriptor {
+ key: OptionKey::SegmentSize,
+ scopes: TOPIC_ONLY,
+ kind: HeaderKind::Uint64,
+ storage: OptionStorage::Map,
+ mutability: OptionMutability::CreateOnly,
+ default: OptionDefault::Const(||
Ok(HeaderValue::from(DEFAULT_SEGMENT_SIZE))),
+ visibility: OptionVisibility::Public,
+ critical: false,
+ description: || {
+ format!(
+ "Segment size in bytes, or a byte-size string (e.g. 128 MiB);
a 512-byte \
+ multiple within
{MIN_TOPIC_SEGMENT_SIZE}..={MAX_TOPIC_SEGMENT_SIZE}"
+ )
+ },
+ },
+ OptionDescriptor {
+ key: OptionKey::Durability,
+ scopes: TOPIC_ONLY,
+ kind: HeaderKind::String,
+ storage: OptionStorage::Map,
+ mutability: OptionMutability::CreateOnly,
+ default: OptionDefault::Const(|| Durability::default().encode()),
+ visibility: OptionVisibility::Public,
+ critical: false,
+ description: || {
+ "Message completion: replicated or persisted. Independently
defaults to \
+ replicated. A singleton quorum has one copy. In replicated
groups, persisted \
+ messages use WAL references to segment bodies, retaining their
inodes by hard \
+ link until WAL reclamation. The full body size still counts
against \
+ partition.wal_bytes_max."
+ .to_owned()
+ },
+ },
+ OptionDescriptor {
+ key: OptionKey::ConsumerOffsetDurability,
+ scopes: TOPIC_ONLY,
+ kind: HeaderKind::String,
+ storage: OptionStorage::Map,
+ mutability: OptionMutability::CreateOnly,
+ default: OptionDefault::Const(|| Durability::default().encode()),
+ visibility: OptionVisibility::Public,
+ critical: false,
+ description: || {
+ "Explicit offset completion: replicated or persisted.
Independently defaults to \
+ replicated. Poll auto-commit remains asynchronous. In replicated
groups, persisted \
+ offsets also enable WAL references to segment bodies, retaining
their inodes by \
+ hard link until reclamation, even with replicated message
durability. Full body \
+ sizes count against partition.wal_bytes_max."
+ .to_owned()
+ },
+ },
+ OptionDescriptor {
+ key: OptionKey::PartitionResizePolicy,
+ scopes: TOPIC_ONLY,
+ kind: HeaderKind::String,
+ storage: OptionStorage::Map,
+ mutability: OptionMutability::CreateOnly,
+ default: OptionDefault::Const(||
PartitionResizePolicy::default().encode()),
+ visibility: OptionVisibility::Public,
+ critical: false,
+ description: || {
+ "Immutable after creation: mutable permits partition
addition/deletion without a \
+ bounded cutover or per-key ordering across layouts; fixed rejects
both. Retention \
+ and topic deletion remain destructive."
+ .to_owned()
+ },
+ },
+ OptionDescriptor {
+ key: OptionKey::MessagesRequiredToSave,
+ scopes: TOPIC_ONLY,
+ kind: HeaderKind::Uint32,
+ storage: OptionStorage::Map,
+ mutability: OptionMutability::CreateOnly,
+ default: OptionDefault::Const(||
Ok(HeaderValue::from(DEFAULT_MESSAGES_REQUIRED_TO_SAVE))),
+ visibility: OptionVisibility::Public,
+ critical: false,
+ description: || {
+ format!(
+ "Ordinary message-count flush trigger;
1..={MAX_MESSAGES_REQUIRED_TO_SAVE}. \
+ Required persistence, capacity pressure, and lifecycle
operations can flush \
+ earlier"
+ )
+ },
+ },
+ OptionDescriptor {
+ key: OptionKey::SizeOfMessagesRequiredToSave,
+ scopes: TOPIC_ONLY,
+ kind: HeaderKind::Uint64,
+ storage: OptionStorage::Map,
+ mutability: OptionMutability::CreateOnly,
+ default: OptionDefault::Const(|| {
+ Ok(HeaderValue::from(DEFAULT_SIZE_OF_MESSAGES_REQUIRED_TO_SAVE))
+ }),
+ visibility: OptionVisibility::Public,
+ critical: false,
+ description: || {
+ format!(
+ "Flush the journal once it holds this many bytes, or a
byte-size string; \
+ whichever threshold trips first flushes. At most \
+ {MAX_SIZE_OF_MESSAGES_REQUIRED_TO_SAVE}"
+ )
+ },
+ },
+ OptionDescriptor {
+ key: OptionKey::PreallocateSegments,
+ scopes: TOPIC_ONLY,
+ kind: HeaderKind::Bool,
+ storage: OptionStorage::Map,
+ mutability: OptionMutability::CreateOnly,
+ default: OptionDefault::Const(||
Ok(HeaderValue::from(DEFAULT_PREALLOCATE_SEGMENTS))),
+ visibility: OptionVisibility::Public,
+ critical: false,
+ description: || {
+ format!(
+ "Request segment_size bytes of disk reservation when each
segment opens. \
+ Unsupported or failed reservations fall back to ordinary
allocation with a \
+ warning. Reservation runs inline on the owning shard.
segment_size * \
+ partitions_count is capped at {MAX_PREALLOCATED_TOPIC_BYTES}
bytes per create"
+ )
+ },
+ },
+ OptionDescriptor {
+ key: OptionKey::EncryptionKey,
+ scopes: STREAM_ONLY,
+ kind: HeaderKind::Raw,
+ storage: OptionStorage::Map,
+ mutability: OptionMutability::CreateOnly,
+ default: OptionDefault::None,
+ visibility: OptionVisibility::Secret,
+ critical: true,
+ description: || {
+ "The stream's own data encryption key: 32 random bytes, raw or
standard padded \
+ base64. Create-only and write-only: the server stores it wrapped
under a key \
+ derived from the server encryption key and never returns it.
Requires the server \
+ encryption key to be configured. Without it, a server with
encryption enabled \
+ generates a key for the stream. The key travels in the create
request over the \
+ transport the client uses."
+ .to_owned()
+ },
+ },
+ OptionDescriptor {
+ key: OptionKey::EncryptionKeyWrapped,
+ scopes: STREAM_ONLY,
+ kind: HeaderKind::Raw,
+ storage: OptionStorage::Map,
+ mutability: OptionMutability::CreateOnly,
+ default: OptionDefault::None,
+ visibility: OptionVisibility::Internal,
+ critical: true,
+ description: || {
+ "The stream key wrapped by the server under one or more
key-encryption keys, \
+ never returned"
+ .to_owned()
+ },
+ },
+ OptionDescriptor {
+ key: OptionKey::EncryptionKeySource,
+ scopes: STREAM_ONLY,
+ kind: HeaderKind::String,
+ storage: OptionStorage::Computed,
+ mutability: OptionMutability::ReadOnly,
+ default: OptionDefault::None,
+ visibility: OptionVisibility::ReadOnly,
+ critical: false,
+ description: || {
+ "Who chose the stream's at-rest key: stream when the client sent
it, server when \
+ the server generated it, none when the stream's messages are
stored unencrypted"
+ .to_owned()
+ },
+ },
+ OptionDescriptor {
+ key: OptionKey::EncryptionKeyFingerprint,
+ scopes: STREAM_ONLY,
+ kind: HeaderKind::String,
+ storage: OptionStorage::Computed,
+ mutability: OptionMutability::ReadOnly,
+ default: OptionDefault::None,
+ visibility: OptionVisibility::ReadOnly,
+ critical: false,
+ description: || {
+ "The first 8 bytes of SHA-256 over the label
iggy/stream-key-fingerprint/v1 and \
+ the stream's key, as 16 lowercase hex characters. Absent when the
source is none"
+ .to_owned()
+ },
+ },
+ OptionDescriptor {
+ key: OptionKey::EncryptionKekFingerprint,
+ scopes: STREAM_ONLY,
+ kind: HeaderKind::String,
+ storage: OptionStorage::Computed,
+ mutability: OptionMutability::ReadOnly,
+ default: OptionDefault::None,
+ visibility: OptionVisibility::ReadOnly,
+ critical: false,
+ description: || {
+ "Fingerprints of the key-encryption keys that wrap the stream's
key, \
+ comma-separated: one outside a key rotation, one per configured
server key \
+ during it. Absent when the source is none"
+ .to_owned()
+ },
+ },
+ OptionDescriptor {
+ key: OptionKey::CriticalKeys,
+ scopes: ALL_SCOPES,
+ kind: HeaderKind::String,
+ storage: OptionStorage::Map,
+ mutability: OptionMutability::CreateOnly,
+ default: OptionDefault::None,
+ visibility: OptionVisibility::Internal,
+ critical: false,
+ description: || {
+ "Sorted comma-separated names of stored keys a node must
understand before \
+ serving the resource"
+ .to_owned()
+ },
+ },
+];
+
+/// One option key's catalog entry.
+pub struct OptionDescriptor {
+ pub key: OptionKey,
+ pub scopes: &'static [OptionsScope],
+ /// Canonical kind: what admission stores whatever kind the client sent.
+ pub kind: HeaderKind,
+ pub storage: OptionStorage,
+ pub mutability: OptionMutability,
+ pub default: OptionDefault,
+ pub visibility: OptionVisibility,
+ /// A node that does not understand the key must not serve the resource.
+ pub critical: bool,
+ pub description: fn() -> String,
+}
+
+/// Where a key's value lives.
+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+pub enum OptionStorage {
+ /// In the resource's options map.
+ Map,
+ /// In a typed, indexed field of the resource, such as its name.
+ Column,
+ /// Nowhere: computed when the resource is read.
+ Computed,
+}
+
+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+pub enum OptionMutability {
+ CreateOnly,
+ Updatable,
+ ReadOnly,
+}
+
+#[derive(Clone, Copy)]
+pub enum OptionDefault {
+ None,
+ Const(fn() -> Result<HeaderValue, IggyError>),
+}
+
+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+pub enum OptionVisibility {
+ /// Accepted from clients and returned on reads.
+ Public,
+ /// Accepted from clients, never returned.
+ Secret,
+ /// Written by the server, never accepted from clients, never returned.
+ Internal,
+ /// Computed by the server on reads, never accepted from clients.
+ ReadOnly,
+}
+
+/// Which command an options block arrived on.
+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+pub enum OptionOperation {
+ Create,
+ Update,
+}
+
+/// Who chose a stream's at-rest key, as reported on reads.
+#[derive(
+ Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize, Display,
EnumString, IntoStaticStr,
+)]
+#[serde(rename_all = "snake_case")]
+#[strum(serialize_all = "snake_case")]
+pub enum StreamEncryptionSource {
+ /// The client sent the key.
+ Stream,
+ /// The server generated the key.
+ Server,
+ /// Stored unencrypted.
+ None,
+}
+
+/// The fixed columns a create carries or an update patches.
+#[derive(Debug, Clone, Default, PartialEq)]
+pub struct OptionColumns {
+ pub name: Option<String>,
+ pub username: Option<String>,
+ pub status: Option<UserStatus>,
+}
+
+/// What admission may know about the resource beyond the request itself.
+#[derive(Debug, Clone, Default)]
+pub struct OptionAdmissionContext {
+ /// The resource's stored options, for an update's capacity check.
+ pub current: Option<ResourceOptions>,
+ /// The create command's own fixed fields.
+ pub create_columns: OptionColumns,
+}
+
+/// An options block that passed admission, in canonical kinds.
+#[derive(Clone, Default, PartialEq)]
+pub struct AdmittedOptions {
+ /// Map keys to set, explicit and canonical. Zero sentinels that mean "use
+ /// the default" are dropped, so they leave the stored value alone.
+ pub explicit: ResourceOptions,
+ /// Map keys to reset to their defaults.
+ pub reset: Vec<OptionKey>,
+ /// Column values the update sets, or the create's validated columns.
+ pub columns: OptionColumns,
+}
+
+impl Debug for AdmittedOptions {
+ /// Prints secret values as `<redacted>`.
+ fn fmt(&self, formatter: &mut Formatter<'_>) -> std::fmt::Result {
+ let explicit: BTreeMap<String, &dyn Debug> = self
+ .explicit
+ .iter()
+ .map(|(key, option)| {
+ let secret = OptionKey::try_from(key)
+ .is_ok_and(|known| known.descriptor().visibility !=
OptionVisibility::Public);
+ let shown: &dyn Debug = if secret { &"<redacted>" } else {
&option.value };
+ (String::from_utf8_lossy(key.as_bytes()).into_owned(), shown)
+ })
+ .collect();
+ formatter
+ .debug_struct("AdmittedOptions")
+ .field("explicit", &explicit)
+ .field("reset", &self.reset)
+ .field("columns", &self.columns)
+ .finish()
+ }
+}
+
+impl From<StreamKeyOrigin> for StreamEncryptionSource {
+ fn from(origin: StreamKeyOrigin) -> Self {
+ match origin {
+ StreamKeyOrigin::Client => Self::Stream,
+ StreamKeyOrigin::Server => Self::Server,
+ }
+ }
+}
+
+impl OptionKey {
+ /// This key's catalog entry.
+ #[must_use]
+ pub fn descriptor(self) -> &'static OptionDescriptor {
+ &CATALOG[self as usize]
+ }
+}
+
+impl OptionDescriptor {
+ #[must_use]
+ pub fn applies_to(&self, scope: OptionsScope) -> bool {
+ self.scopes.contains(&scope)
+ }
+
+ #[must_use]
+ pub fn is_updatable(&self) -> bool {
+ self.mutability == OptionMutability::Updatable
+ }
+
+ /// An update may reset only a mutable map key with a default. A default
+ /// alone does not make a create-only setting resettable.
+ #[must_use]
+ pub fn is_resettable(&self) -> bool {
+ self.is_updatable()
+ && self.storage == OptionStorage::Map
+ && matches!(self.default, OptionDefault::Const(_))
+ }
+
+ /// Whether `DescribeOptions` lists this key. Internal keys are the
+ /// server's bookkeeping and stay out of the catalog clients see.
+ #[must_use]
+ pub fn is_described(&self) -> bool {
+ self.visibility != OptionVisibility::Internal
+ }
+
+ /// The `DescribeOptions` flag byte.
+ #[must_use]
+ pub fn flags(&self) -> u8 {
+ let mut flags = 0;
+ if self.is_updatable() {
+ flags |= OPTION_FLAG_UPDATABLE;
+ }
+ if self.is_resettable() {
+ flags |= OPTION_FLAG_RESETTABLE;
+ }
+ if self.visibility == OptionVisibility::Secret {
+ flags |= OPTION_FLAG_SECRET;
+ }
+ if self.visibility == OptionVisibility::ReadOnly {
+ flags |= OPTION_FLAG_READ_ONLY;
+ }
+ if self.critical {
+ flags |= OPTION_FLAG_CRITICAL;
+ }
+ flags
+ }
+
+ /// The key's default in its canonical kind, or `None` without one.
+ ///
+ /// # Errors
+ ///
+ /// Propagates a default that does not fit a header value, which the
+ /// catalog tests rule out.
+ pub fn default_value(&self) -> Result<Option<HeaderValue>, IggyError> {
+ match self.default {
+ OptionDefault::None => Ok(None),
+ OptionDefault::Const(default) => default().map(Some),
+ }
+ }
+}
+
+/// Catalog entries valid in `scope`, in catalog order.
+pub fn descriptors_for(scope: OptionsScope) -> impl Iterator<Item = &'static
OptionDescriptor> {
+ CATALOG
+ .iter()
+ .filter(move |descriptor| descriptor.applies_to(scope))
+}
+
+/// The catalog default of every map key in `scope` that `explicit` does not
+/// set, as derived entries, so a create stores a concrete value for each.
+///
+/// # Errors
+///
+/// Propagates a default that does not fit a header value, which the catalog
+/// tests rule out.
+pub fn derived_defaults(
+ scope: OptionsScope,
+ explicit: &ResourceOptions,
+) -> Result<ResourceOptions, IggyError> {
+ let mut derived = ResourceOptions::new();
+ for descriptor in
+ descriptors_for(scope).filter(|descriptor| descriptor.storage ==
OptionStorage::Map)
+ {
+ let key = HeaderKey::from(descriptor.key);
+ if explicit.contains_key(&key) {
+ continue;
+ }
+ if let Some(default) = descriptor.default_value()? {
+ derived.insert(key, OptionValue::derived(default));
+ }
+ }
+ Ok(derived)
+}
+
+/// Validate and canonicalize a create or update options block.
+///
+/// Checks run in a fixed order so every transport and SDK observes the same
+/// error category for the same request:
+///
+/// 1. `InvalidCommand`: a name repeated across set and reset, a repeated reset
+/// name, an update with nothing to change, a reset on a create.
+/// 2. `UnsupportedOptionKey`: a key outside the catalog or outside `scope`.
+/// 3. `ReadOnlyOption`: an internal or read-only key.
+/// 4. `OptionNotUpdatable`: a create-only key in an update's set or reset.
+/// 5. `OptionNotResettable`: a reset of a column or of a key without default.
+/// 6. Value errors: a column inside a create block is `InvalidOptionValue`,
+/// column values keep their own errors, other keys `InvalidOptionValue`.
+/// 7. `OptionsBlockTooLarge`: the resulting map exceeds the block limits.
+///
+/// # Errors
+///
+/// As listed above.
+pub fn admit(
+ scope: OptionsScope,
+ operation: OptionOperation,
+ block: &WireOptions,
+ reset: &[String],
+ context: &OptionAdmissionContext,
+) -> Result<AdmittedOptions, IggyError> {
+ let mut reset_names = BTreeSet::new();
+ for name in reset {
+ if !reset_names.insert(name.as_str()) {
+ return Err(IggyError::InvalidCommand);
+ }
+ }
+ let entries: Vec<WireUserHeaderEntry<'_>> = block.into_iter().collect();
+ let mut set_names = Vec::with_capacity(entries.len());
+ for entry in &entries {
+ // Wire validation already enforced UTF-8 string keys.
+ let name = std::str::from_utf8(entry.key).map_err(|_|
IggyError::InvalidCommand)?;
+ if reset_names.contains(name) {
+ return Err(IggyError::InvalidCommand);
+ }
+ set_names.push(name);
+ }
+ match operation {
+ OptionOperation::Create if !reset.is_empty() => return
Err(IggyError::InvalidCommand),
+ OptionOperation::Update if entries.is_empty() && reset.is_empty() => {
+ return Err(IggyError::InvalidCommand);
+ }
+ _ => {}
+ }
+
+ let set_keys = set_names
+ .iter()
+ .map(|name| scoped_key(scope, name))
+ .collect::<Result<Vec<_>, _>>()?;
+ let reset_keys = reset
+ .iter()
+ .map(|name| scoped_key(scope, name))
+ .collect::<Result<Vec<_>, _>>()?;
+ let all_keys = || set_keys.iter().chain(reset_keys.iter()).copied();
+
+ if let Some(key) = all_keys().find(|key| {
+ matches!(
+ key.descriptor().visibility,
+ OptionVisibility::Internal | OptionVisibility::ReadOnly
+ )
+ }) {
+ return Err(IggyError::ReadOnlyOption(key.to_string()));
+ }
+ if operation == OptionOperation::Update
+ && let Some(key) = all_keys().find(|key|
!key.descriptor().is_updatable())
+ {
+ return Err(IggyError::OptionNotUpdatable(key.to_string()));
+ }
+ if let Some(key) = reset_keys
+ .iter()
+ .find(|key| !key.descriptor().is_resettable())
+ {
+ return Err(IggyError::OptionNotResettable(key.to_string()));
+ }
+ if operation == OptionOperation::Create
+ && let Some(key) = set_keys
+ .iter()
+ .find(|key| key.descriptor().storage == OptionStorage::Column)
+ {
+ return Err(invalid(*key));
+ }
+
+ let mut columns = match operation {
+ OptionOperation::Create => validated_create_columns(scope,
&context.create_columns)?,
+ OptionOperation::Update => OptionColumns::default(),
+ };
+ let mut explicit = ResourceOptions::new();
+ for (key, entry) in set_keys.iter().zip(&entries) {
+ match key.descriptor().storage {
+ OptionStorage::Column => decode_column(scope, *key, entry, &mut
columns)?,
+ OptionStorage::Map => {
+ if let Some(value) = canonical_value(*key, entry)? {
+ explicit.insert(HeaderKey::from(*key),
OptionValue::explicit(value));
+ }
+ }
+ OptionStorage::Computed => return
Err(IggyError::ReadOnlyOption(key.to_string())),
+ }
+ }
+
+ let mut candidate = context.current.clone().unwrap_or_default();
+ for key in &reset_keys {
+ if let Some(default) = key.descriptor().default_value()? {
+ candidate.insert(HeaderKey::from(*key),
OptionValue::derived(default));
+ }
+ }
+ candidate.extend(explicit.clone());
+ crate::wire_conversions::resource_options_to_wire(&candidate,
OptionsProvenance::All)?;
+
+ Ok(AdmittedOptions {
+ explicit,
+ reset: reset_keys,
+ columns,
+ })
+}
+
+/// The canonical value of one map key, or `None` for a zero sentinel that
+/// means "use the default" and therefore sets nothing.
+///
+/// # Errors
+///
+/// `InvalidOptionValue` naming `key` when the value does not decode or
+/// breaks a per-key bound. `InvalidEncryptionKey` for a malformed stream key.
+pub fn canonical_value(
+ key: OptionKey,
+ entry: &WireUserHeaderEntry<'_>,
+) -> Result<Option<HeaderValue>, IggyError> {
+ let value = match key {
+ OptionKey::CompressionAlgorithm => CompressionAlgorithm::decode(key,
entry)?.encode()?,
+ OptionKey::MessageExpiry => {
+ let expiry = IggyExpiry::decode(key, entry)?;
+ if expiry == IggyExpiry::ServerDefault {
+ return Ok(None);
+ }
+ expiry.encode()?
+ }
+ OptionKey::MaxTopicSize => {
+ let size = MaxTopicSize::decode(key, entry)?;
+ if size == MaxTopicSize::ServerDefault {
+ return Ok(None);
+ }
+ size.encode()?
+ }
+ OptionKey::SegmentSize | OptionKey::SizeOfMessagesRequiredToSave => {
+ let size = IggyByteSize::decode(key, entry)?;
+ if size.as_bytes_u64() == 0 {
+ return Ok(None);
+ }
+ size.encode()?
+ }
+ OptionKey::Durability | OptionKey::ConsumerOffsetDurability => {
+ Durability::decode(key, entry)?.encode()?
+ }
+ OptionKey::PartitionResizePolicy => PartitionResizePolicy::decode(key,
entry)?.encode()?,
+ OptionKey::MessagesRequiredToSave => u32::decode(key,
entry)?.encode()?,
+ OptionKey::PreallocateSegments => bool::decode(key, entry)?.encode()?,
+ OptionKey::EncryptionKey => {
+
HeaderValue::try_from(stream_key_from_entry(entry)?.as_bytes().as_slice())?
+ }
+ OptionKey::Name
+ | OptionKey::Username
+ | OptionKey::Status
+ | OptionKey::EncryptionKeyWrapped
+ | OptionKey::EncryptionKeySource
+ | OptionKey::EncryptionKeyFingerprint
+ | OptionKey::EncryptionKekFingerprint
+ | OptionKey::CriticalKeys => return Err(invalid(key)),
+ };
+ Ok(Some(value))
+}
+
+/// The stream key a create block carries: 32 raw bytes, or their canonical
+/// standard padded base64 as a string.
+///
+/// # Errors
+///
+/// `InvalidEncryptionKey` for any other kind, length or encoding.
+pub fn stream_key_from_entry(
+ entry: &WireUserHeaderEntry<'_>,
+) -> Result<StreamEncryptionKey, IggyError> {
+ match HeaderKind::from_code(entry.value_kind.0) {
+ Ok(HeaderKind::Raw) => StreamEncryptionKey::try_from(entry.value),
+ Ok(HeaderKind::String) => std::str::from_utf8(entry.value)
+ .map_err(|_| IggyError::InvalidEncryptionKey)?
+ .parse(),
+ _ => Err(IggyError::InvalidEncryptionKey),
+ }
+}
+
+/// The options a read returns: public map keys by provenance, unknown keys
+/// this build stores without understanding, and the stream's encryption
+/// projections as derived entries.
+///
+/// The projections come from the stored wrap set alone, so every node answers
+/// the same for the same stream, and a cached create reply stays true.
+///
+/// # Errors
+///
+/// `UnsupportedCriticalOption` when the resource holds critical state this
+/// build cannot interpret. Its values are never echoed then.
+pub fn public_options(
+ scope: OptionsScope,
+ stored: &ResourceOptions,
+) -> Result<ResourceOptions, IggyError> {
+ check_critical_options(stored)?;
+ let mut public = ResourceOptions::new();
+ for (key, value) in stored {
+ let returned = OptionKey::try_from(key).map_or(true, |known| {
+ let descriptor = known.descriptor();
+ descriptor.visibility == OptionVisibility::Public
+ && descriptor.storage == OptionStorage::Map
+ });
+ if returned {
+ public.insert(key.clone(), value.clone());
+ }
+ }
+ if scope == OptionsScope::Stream {
+ let mut project = |key: OptionKey, text: &str| -> Result<(),
IggyError> {
+ public.insert(
+ HeaderKey::from(key),
+ OptionValue::derived(HeaderValue::try_from(text)?),
+ );
+ Ok(())
+ };
+ match stream_wrap_set(stored)? {
+ Some(set) => {
+ let source: &'static str =
StreamEncryptionSource::from(set.origin()).into();
+ project(OptionKey::EncryptionKeySource, source)?;
+ project(OptionKey::EncryptionKeyFingerprint,
&set.key_fingerprint())?;
+ project(
+ OptionKey::EncryptionKekFingerprint,
+ &set.kek_fingerprints().join(","),
+ )?;
+ }
+ None => {
+ let source: &'static str = StreamEncryptionSource::None.into();
+ project(OptionKey::EncryptionKeySource, source)?;
+ }
+ }
+ }
+ Ok(public)
+}
+
+/// The stream's stored wrap set, if the stream is encrypted.
+///
+/// # Errors
+///
+/// `UnsupportedCriticalOption` for a set this build cannot read.
+pub fn stream_wrap_set(stored: &ResourceOptions) -> Result<Option<WrapSet>,
IggyError> {
+ stored
+ .get(&HeaderKey::from(OptionKey::EncryptionKeyWrapped))
+ .map(|option| WrapSet::try_from(option.value.as_bytes()))
+ .transpose()
+}
+
+/// The canonical `critical_keys` marker naming `keys`.
+///
+/// # Errors
+///
+/// `InvalidOptionValue` when the list is empty or does not fit the marker.
+pub fn critical_keys_marker(keys: &[OptionKey]) -> Result<HeaderValue,
IggyError> {
+ let names: BTreeSet<&'static str> = keys.iter().map(|key|
(*key).into()).collect();
+ let marker = names.into_iter().collect::<Vec<_>>().join(",");
+ if marker.is_empty() || marker.len() > CRITICAL_KEYS_MAX_LENGTH {
+ return Err(invalid(OptionKey::CriticalKeys));
+ }
+ HeaderValue::try_from(marker.as_str())
+}
+
+/// Refuse stored state this build must understand but cannot.
+///
+/// A malformed marker, a listed key this build does not know as critical, a
+/// listed key missing from the map, and a wrapped key in an unknown envelope
+/// version all fail. A wrapped key is critical even when no marker names it.
+///
+/// # Errors
+///
+/// `UnsupportedCriticalOption` naming the offending key.
+pub fn check_critical_options(stored: &ResourceOptions) -> Result<(),
IggyError> {
+ if let Some(marker) =
stored.get(&HeaderKey::from(OptionKey::CriticalKeys)) {
+ let names = parse_critical_marker(&marker.value)?;
+ for name in names {
+ let known = name
+ .parse::<OptionKey>()
+ .ok()
+ .filter(|key| key.descriptor().critical)
+ .ok_or_else(||
IggyError::UnsupportedCriticalOption(key_preview(name)))?;
+ if !stored.contains_key(&HeaderKey::from(known)) {
+ return
Err(IggyError::UnsupportedCriticalOption(name.to_owned()));
+ }
+ }
+ }
+ stream_wrap_set(stored).map(|_| ())
+}
+
+/// A string key in `scope`, or `UnsupportedOptionKey`.
+fn scoped_key(scope: OptionsScope, name: &str) -> Result<OptionKey, IggyError>
{
+ name.parse::<OptionKey>()
+ .ok()
+ .filter(|key| key.descriptor().applies_to(scope))
+ .ok_or_else(|| IggyError::UnsupportedOptionKey(key_preview(name)))
+}
+
+fn validated_create_columns(
+ scope: OptionsScope,
+ columns: &OptionColumns,
+) -> Result<OptionColumns, IggyError> {
+ if let Some(name) = &columns.name {
+ validate_name(scope, name)?;
+ }
+ if let Some(username) = &columns.username {
+ validate_username(username)?;
+ }
+ Ok(columns.clone())
+}
+
+fn decode_column(
+ scope: OptionsScope,
+ key: OptionKey,
+ entry: &WireUserHeaderEntry<'_>,
+ columns: &mut OptionColumns,
+) -> Result<(), IggyError> {
+ let text = (entry.value_kind ==
WireHeaderKind(HeaderKind::String.as_code()))
+ .then(|| std::str::from_utf8(entry.value).ok())
+ .flatten();
+ match key {
+ OptionKey::Name => {
+ let name = text.ok_or_else(|| name_error(scope))?;
+ validate_name(scope, name)?;
+ columns.name = Some(name.to_owned());
+ }
+ OptionKey::Username => {
+ let username = text.ok_or(IggyError::InvalidUsername)?;
+ validate_username(username)?;
+ columns.username = Some(username.to_owned());
+ }
+ OptionKey::Status => {
+ columns.status = Some(
+ text.ok_or(IggyError::InvalidUserStatus)?
+ .parse()
+ .map_err(|_| IggyError::InvalidUserStatus)?,
+ );
+ }
+ _ => return Err(invalid(key)),
+ }
+ Ok(())
+}
+
+fn validate_name(scope: OptionsScope, name: &str) -> Result<(), IggyError> {
+ if name.is_empty() || name.len() > MAX_RESOURCE_NAME_LENGTH {
+ return Err(name_error(scope));
+ }
+ Ok(())
+}
+
+fn validate_username(username: &str) -> Result<(), IggyError> {
+ if !(MIN_USERNAME_LENGTH..=MAX_USERNAME_LENGTH).contains(&username.len()) {
+ return Err(IggyError::InvalidUsername);
+ }
+ Ok(())
+}
+
+const fn name_error(scope: OptionsScope) -> IggyError {
+ match scope {
+ OptionsScope::Topic => IggyError::InvalidTopicName,
+ OptionsScope::Stream | OptionsScope::User =>
IggyError::InvalidStreamName,
+ }
+}
+
+fn parse_critical_marker(value: &HeaderValue) -> Result<Vec<&str>, IggyError> {
+ let malformed = ||
IggyError::UnsupportedCriticalOption(OptionKey::CriticalKeys.to_string());
+ if value.kind() != HeaderKind::String || value.as_bytes().len() >
CRITICAL_KEYS_MAX_LENGTH {
+ return Err(malformed());
+ }
+ let text = std::str::from_utf8(value.as_bytes()).map_err(|_| malformed())?;
+ let names: Vec<&str> = text.split(',').collect();
+ let well_formed = names.iter().all(|name| {
+ !name.is_empty()
+ && name
+ .bytes()
+ .all(|byte| byte.is_ascii_lowercase() || byte.is_ascii_digit()
|| byte == b'_')
+ });
+ if !well_formed || names.windows(2).any(|pair| pair[0] >= pair[1]) {
+ return Err(malformed());
+ }
+ Ok(names)
+}
+
+fn key_preview(key: &str) -> String {
+ key.chars().take(ERROR_KEY_PREVIEW_LENGTH).collect()
+}
+
+#[cfg(test)]
+mod tests {
+ use super::super::topic_option_keys;
+ use super::*;
+ use crate::wire_conversions::resource_options_to_wire;
+ use std::str::FromStr;
+ use strum::IntoEnumIterator;
+
+ #[test]
+ fn
given_catalog_when_listing_keys_then_should_never_name_the_reset_field() {
Review Comment:
Fixed across the PR. Every test the PR adds, in Rust and in the foreign
SDKs, is renamed to `given_X_when_Y_should_Z`, and the new BDD scenario titles
no longer derive `then_should` names. This one is now
`given_catalog_when_listing_keys_should_never_name_the_reset_field`
(`core/common/src/types/options/catalog.rs:1065`).
##########
core/metadata/src/stm/stream.rs:
##########
@@ -2100,6 +2101,11 @@ impl StateHandler for CreateStreamRequest {
let Ok(options) = resource_options_from_wire(&self.options, true) else
{
return ApplyReply::err(CreateStreamResult::InvalidOptionValue);
};
+ // The edge replaces a client key with its wrapped form. A plaintext
key
+ // reaching apply means that step was skipped, and it is never stored.
+ if options.contains_key(&HeaderKey::from(OptionKey::EncryptionKey)) {
Review Comment:
Fixed. `prepare_request` now refuses a secret catalog key in a
`CreateStream` before the prepare is sealed, so a plaintext key never reaches
the journal (`core/metadata/src/impls/metadata.rs:4384`). The apply refusal
stays for WAL replay.
`given_plaintext_stream_key_when_preparing_create_stream_should_refuse_before_replication`
pins it.
##########
core/metadata/src/stm/stream.rs:
##########
@@ -2434,63 +2461,145 @@ impl StateHandler for UpdateTopicRequest {
if state.lifecycle_blocks(stream_id, Some(topic_id)) {
return
ApplyReply::err(iggy_common::IggyError::LifecycleBusy.as_code());
}
-
let Some(stream) = state.items.get_mut(stream_id) else {
return ApplyReply::err(UpdateTopicResult::StreamNotFound);
};
- let Some(topic) = stream.topics.get_mut(topic_id) else {
+ let Some(topic) = stream.topics.get(topic_id) else {
return ApplyReply::err(UpdateTopicResult::TopicNotFound);
};
-
- let new_name_arc: Arc<str> = Arc::from(self.name.as_str());
- if let Some(&existing_id) = stream.topic_index.get(&new_name_arc)
+ // Every check runs before any mutation, so a refused patch leaves the
+ // topic exactly as it was: no half-applied rename or option change.
+ // The blocks are read leniently, a key this build does not know is
+ // stored rather than refused, so the verdict never depends on the
+ // build applying it.
+ let new_name: Option<Arc<str>> = self.name.as_ref().map(|name|
Arc::from(name.as_str()));
+ if let Some(name) = &new_name
+ && let Some(&existing_id) = stream.topic_index.get(name)
&& existing_id != topic_id
{
return ApplyReply::err(UpdateTopicResult::NameAlreadyExists);
}
+ let options = match patched_options(
+ &topic.options,
+ &self.explicit_options,
+ &self.derived_options,
+ ) {
+ Ok(options) => options,
+ Err(PatchError::InvalidValue) => {
+ return ApplyReply::err(UpdateTopicResult::InvalidOptionValue);
+ }
+ Err(PatchError::TooLarge) => {
+ return
ApplyReply::err(UpdateTopicResult::OptionsBlockTooLarge);
+ }
+ };
- // Decoded before any mutation: a malformed block must leave the topic
- // untouched rather than half-renamed.
- let Ok(mut updated_options) =
resource_options_from_wire(&self.options, true) else {
- return ApplyReply::err(UpdateTopicResult::InvalidOptionValue);
+ let Some(topic) = stream.topics.get_mut(topic_id) else {
+ return ApplyReply::err(UpdateTopicResult::TopicNotFound);
};
- if self
- .options
- .into_iter()
- .any(|entry| entry.key ==
topic_option_keys::PARTITION_RESIZE_POLICY.as_bytes())
- {
- return ApplyReply::err(UpdateTopicResult::UnsupportedOptionKey);
+ if let Some(name) = new_name {
+ stream.topic_index.remove(&topic.name);
+ topic.name = name.clone();
+ stream.topic_index.insert(name, topic_id);
}
- // Read leniently, like every other committed op: a key this build does
- // not know is skipped rather than failing an operation its peers
- // accepted.
- let updated = TopicCreateOptions::parse_committed(&self.options);
- // Default sentinels leave both the effective value and its provenance
- // unchanged, just as an omitted key does.
- updated_options.retain(|key, _| match key.as_str() {
- Ok(topic_option_keys::MESSAGE_EXPIRY) =>
updated.message_expiry.is_some(),
- Ok(topic_option_keys::MAX_TOPIC_SIZE) =>
updated.max_topic_size.is_some(),
- _ => true,
- });
+ topic.set_options(options);
+ ApplyReply::ok(Bytes::new())
+ }
+}
- stream.topic_index.remove(&topic.name);
- topic.name = new_name_arc.clone();
- if let Some(compression_algorithm) = updated.compression_algorithm {
- topic.compression_algorithm = compression_algorithm;
- }
- if let Some(message_expiry) = updated.message_expiry {
- topic.message_expiry = message_expiry;
- }
- if let Some(max_topic_size) = updated.max_topic_size {
- topic.max_topic_size = max_topic_size;
+impl StateHandler for RewrapStreamKeysRequest {
+ type State = StreamsInner;
+ /// Replace the wrap set of every stream incarnation an entry still names.
+ /// An entry for a recreated stream, another key or a malformed set is
+ /// skipped, so a late or repeated rewrap never touches the wrong stream.
+ /// Reply body: the number of sets replaced, `u32` LE.
+ fn apply(&self, state: &mut StreamsInner, _timestamp: IggyTimestamp) ->
ApplyReply {
+ let mut replaced = 0u32;
+ for entry in &self.entries {
+ let Some(stream) = state.items.get_mut(entry.stream_id as usize)
else {
+ continue;
+ };
+ if stream.created_op != entry.created_op {
Review Comment:
Fixed. Added one test per guard:
`given_rewrap_entry_of_earlier_incarnation_with_same_key_when_applying_should_skip_it`
(`core/metadata/src/stm/stream.rs:3455`) needs `created_op`, and
`given_rewrap_entry_of_another_key_when_applying_should_skip_it`
(`core/metadata/src/stm/stream.rs:3496`) needs the stored fingerprint check.
Removing either guard fails only its own test.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
