spetz commented on code in PR #4615:
URL: https://github.com/apache/iggy/pull/4615#discussion_r4337739153
##########
core/server/src/http/handlers.rs:
##########
@@ -1882,45 +1747,65 @@ pub(in crate::http) async fn create_user(
Ok(Json(decode_user_details(&payload)?))
}
-/// `PUT /users/{user_id}`: update a user's username and/or status. Returns
204.
+/// `PUT /users/{user_id}`: update the user's username, status and options. An
+/// absent or `null` key keeps its value, a key in `reset` returns to its
+/// default. Returns 204 on commit.
pub(in crate::http) async fn update_user(
State(state): State<HttpState>,
identity: Authenticated,
Path(user_id): Path<String>,
- Json(command): Json<UpdateUser>,
+ update: Result<Json<OptionsUpdate>, JsonRejection>,
) -> Result<StatusCode, WriteError> {
let user_id =
Identifier::from_str_value(&user_id).map_err(WriteError::Rejected)?;
- // Rejects an oversized replacement username; a no-op when username is
absent.
- command.validate().map_err(WriteError::Rejected)?;
- let user_update_options = UserUpdateOptions {
- raw: command.options,
- }
- .to_wire()
- .map_err(WriteError::Rejected)?;
- validate_option_keys(&user_update_options, UPDATABLE_USER_OPTION_KEYS)
- .map_err(WriteError::Rejected)?;
- let request = UpdateUserRequest {
+ let target = ResourceTarget::User {
user_id: identifier_to_wire(&user_id).map_err(WriteError::Rejected)?,
- username: command
- .username
- .as_deref()
- .map(WireName::new)
- .transpose()
- .map_err(|_| WriteError::Rejected(IggyError::InvalidUsername))?,
- status: command.status.map(|status| status.as_code()),
- options: user_update_options,
+ };
+ submit_update(&state, &identity, target, update).await
+}
+
+/// Submit an update body as the generic `UpdateResource` the binary transports
+/// send, so admission and apply are shared.
+async fn submit_update(
+ state: &HttpState,
+ identity: &Authenticated,
+ target: ResourceTarget,
+ update: Result<Json<OptionsUpdate>, JsonRejection>,
+) -> Result<StatusCode, WriteError> {
+ // Duplicate members, nested values or a non-object body are malformed
+ // structure, the `InvalidCommand` the binary decoder answers too.
+ let Json(update) = update.map_err(|_|
WriteError::Rejected(IggyError::InvalidCommand))?;
+ let parts = update.to_patch().map_err(WriteError::Rejected)?;
+ let request = UpdateResourceRequest {
+ target,
+ reset: parts.reset,
+ options: parts.set,
};
let body = request.to_bytes();
SendWrapper::new(submit_write(
- &state,
+ state,
&identity.session,
- Operation::UpdateUser,
+ Operation::UpdateResource,
body,
))
.await?;
Ok(StatusCode::NO_CONTENT)
}
+/// Refuses a create body member that names a stored option of `scope`, an
+/// option the client placed beside `options` instead of inside it.
+fn refuse_misplaced_options(
Review Comment:
Fixed. Both refusals now name the member in `field`,
`"field":"message_expiry"` for the old topic create and `"field":"options"` for
the old stream update (`core/server/src/http/handlers.rs:1806`,
`core/server/src/http/extractor.rs:161`). The code stays `InvalidCommand`.
`given_option_beside_options_when_creating_should_reject_invalid_command`
asserts the field.
##########
core/server/src/partition_reconciler.rs:
##########
@@ -582,6 +586,23 @@ async fn reconcile_once(ctx: &ReconcilerCtx) -> bool {
true
}
+/// Forget the keys of deleted stream incarnations, so a reused stream id can
+/// never meet a previous stream's key.
+fn reconcile_stream_keys(ctx: &ReconcilerCtx) {
Review Comment:
Fixed. `reconcile_stream_keys` returns at once when the shard holds no
stream key and otherwise checks each keyring entry with one slab lookup
(`core/server/src/partition_reconciler.rs:590`).
`given_deleted_encrypted_stream_when_reconciling_should_forget_its_key` pins
the prune, which no test covered before.
##########
core/server/src/partition_reconciler.rs:
##########
@@ -748,11 +769,54 @@ async fn reconcile_additions(
// Resolved only for namespaces actually built, not once per committed
// partition every pass. A stream, topic or partition that vanished
// between the target snapshot and this read defers to the next pass.
- let Some((partition_stats, topic_runtime, partition_metadata)) =
- fetch_partition_build_inputs(ctx, ns)
+ let Some(PartitionBuildInputs {
+ stats: partition_stats,
+ runtime: topic_runtime,
+ metadata: partition_metadata,
+ stream_key,
+ encryption,
+ }) = fetch_partition_build_inputs(ctx, ns)
else {
continue;
};
+ let topic_runtime = match topic_runtime {
+ Ok(topic_runtime) => topic_runtime,
+ Err(error) => {
+ ctx.record_failure(ns, FailureCause::Add, now);
+ ctx.shard.metrics().record_partition_reconcile_failure();
+ error!(
+ shard = shard_id,
+ stream_id = ns.stream_id(),
+ topic_id = ns.topic_id(),
+ partition_id = ns.partition_id(),
+ %error,
+ "Partition of topic {} cannot be built, its options hold
critical state this node cannot read",
+ ns.topic_id()
+ );
+ continue;
+ }
+ };
+ // An encrypted partition is built only with its own stream's key. A
+ // key that does not open here leaves the partition unbuilt, never
+ // served as plaintext or under another key.
+ let encryptor = match partitions.resolve_stream_encryptor(stream_key,
&encryption) {
+ Ok(encryptor) => encryptor,
+ Err(error) => {
+ ctx.record_failure(ns, FailureCause::Add, now);
+ ctx.shard.metrics().record_partition_reconcile_failure();
+ error!(
+ shard = shard_id,
+ stream_id = ns.stream_id(),
+ topic_id = ns.topic_id(),
+ partition_id = ns.partition_id(),
+ kek_fingerprints =
%partitions.key_context().kek_fingerprints().join(","),
+ %error,
+ "Partition of stream {} cannot be built, its key does not
open on this node",
+ ns.stream_id()
+ );
+ continue;
Review Comment:
Fixed.
`given_stream_key_that_does_not_open_when_reconciling_should_not_build_its_partition`
(`core/server/src/partition_reconciler.rs:2781`) requires the stream under
another key to stay unbuilt, with an Add failure recorded and no partition
directory. Your mutant fails it with `left: Some(false)`, `right: None`.
##########
core/server/src/responses.rs:
##########
@@ -589,155 +587,103 @@ where
S: 'static,
SB: SuperblockStore + 'static,
{
- shard.plane.metadata().mux_stm.streams().read(|streams| {
- streams
- .items
- .iter()
- .map(|(_, stream)| stream_response(stream))
- .collect::<Result<Vec<_>, _>>()
- .map(|streams| GetStreamsResponse { streams })
- })
+ shard
+ .plane
+ .metadata()
+ .mux_stm
+ .streams()
+ .read(|streams| listed_streams(streams).map(|streams|
GetStreamsResponse { streams }))
Review Comment:
Fixed. Stream, topic and user reads copy only the stored fields and options
under the read guard (`core/server/src/responses.rs:95`) and run the
projection, the wrap set parse and the fingerprint hex after it is released.
The hex now fills one preallocated string
(`core/common/src/utils/stream_key.rs:844`).
--
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]
