spetz commented on code in PR #4615:
URL: https://github.com/apache/iggy/pull/4615#discussion_r4337741221
##########
core/server/src/rewrite.rs:
##########
@@ -393,70 +400,499 @@ where
create_partitions.partitions_count,
);
Ok(())
- }),
- Operation::DeletePartitions =>
DeletePartitionsRequest::decode_from(request_body(request))
+ })?,
+ Operation::DeletePartitions =>
DeletePartitionsRequest::decode_from(request_body(&request))
.map_err(|_| IggyError::InvalidCommand)
.and_then(|delete_partitions| {
validate_partitions_change_count(delete_partitions.partitions_count)
- }),
- // Only the updatable subset: the create-time knobs are pushed to
- // partitions when the topic is built and nothing re-pushes them, so
- // accepting one here would store a value no partition ever sees.
- Operation::UpdateTopic =>
UpdateTopicRequest::decode_from(request_body(request))
- .map_err(|_| IggyError::InvalidCommand)
- .and_then(|update_topic| {
- validate_option_keys(&update_topic.options,
UPDATABLE_TOPIC_OPTION_KEYS)?;
- let options =
TopicCreateOptions::parse(&update_topic.options)?;
- let Some(max_topic_size) = options.max_topic_size else {
- return Ok(());
- };
- // An update can lower the cap below one segment just as a
- // create can, and the stored map would then report a size the
- // topic can never enforce. The floor is this topic's own
- // segment size, since that key is create-only.
- let metadata = shard.plane.metadata();
- let streams = metadata.mux_stm.streams();
- let segment_size = streams
- .topic_segment_size(&update_topic.stream_id,
&update_topic.topic_id)
- .map_or_else(
- || iggy_common::DEFAULT_SEGMENT_SIZE,
- |segment_size| segment_size.as_bytes_u64(),
- );
- validate_topic_size_floor(max_topic_size, segment_size)?;
- let partitions_count = streams
- .topic_partitions_count(&update_topic.stream_id,
&update_topic.topic_id)
- .unwrap_or(0);
- warn_unenforceable_topic_size(
- max_topic_size,
- segment_size,
- shard.bus_max_message_size(),
- u32::try_from(partitions_count).unwrap_or(u32::MAX),
- );
- Ok(())
- }),
- Operation::UpdateStream =>
UpdateStreamRequest::decode_from(request_body(request))
- .map_err(|_| IggyError::InvalidCommand)
- .and_then(|update_stream| {
- validate_option_keys(&update_stream.options,
UPDATABLE_STREAM_OPTION_KEYS)
- }),
- Operation::UpdateUser =>
UpdateUserRequest::decode_from(request_body(request))
- .map_err(|_| IggyError::InvalidCommand)
- .and_then(|update_user| {
- validate_option_keys(&update_user.options,
UPDATABLE_USER_OPTION_KEYS)
- }),
- Operation::CreateStream =>
CreateStreamRequest::decode_from(request_body(request))
+ })?,
+ Operation::UpdateResource =>
UpdateResourceRequest::decode_from(request_body(&request))
.map_err(|_| IggyError::InvalidCommand)
- .and_then(|create_stream|
validate_option_keys(&create_stream.options, &[])),
- Operation::CreateUser =>
CreateUserRequest::decode_from(request_body(request))
- .map_err(|_| IggyError::InvalidCommand)
- .and_then(|create_user| validate_option_keys(&create_user.options,
&[])),
- _ => Ok(()),
+ .and_then(|patch| patch_bounds(shard, &patch))?,
+ Operation::CreateStream => {
+ let mut create_stream =
CreateStreamRequest::decode_from(request_body(&request))
+ .map_err(|_| IggyError::InvalidCommand)?;
+ let admitted = admit_create(
+ OptionsScope::Stream,
+ &create_stream.options,
+ OptionColumns {
+ name: Some(create_stream.name.as_str().to_owned()),
+ ..OptionColumns::default()
+ },
+ )?;
+ if let Some(options) = canonical_block(&create_stream.options,
&admitted)? {
+ create_stream.options = options;
+ return rewrite_request_body(&request,
&create_stream.to_bytes());
+ }
+ }
+ Operation::CreateUser => {
+ let mut create_user =
CreateUserRequest::decode_from(request_body(&request))
+ .map_err(|_| IggyError::InvalidCommand)?;
+ let admitted = admit_create(
+ OptionsScope::User,
+ &create_user.options,
+ OptionColumns {
+ username: Some(create_user.username.as_str().to_owned()),
+ ..OptionColumns::default()
+ },
+ )?;
+ UserStatus::from_code(create_user.status).map_err(|_|
IggyError::InvalidUserStatus)?;
+ if let Some(options) = canonical_block(&create_user.options,
&admitted)? {
+ create_user.options = options;
+ return rewrite_request_body(&request, &create_user.to_bytes());
+ }
+ }
+ _ => {}
+ }
+ Ok(request)
+}
+
+/// The admitted canonical block when it differs from what the client sent.
+fn canonical_block(
+ client: &WireOptions,
+ admitted: &AdmittedOptions,
+) -> Result<Option<WireOptions>, IggyError> {
+ let canonical = resource_options_to_wire(&admitted.explicit,
OptionsProvenance::All)?;
+ Ok((canonical.as_bytes() != client.as_bytes()).then_some(canonical))
+}
+
+/// Catalog admission for a generic patch, plus the floor a topic size cap
+/// keeps against the topic's own segment size.
+fn patch_bounds<B, MJ, S, SB>(
+ shard: &Rc<ShellShard<B, MJ, S, SB>>,
+ patch: &UpdateResourceRequest,
+) -> Result<(), IggyError>
+where
+ B: ShellBus,
+ MJ: JournalHandle + 'static,
+ MJ::Target: Journal<Entry = Message<PrepareHeader>, Header =
PrepareHeader>,
+ S: 'static,
+ SB: SuperblockStore + 'static,
+{
+ let metadata = shard.plane.metadata();
+ let admitted = admit_patch(
+ patch,
+ stored_options(
+ metadata.mux_stm.streams(),
+ metadata.mux_stm.users(),
+ &patch.target,
+ ),
+ )?;
+ let ResourceTarget::Topic {
+ stream_id,
+ topic_id,
+ } = &patch.target
+ else {
+ return Ok(());
+ };
+ let Some(max_topic_size) =
Review Comment:
Fixed. The `UpdateTopicResolved` apply now repeats the segment floor against
committed state and refuses with `InvalidTopicSize`
(`core/metadata/src/stm/stream.rs:2496`).
`given_cap_below_topic_segment_size_when_applying_update_should_refuse_it_unchanged`
failed on the old apply, which committed 0, and passes now.
##########
core/server/src/rewrite.rs:
##########
@@ -393,70 +400,499 @@ where
create_partitions.partitions_count,
);
Ok(())
- }),
- Operation::DeletePartitions =>
DeletePartitionsRequest::decode_from(request_body(request))
+ })?,
+ Operation::DeletePartitions =>
DeletePartitionsRequest::decode_from(request_body(&request))
.map_err(|_| IggyError::InvalidCommand)
.and_then(|delete_partitions| {
validate_partitions_change_count(delete_partitions.partitions_count)
- }),
- // Only the updatable subset: the create-time knobs are pushed to
- // partitions when the topic is built and nothing re-pushes them, so
- // accepting one here would store a value no partition ever sees.
- Operation::UpdateTopic =>
UpdateTopicRequest::decode_from(request_body(request))
- .map_err(|_| IggyError::InvalidCommand)
- .and_then(|update_topic| {
- validate_option_keys(&update_topic.options,
UPDATABLE_TOPIC_OPTION_KEYS)?;
- let options =
TopicCreateOptions::parse(&update_topic.options)?;
- let Some(max_topic_size) = options.max_topic_size else {
- return Ok(());
- };
- // An update can lower the cap below one segment just as a
- // create can, and the stored map would then report a size the
- // topic can never enforce. The floor is this topic's own
- // segment size, since that key is create-only.
- let metadata = shard.plane.metadata();
- let streams = metadata.mux_stm.streams();
- let segment_size = streams
- .topic_segment_size(&update_topic.stream_id,
&update_topic.topic_id)
- .map_or_else(
- || iggy_common::DEFAULT_SEGMENT_SIZE,
- |segment_size| segment_size.as_bytes_u64(),
- );
- validate_topic_size_floor(max_topic_size, segment_size)?;
- let partitions_count = streams
- .topic_partitions_count(&update_topic.stream_id,
&update_topic.topic_id)
- .unwrap_or(0);
- warn_unenforceable_topic_size(
- max_topic_size,
- segment_size,
- shard.bus_max_message_size(),
- u32::try_from(partitions_count).unwrap_or(u32::MAX),
- );
- Ok(())
- }),
- Operation::UpdateStream =>
UpdateStreamRequest::decode_from(request_body(request))
- .map_err(|_| IggyError::InvalidCommand)
- .and_then(|update_stream| {
- validate_option_keys(&update_stream.options,
UPDATABLE_STREAM_OPTION_KEYS)
- }),
- Operation::UpdateUser =>
UpdateUserRequest::decode_from(request_body(request))
- .map_err(|_| IggyError::InvalidCommand)
- .and_then(|update_user| {
- validate_option_keys(&update_user.options,
UPDATABLE_USER_OPTION_KEYS)
- }),
- Operation::CreateStream =>
CreateStreamRequest::decode_from(request_body(request))
+ })?,
+ Operation::UpdateResource =>
UpdateResourceRequest::decode_from(request_body(&request))
.map_err(|_| IggyError::InvalidCommand)
- .and_then(|create_stream|
validate_option_keys(&create_stream.options, &[])),
- Operation::CreateUser =>
CreateUserRequest::decode_from(request_body(request))
- .map_err(|_| IggyError::InvalidCommand)
- .and_then(|create_user| validate_option_keys(&create_user.options,
&[])),
- _ => Ok(()),
+ .and_then(|patch| patch_bounds(shard, &patch))?,
+ Operation::CreateStream => {
+ let mut create_stream =
CreateStreamRequest::decode_from(request_body(&request))
+ .map_err(|_| IggyError::InvalidCommand)?;
+ let admitted = admit_create(
+ OptionsScope::Stream,
+ &create_stream.options,
+ OptionColumns {
+ name: Some(create_stream.name.as_str().to_owned()),
+ ..OptionColumns::default()
+ },
+ )?;
+ if let Some(options) = canonical_block(&create_stream.options,
&admitted)? {
+ create_stream.options = options;
+ return rewrite_request_body(&request,
&create_stream.to_bytes());
+ }
+ }
+ Operation::CreateUser => {
+ let mut create_user =
CreateUserRequest::decode_from(request_body(&request))
+ .map_err(|_| IggyError::InvalidCommand)?;
+ let admitted = admit_create(
+ OptionsScope::User,
+ &create_user.options,
+ OptionColumns {
+ username: Some(create_user.username.as_str().to_owned()),
+ ..OptionColumns::default()
+ },
+ )?;
+ UserStatus::from_code(create_user.status).map_err(|_|
IggyError::InvalidUserStatus)?;
+ if let Some(options) = canonical_block(&create_user.options,
&admitted)? {
+ create_user.options = options;
+ return rewrite_request_body(&request, &create_user.to_bytes());
+ }
+ }
+ _ => {}
+ }
+ Ok(request)
+}
+
+/// The admitted canonical block when it differs from what the client sent.
+fn canonical_block(
+ client: &WireOptions,
+ admitted: &AdmittedOptions,
+) -> Result<Option<WireOptions>, IggyError> {
+ let canonical = resource_options_to_wire(&admitted.explicit,
OptionsProvenance::All)?;
+ Ok((canonical.as_bytes() != client.as_bytes()).then_some(canonical))
+}
+
+/// Catalog admission for a generic patch, plus the floor a topic size cap
+/// keeps against the topic's own segment size.
+fn patch_bounds<B, MJ, S, SB>(
+ shard: &Rc<ShellShard<B, MJ, S, SB>>,
+ patch: &UpdateResourceRequest,
+) -> Result<(), IggyError>
+where
+ B: ShellBus,
+ MJ: JournalHandle + 'static,
+ MJ::Target: Journal<Entry = Message<PrepareHeader>, Header =
PrepareHeader>,
+ S: 'static,
+ SB: SuperblockStore + 'static,
+{
+ let metadata = shard.plane.metadata();
+ let admitted = admit_patch(
+ patch,
+ stored_options(
+ metadata.mux_stm.streams(),
+ metadata.mux_stm.users(),
+ &patch.target,
+ ),
+ )?;
+ let ResourceTarget::Topic {
+ stream_id,
+ topic_id,
+ } = &patch.target
+ else {
+ return Ok(());
+ };
+ let Some(max_topic_size) =
+
TopicCreateOptions::from_resource_options(&admitted.explicit).max_topic_size
+ else {
+ return Ok(());
+ };
+ // An update can lower the cap below one segment just as a create can, and
+ // the stored map would then report a size the topic can never enforce. The
+ // floor is this topic's own segment size, since that key is create-only.
+ let streams = metadata.mux_stm.streams();
Review Comment:
Fixed. The edge runs apply's permission rule before reading the target
(`core/server/src/rewrite.rs:501`), and a caller it refuses gets apply's
committed `Unauthorized`, so no edge refusal reveals what the target stores.
`given_caller_without_update_permission_when_patching_topic_below_its_segment_should_leave_the_refusal_to_apply`
pins it.
##########
core/server/src/rewrite.rs:
##########
@@ -393,70 +400,499 @@ where
create_partitions.partitions_count,
);
Ok(())
- }),
- Operation::DeletePartitions =>
DeletePartitionsRequest::decode_from(request_body(request))
+ })?,
+ Operation::DeletePartitions =>
DeletePartitionsRequest::decode_from(request_body(&request))
.map_err(|_| IggyError::InvalidCommand)
.and_then(|delete_partitions| {
validate_partitions_change_count(delete_partitions.partitions_count)
- }),
- // Only the updatable subset: the create-time knobs are pushed to
- // partitions when the topic is built and nothing re-pushes them, so
- // accepting one here would store a value no partition ever sees.
- Operation::UpdateTopic =>
UpdateTopicRequest::decode_from(request_body(request))
- .map_err(|_| IggyError::InvalidCommand)
- .and_then(|update_topic| {
- validate_option_keys(&update_topic.options,
UPDATABLE_TOPIC_OPTION_KEYS)?;
- let options =
TopicCreateOptions::parse(&update_topic.options)?;
- let Some(max_topic_size) = options.max_topic_size else {
- return Ok(());
- };
- // An update can lower the cap below one segment just as a
- // create can, and the stored map would then report a size the
- // topic can never enforce. The floor is this topic's own
- // segment size, since that key is create-only.
- let metadata = shard.plane.metadata();
- let streams = metadata.mux_stm.streams();
- let segment_size = streams
- .topic_segment_size(&update_topic.stream_id,
&update_topic.topic_id)
- .map_or_else(
- || iggy_common::DEFAULT_SEGMENT_SIZE,
- |segment_size| segment_size.as_bytes_u64(),
- );
- validate_topic_size_floor(max_topic_size, segment_size)?;
- let partitions_count = streams
- .topic_partitions_count(&update_topic.stream_id,
&update_topic.topic_id)
- .unwrap_or(0);
- warn_unenforceable_topic_size(
- max_topic_size,
- segment_size,
- shard.bus_max_message_size(),
- u32::try_from(partitions_count).unwrap_or(u32::MAX),
- );
- Ok(())
- }),
- Operation::UpdateStream =>
UpdateStreamRequest::decode_from(request_body(request))
- .map_err(|_| IggyError::InvalidCommand)
- .and_then(|update_stream| {
- validate_option_keys(&update_stream.options,
UPDATABLE_STREAM_OPTION_KEYS)
- }),
- Operation::UpdateUser =>
UpdateUserRequest::decode_from(request_body(request))
- .map_err(|_| IggyError::InvalidCommand)
- .and_then(|update_user| {
- validate_option_keys(&update_user.options,
UPDATABLE_USER_OPTION_KEYS)
- }),
- Operation::CreateStream =>
CreateStreamRequest::decode_from(request_body(request))
+ })?,
+ Operation::UpdateResource =>
UpdateResourceRequest::decode_from(request_body(&request))
.map_err(|_| IggyError::InvalidCommand)
- .and_then(|create_stream|
validate_option_keys(&create_stream.options, &[])),
- Operation::CreateUser =>
CreateUserRequest::decode_from(request_body(request))
- .map_err(|_| IggyError::InvalidCommand)
- .and_then(|create_user| validate_option_keys(&create_user.options,
&[])),
- _ => Ok(()),
+ .and_then(|patch| patch_bounds(shard, &patch))?,
+ Operation::CreateStream => {
+ let mut create_stream =
CreateStreamRequest::decode_from(request_body(&request))
+ .map_err(|_| IggyError::InvalidCommand)?;
+ let admitted = admit_create(
+ OptionsScope::Stream,
+ &create_stream.options,
+ OptionColumns {
+ name: Some(create_stream.name.as_str().to_owned()),
+ ..OptionColumns::default()
+ },
+ )?;
+ if let Some(options) = canonical_block(&create_stream.options,
&admitted)? {
+ create_stream.options = options;
+ return rewrite_request_body(&request,
&create_stream.to_bytes());
+ }
+ }
+ Operation::CreateUser => {
+ let mut create_user =
CreateUserRequest::decode_from(request_body(&request))
+ .map_err(|_| IggyError::InvalidCommand)?;
+ let admitted = admit_create(
+ OptionsScope::User,
+ &create_user.options,
+ OptionColumns {
+ username: Some(create_user.username.as_str().to_owned()),
+ ..OptionColumns::default()
+ },
+ )?;
+ UserStatus::from_code(create_user.status).map_err(|_|
IggyError::InvalidUserStatus)?;
+ if let Some(options) = canonical_block(&create_user.options,
&admitted)? {
+ create_user.options = options;
+ return rewrite_request_body(&request, &create_user.to_bytes());
+ }
+ }
+ _ => {}
+ }
+ Ok(request)
+}
+
+/// The admitted canonical block when it differs from what the client sent.
+fn canonical_block(
+ client: &WireOptions,
+ admitted: &AdmittedOptions,
+) -> Result<Option<WireOptions>, IggyError> {
+ let canonical = resource_options_to_wire(&admitted.explicit,
OptionsProvenance::All)?;
+ Ok((canonical.as_bytes() != client.as_bytes()).then_some(canonical))
+}
+
+/// Catalog admission for a generic patch, plus the floor a topic size cap
+/// keeps against the topic's own segment size.
+fn patch_bounds<B, MJ, S, SB>(
+ shard: &Rc<ShellShard<B, MJ, S, SB>>,
+ patch: &UpdateResourceRequest,
+) -> Result<(), IggyError>
+where
+ B: ShellBus,
+ MJ: JournalHandle + 'static,
+ MJ::Target: Journal<Entry = Message<PrepareHeader>, Header =
PrepareHeader>,
+ S: 'static,
+ SB: SuperblockStore + 'static,
+{
+ let metadata = shard.plane.metadata();
+ let admitted = admit_patch(
+ patch,
+ stored_options(
+ metadata.mux_stm.streams(),
+ metadata.mux_stm.users(),
+ &patch.target,
+ ),
+ )?;
+ let ResourceTarget::Topic {
+ stream_id,
+ topic_id,
+ } = &patch.target
+ else {
+ return Ok(());
+ };
+ let Some(max_topic_size) =
+
TopicCreateOptions::from_resource_options(&admitted.explicit).max_topic_size
+ else {
+ return Ok(());
+ };
+ // An update can lower the cap below one segment just as a create can, and
+ // the stored map would then report a size the topic can never enforce. The
+ // floor is this topic's own segment size, since that key is create-only.
+ let streams = metadata.mux_stm.streams();
+ let segment_size = streams.topic_segment_size(stream_id,
topic_id).map_or_else(
+ || iggy_common::DEFAULT_SEGMENT_SIZE,
+ |segment_size| segment_size.as_bytes_u64(),
+ );
+ validate_topic_size_floor(max_topic_size, segment_size)?;
+ let partitions_count = streams
+ .topic_partitions_count(stream_id, topic_id)
+ .unwrap_or(0);
+ warn_unenforceable_topic_size(
+ max_topic_size,
+ segment_size,
+ shard.bus_max_message_size(),
+ u32::try_from(partitions_count).unwrap_or(u32::MAX),
+ );
+ Ok(())
+}
+
+/// Catalog admission for a create's options block, with the create's own
+/// fixed fields as columns so the block cannot shadow them.
+fn admit_create(
+ scope: OptionsScope,
+ options: &WireOptions,
+ create_columns: OptionColumns,
+) -> Result<AdmittedOptions, IggyError> {
+ admit(
+ scope,
+ OptionOperation::Create,
+ options,
+ &[],
+ &OptionAdmissionContext {
+ current: None,
+ create_columns,
+ },
+ )
+}
+
+/// The options the patch target stores, `None` for a target that does not
+/// exist, which apply reports.
+fn stored_options(
Review Comment:
Kept both admissions, because the edge run spares a log entry and the
primary run is the authoritative one for forwarded requests. The second
`CreateStream` decode is folded (`core/server/src/rewrite.rs:444`), and
`check_block_size` sums the encoded size without a clone or an encode
(`core/common/src/types/options/catalog.rs:946`).
##########
core/server/src/stream_key_rewrapper.rs:
##########
@@ -0,0 +1,278 @@
+// 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::shell::{ServerShard, ShellBus, ShellShard};
+use bytes::Bytes;
+use consensus::{MetadataHandle, PartitionsHandle};
+use iggy_binary_protocol::PrepareHeader;
+use iggy_binary_protocol::requests::streams::{RewrapStreamKeyEntry,
RewrapStreamKeysRequest};
+use iggy_common::{ServerKeyContext, stream_wrap_set};
+use journal::superblock::SuperblockStore;
+use journal::{Journal, JournalHandle};
+use metadata::impls::metadata::StreamsFrontend;
+use metadata::stm::stream::StreamsInner;
+use server_common::Message;
+use shard::Receiver;
+use std::rc::Rc;
+use std::time::Duration;
+use tracing::{info, trace, warn};
+
+/// How often the metadata primary compares stored wrap sets with this node's
+/// configured keys. A pass with nothing to change does no cryptography.
+pub const REWRAP_INTERVAL: Duration = Duration::from_secs(5);
+/// Upper bound on streams rewrapped by one replicated operation, so one
+/// rotation never monopolizes the metadata pipeline. A full batch runs the
+/// next pass at once.
+const MAX_ENTRIES_PER_OPERATION: usize = 64;
+
+/// Outcome of one rewrap pass.
+#[derive(Debug, PartialEq, Eq)]
+enum Pass {
+ Stopped,
+ HitCap,
+ Drained,
+}
+
+/// Run the rewrapper until `stop` fires. Only the caught-up metadata primary
+/// proposes, so a rotation is replicated once and every replica stores the
+/// same wrap sets. A node never rewrites them locally.
+pub async fn run_stream_key_rewrapper(shard: Rc<ServerShard>, stop:
Receiver<()>) {
+ trace!(shard = shard.id, "stream key rewrapper started");
+ 'ticks: loop {
+ match compio::time::timeout(REWRAP_INTERVAL, stop.recv()).await {
+ Ok(_) => break 'ticks,
+ Err(_) => loop {
+ match rewrap_pass(&shard, &stop).await {
+ Pass::HitCap => {}
+ Pass::Drained => break,
+ Pass::Stopped => break 'ticks,
+ }
+ },
+ }
+ }
+ trace!(shard = shard.id, "stream key rewrapper exited");
+}
+
+async fn rewrap_pass<B, MJ, S, SB>(
+ shard: &Rc<ShellShard<B, MJ, S, SB>>,
+ stop: &Receiver<()>,
+) -> Pass
+where
+ B: ShellBus,
+ MJ: JournalHandle + 'static,
+ MJ::Target: Journal<Entry = Message<PrepareHeader>, Header =
PrepareHeader>,
+ S: 'static,
+ SB: SuperblockStore + 'static,
+{
+ if stop.try_recv().is_ok() {
+ return Pass::Stopped;
+ }
+ let metadata = shard.plane.metadata();
+ if !metadata.is_caught_up_primary() {
+ return Pass::Drained;
+ }
+ let context = shard.plane.partitions().key_context();
+ let entries = metadata
+ .mux_stm
+ .streams()
+ .read(|streams| planned_entries(streams, &context));
+ if entries.is_empty() {
+ return Pass::Drained;
+ }
+ let planned = entries.len();
+ match metadata
+ .submit_rewrap_stream_keys_in_process(RewrapStreamKeysRequest {
entries })
+ .await
+ {
+ Ok(commit) => {
+ info!(
+ streams = planned,
+ commit,
+ kek_fingerprints = %context.kek_fingerprints().join(","),
+ "Rewrapped {planned} stream keys under the configured server
keys"
+ );
+ if planned == MAX_ENTRIES_PER_OPERATION {
+ Pass::HitCap
+ } else {
+ Pass::Drained
+ }
+ }
+ Err(error) => {
+ warn!(
+ ?error,
+ "Stream key rewrap not committed, retrying on the next pass"
+ );
+ Pass::Drained
+ }
+ }
+}
+
+/// The streams whose stored wrap set differs from what this node's keys
+/// call for, with the set each should store, at most one operation's worth.
+fn planned_entries(
+ streams: &StreamsInner,
+ context: &ServerKeyContext,
+) -> Vec<RewrapStreamKeyEntry> {
+ streams
+ .items
+ .iter()
+ .filter_map(|(_, stream)| {
+ let stored = stream_wrap_set(&stream.options).ok()??;
Review Comment:
Fixed. `target_set` returns an error for a set this node cannot rewrap, and
each stuck stream warns once with its stored and local KEK fingerprints
(`core/server/src/stream_key_rewrapper.rs:147`). The first sweep that finds
every set settled logs it, which is the phase 2 signal.
`given_sets_this_node_cannot_rewrap_when_planning_should_report_them` pins the
report.
##########
core/server/src/stream_key_rewrapper.rs:
##########
@@ -0,0 +1,278 @@
+// 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::shell::{ServerShard, ShellBus, ShellShard};
+use bytes::Bytes;
+use consensus::{MetadataHandle, PartitionsHandle};
+use iggy_binary_protocol::PrepareHeader;
+use iggy_binary_protocol::requests::streams::{RewrapStreamKeyEntry,
RewrapStreamKeysRequest};
+use iggy_common::{ServerKeyContext, stream_wrap_set};
+use journal::superblock::SuperblockStore;
+use journal::{Journal, JournalHandle};
+use metadata::impls::metadata::StreamsFrontend;
+use metadata::stm::stream::StreamsInner;
+use server_common::Message;
+use shard::Receiver;
+use std::rc::Rc;
+use std::time::Duration;
+use tracing::{info, trace, warn};
+
+/// How often the metadata primary compares stored wrap sets with this node's
+/// configured keys. A pass with nothing to change does no cryptography.
+pub const REWRAP_INTERVAL: Duration = Duration::from_secs(5);
+/// Upper bound on streams rewrapped by one replicated operation, so one
+/// rotation never monopolizes the metadata pipeline. A full batch runs the
+/// next pass at once.
+const MAX_ENTRIES_PER_OPERATION: usize = 64;
+
+/// Outcome of one rewrap pass.
+#[derive(Debug, PartialEq, Eq)]
+enum Pass {
+ Stopped,
+ HitCap,
+ Drained,
+}
+
+/// Run the rewrapper until `stop` fires. Only the caught-up metadata primary
+/// proposes, so a rotation is replicated once and every replica stores the
+/// same wrap sets. A node never rewrites them locally.
+pub async fn run_stream_key_rewrapper(shard: Rc<ServerShard>, stop:
Receiver<()>) {
+ trace!(shard = shard.id, "stream key rewrapper started");
+ 'ticks: loop {
+ match compio::time::timeout(REWRAP_INTERVAL, stop.recv()).await {
+ Ok(_) => break 'ticks,
+ Err(_) => loop {
+ match rewrap_pass(&shard, &stop).await {
+ Pass::HitCap => {}
+ Pass::Drained => break,
+ Pass::Stopped => break 'ticks,
+ }
+ },
+ }
+ }
+ trace!(shard = shard.id, "stream key rewrapper exited");
+}
+
+async fn rewrap_pass<B, MJ, S, SB>(
+ shard: &Rc<ShellShard<B, MJ, S, SB>>,
+ stop: &Receiver<()>,
+) -> Pass
+where
+ B: ShellBus,
+ MJ: JournalHandle + 'static,
+ MJ::Target: Journal<Entry = Message<PrepareHeader>, Header =
PrepareHeader>,
+ S: 'static,
+ SB: SuperblockStore + 'static,
+{
+ if stop.try_recv().is_ok() {
+ return Pass::Stopped;
+ }
+ let metadata = shard.plane.metadata();
+ if !metadata.is_caught_up_primary() {
+ return Pass::Drained;
+ }
+ let context = shard.plane.partitions().key_context();
+ let entries = metadata
+ .mux_stm
+ .streams()
+ .read(|streams| planned_entries(streams, &context));
+ if entries.is_empty() {
+ return Pass::Drained;
+ }
+ let planned = entries.len();
+ match metadata
+ .submit_rewrap_stream_keys_in_process(RewrapStreamKeysRequest {
entries })
+ .await
+ {
+ Ok(commit) => {
+ info!(
+ streams = planned,
+ commit,
+ kek_fingerprints = %context.kek_fingerprints().join(","),
+ "Rewrapped {planned} stream keys under the configured server
keys"
+ );
+ if planned == MAX_ENTRIES_PER_OPERATION {
+ Pass::HitCap
+ } else {
+ Pass::Drained
+ }
+ }
+ Err(error) => {
+ warn!(
+ ?error,
+ "Stream key rewrap not committed, retrying on the next pass"
+ );
+ Pass::Drained
+ }
+ }
+}
+
+/// The streams whose stored wrap set differs from what this node's keys
+/// call for, with the set each should store, at most one operation's worth.
+fn planned_entries(
+ streams: &StreamsInner,
+ context: &ServerKeyContext,
+) -> Vec<RewrapStreamKeyEntry> {
+ streams
+ .items
+ .iter()
+ .filter_map(|(_, stream)| {
+ let stored = stream_wrap_set(&stream.options).ok()??;
+ let target = context.target_set(&stored).ok()??;
+ Some(RewrapStreamKeyEntry {
+ stream_id: u32::try_from(stream.id).ok()?,
+ created_op: stream.created_op,
+ key_fingerprint:
stored.key_fingerprint_bytes().try_into().ok()?,
+ wrap_set: Bytes::copy_from_slice(target.as_ref()),
+ })
+ })
+ .take(MAX_ENTRIES_PER_OPERATION)
Review Comment:
Fixed. A sweep scans every stream once, collects the unsettled ids and
proposes them in operations of at most 64 entries
(`core/server/src/stream_key_rewrapper.rs:228`). `HitCap` and the re-pass are
gone.
`given_more_stale_streams_than_one_operation_when_sweeping_should_plan_each_from_one_scan`
plans 136 stale streams from one scan.
--
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]
