spetz commented on code in PR #4615:
URL: https://github.com/apache/iggy/pull/4615#discussion_r4337737015
##########
core/partitions/src/iggy_partitions.rs:
##########
@@ -799,8 +833,13 @@ where
// user headers per message. The ciphertext is what gets journaled,
// replicated, checksummed, and persisted -- every replica stores
// identical bytes -- and the poll reply is the single decrypt point.
+ // The partition's own encryptor: each stream has its own key, pinned
+ // to the partition when it was built.
+ let encryptor = self
Review Comment:
Fixed. `on_request_with_reply` does one `get_mut_by_ns` lookup
(`core/partitions/src/iggy_partitions.rs:836`) and encrypts through
`partition.encryptor()` by reference
(`core/partitions/src/iggy_partitions.rs:860`), so there is no second lookup
and no `Arc` clone per batch. The partitions lib suite passes unchanged.
##########
core/server/src/boot/mod.rs:
##########
@@ -911,6 +911,26 @@ async fn shard_main(
None
};
+ // Stream key rewrapper: shard 0 only, and only when this node holds a
+ // server key. Each pass no-ops unless this node is the caught-up metadata
+ // primary, so a key rotation is replicated once to every replica.
+ let stream_key_rewrapper_stop =
+ if shard_id == 0 &&
shard.plane.partitions().key_context().accepts_stream_keys() {
Review Comment:
Fixed. The rewrapper starts only on a node with previous keys or retired
fingerprints (`core/server/src/boot/mod.rs:930`), and a tick returns at once
unless this node is the caught-up primary and metadata moved since the last
sweep (`core/server/src/stream_key_rewrapper.rs:128`).
`given_unchanged_metadata_when_sweeping_again_should_skip_the_scan` pins the
skip.
##########
core/server/src/boot/recovery.rs:
##########
@@ -127,16 +142,38 @@ pub(in crate::boot) async fn build_shard_for_thread(
let owned_partitions_capacity = total_partitions
.div_ceil(usize::from(total_shards).max(1))
.saturating_mul(2);
- // At-rest encryption: built once per shard from the shared config; the
- // ingestion path encrypts on the primary and the poll reply decrypts.
- // A bad key fails the boot rather than silently serving plaintext.
- let encryptor = if config.encryption.enabled {
- let aes = Aes256GcmEncryptor::from_base64_key(&config.encryption.key)
- .map_err(|error| ServerError::Iggy(Box::new(error)))?;
- Some(Arc::new(EncryptorKind::Aes256Gcm(aes)))
- } else {
- None
- };
+ // Each encrypted stream holds its own key, wrapped under the server key.
+ // Every one must open here before anything serves, so a changed config key
+ // stops the boot instead of failing sends and polls one by one.
+ let key_context = Arc::new(
+ ServerKeyContext::from_config(
+ &config.encryption.key,
+ &config.encryption.previous_keys,
+ config.encryption.enabled,
+ )
+ .map_err(|error| ServerError::Iggy(Box::new(error)))?,
+ );
+ let unopenable = metadata.mux_stm.streams().read(|inner| {
+ inner.items.iter().find_map(
+ |(_, stream)| match StreamEncryption::from(&stream.options) {
+ StreamEncryption::Plaintext => None,
+ StreamEncryption::Unavailable(error) => Some((stream.id,
Vec::new(), error)),
+ StreamEncryption::Wrapped(set) => key_context
+ .open(&set)
+ .err()
+ .map(|error| (stream.id, set.kek_fingerprints(), error)),
+ },
+ )
+ });
+ if let Some((stream_id, stored_kek_fingerprints, source)) = unopenable {
+ return Err(ServerError::StreamKeyUnavailable {
Review Comment:
Fixed. A cluster replica whose stream key does not open logs both
fingerprint sets and leaves that stream's partitions to the reconciler instead
of refusing boot (`core/server/src/boot/recovery.rs:253`), and a standalone
replay still refuses.
`given_replica_that_missed_a_rotation_when_booting_with_the_new_key_should_serve_after_catching_up`
pins it.
##########
core/server/src/boot/recovery.rs:
##########
@@ -204,8 +248,20 @@ 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 {
+ for OwnedPartition {
+ stream_id,
+ topic_id,
+ stats: partition_stats,
+ metadata: partition_metadata,
+ runtime: topic_runtime,
+ identity,
+ encryption,
+ } in owned
+ {
let namespace = IggyNamespace::new(stream_id, topic_id,
partition_metadata.id);
+ let encryptor = partitions
Review Comment:
Fixed. Only shard 0 runs the boot-wide check, and only for encrypted streams
that own no partition (`core/server/src/boot/recovery.rs:159`). Every other set
is opened once by each shard that owns one of its partitions
(`core/server/src/boot/recovery.rs:216`). Keyrings stay per shard, so no
`Arc<EncryptorKind>` is shared across cores.
--
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]
