spetz commented on code in PR #4615:
URL: https://github.com/apache/iggy/pull/4615#discussion_r4340259441
##########
core/server/src/boot/recovery.rs:
##########
@@ -204,8 +235,35 @@ pub(in crate::boot) async fn build_shard_for_thread(
// bundle was broadcast (see `MetadataHandoff::Owner`). All shards
// here only add their per-partition deltas, so the shared
// `Arc<TopicStats>` atomics race only against other atomic adds.
- for (stream_id, topic_id, partition_stats, partition_metadata,
topic_runtime) in owned {
+ let mut deferred_streams = BTreeSet::new();
+ for OwnedPartition {
+ stream_id,
+ topic_id,
+ stats: partition_stats,
+ metadata: partition_metadata,
+ runtime: topic_runtime,
+ identity,
+ encryption,
+ } in owned
+ {
+ if deferred_streams.contains(&stream_id) {
+ continue;
+ }
let namespace = IggyNamespace::new(stream_id, topic_id,
partition_metadata.id);
+ let encryptor = match partitions.resolve_stream_encryptor(identity,
&encryption) {
+ Ok(encryptor) => encryptor,
+ Err(source) => {
+ defer_or_refuse_stream_key(
+ stream_id,
+ &encryption,
+ &partitions.key_context(),
+ source,
+ solo,
+ )?;
+ deferred_streams.insert(stream_id);
+ continue;
Review Comment:
The decision now lives in `owned_partition_key` at
`core/server/src/boot/recovery.rs:407`, and
`given_stream_key_that_does_not_open_when_deciding_owned_partitions_should_leave_them_unbuilt`
fails if a key that does not open yields a build without an encryptor. The
wrong-key cluster test in
`core/integration/tests/cluster/stream_encryption.rs:116` also asserts that
node has no partition directory.
##########
core/server/src/boot/recovery.rs:
##########
@@ -398,6 +458,98 @@ const _: () =
assert!(consensus::DVC_HEADERS_MAX ==
iggy_binary_protocol::consensus::DVC_HEADERS_MAX);
const _: () = assert!(consensus::DVC_HEADERS_MAX == u128::BITS as usize);
+/// The boot check of encrypted streams that own no partition, whose keys no
+/// shard opens while building its partitions.
+fn refuse_partitionless_stream_keys(
+ streams: &Streams,
+ key_context: &ServerKeyContext,
+ solo: bool,
+) -> Result<(), ServerError> {
+ let unopenable: Vec<(usize, StreamEncryption, IggyError)> =
streams.read(|inner| {
+ inner
+ .items
+ .iter()
+ .filter(|(_, stream)| {
+ stream
+ .topics
+ .iter()
+ .all(|(_, topic)| topic.partitions.is_empty())
+ })
+ .filter_map(|(_, stream)| {
+ let encryption = StreamEncryption::from(&stream.options);
+ let source = match &encryption {
+ StreamEncryption::Plaintext => None,
+ StreamEncryption::Unavailable(error) =>
Some(error.clone()),
+ StreamEncryption::Wrapped(set) =>
key_context.open(set).err(),
+ }?;
+ Some((stream.id, encryption, source))
+ })
+ .collect()
+ });
+ for (stream_id, encryption, source) in unopenable {
+ defer_or_refuse_stream_key(stream_id, &encryption, key_context,
source, solo)?;
+ }
+ Ok(())
+}
+
+/// A stream key that does not open at boot. A cluster replica's metadata can
+/// trail the commit point until catch-up or state transfer, so the stream
waits
+/// for the reconciler, which builds its partitions once a wrap this node opens
+/// commits. Catch-up heals neither a node without a key nor unreadable
critical
+/// state, and a solo replay is the whole journal, so these refuse.
+fn defer_or_refuse_stream_key(
+ stream_id: usize,
+ encryption: &StreamEncryption,
+ key_context: &ServerKeyContext,
+ source: IggyError,
+ solo: bool,
+) -> Result<(), ServerError> {
+ let stored_kek_fingerprints = match encryption {
+ StreamEncryption::Wrapped(set) => set.kek_fingerprints().join(","),
+ StreamEncryption::Plaintext | StreamEncryption::Unavailable(_) =>
String::new(),
+ };
+ let error = ServerError::StreamKeyUnavailable {
+ stream_id,
+ stored_kek_fingerprints,
+ node_kek_fingerprints: key_context.kek_fingerprints().join(","),
+ source: Box::new(source),
+ };
+ if solo
Review Comment:
A node whose keys leave a committed stream unopenable now refuses to wrap a
new create with `StreamEncryptionUnavailable`
(`core/server/src/stream_keys.rs:129`), and the primary refuses a CreateStream
wrap set none of its keys opens before it is sealed into a prepare
(`core/metadata/src/impls/metadata.rs:4414`). KEK agreement in the replica
handshake stays a follow-up.
##########
core/server/src/stream_key_rewrapper.rs:
##########
@@ -0,0 +1,538 @@
+// 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;
+use iggy_binary_protocol::PrepareHeader;
+use iggy_binary_protocol::requests::streams::{RewrapStreamKeyEntry,
RewrapStreamKeysRequest};
+use iggy_common::{HeaderKey, IggyError, OptionKey, ServerKeyContext,
stream_wrap_set};
+use journal::superblock::SuperblockStore;
+use journal::{Journal, JournalHandle};
+use metadata::impls::metadata::StreamsFrontend;
+use metadata::stm::stream::{Stream, StreamsInner};
+use server_common::Message;
+use shard::Receiver;
+use std::collections::BTreeSet;
+use std::rc::Rc;
+use std::time::Duration;
+use tracing::{info, trace, warn};
+
+/// How often the metadata primary checks whether a sweep is due. A tick with
+/// an unchanged view and applied op does nothing else.
+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.
+const MAX_ENTRIES_PER_OPERATION: usize = 64;
+
+/// What the rewrapper keeps between ticks.
+#[derive(Debug, Default)]
+struct Rewrapper {
+ /// Metadata view and last applied op of the last finished sweep.
+ swept: Option<(u32, u64)>,
+ /// Stream incarnations already reported as stuck, so each warns once.
+ stuck: BTreeSet<(usize, u64)>,
+ /// Whether the last finished sweep found every set settled.
+ settled: bool,
+}
+
+/// Outcome of one tick.
+#[derive(Debug, PartialEq, Eq)]
+enum Sweep {
+ Stopped,
+ NotPrimary,
+ Unchanged,
+ Interrupted,
+ Finished,
+}
+
+/// A stream that needs a rewrap this node cannot make.
+#[derive(Debug)]
+struct StuckStream {
+ stream_id: usize,
+ created_op: u64,
+ kek_fingerprints: String,
+ error: IggyError,
+}
+
+/// Run the rewrapper until `stop` fires, planning with `context`, this node's
+/// keys plus the retired fingerprints. Spawned only when that context is
+/// rotating. 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>,
+ context: ServerKeyContext,
+ stop: Receiver<()>,
+) {
+ trace!(shard = shard.id, "stream key rewrapper started");
+ let mut rewrapper = Rewrapper::default();
+ loop {
+ if compio::time::timeout(REWRAP_INTERVAL, stop.recv())
+ .await
+ .is_ok()
+ || rewrapper.sweep(&shard, &context, &stop).await == Sweep::Stopped
+ {
+ break;
+ }
+ }
+ trace!(shard = shard.id, "stream key rewrapper exited");
+}
+
+impl Rewrapper {
+ /// Scan every stream once when this node is the caught-up primary and
+ /// the metadata moved since the last finished sweep, then propose the
+ /// unsettled ones in bounded operations.
+ async fn sweep<B, MJ, S, SB>(
+ &mut self,
+ shard: &Rc<ShellShard<B, MJ, S, SB>>,
+ context: &ServerKeyContext,
+ stop: &Receiver<()>,
+ ) -> Sweep
+ 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 Sweep::Stopped;
+ }
+ let metadata = shard.plane.metadata();
+ let Some(consensus) = metadata
+ .consensus
+ .as_ref()
+ .filter(|_| metadata.is_caught_up_primary())
+ else {
+ self.swept = None;
+ return Sweep::NotPrimary;
+ };
+ // Read with the scan below and no await between, so the mark names
+ // exactly the state the scan saw.
+ let mark = (consensus.view(), consensus.commit_min());
Review Comment:
Added
`given_settled_primary_when_a_stale_stream_commits_should_rescan_and_rewrap_it`
at `core/server/src/stream_key_rewrapper.rs:373`. It settles a journal-backed
primary, commits an old-key create through the replicate path and requires the
next sweep to rewrap it. The `(consensus.view(), 0)` mutant fails it.
--
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]
