spetz commented on code in PR #4152:
URL: https://github.com/apache/iggy/pull/4152#discussion_r3999104291
##########
core/connectors/sdk/src/decoders/proto.rs:
##########
@@ -384,18 +407,32 @@ impl ProtoStreamDecoder {
Type::Bool => {
simd_json::OwnedValue::Static(simd_json::StaticNode::Bool(value != 0))
}
- Type::Int32 | Type::Sint32 | Type::Sfixed32 => {
- simd_json::OwnedValue::from(value as i32)
+ Type::Int32 | Type::Sfixed32 =>
simd_json::OwnedValue::from(value as i32),
+ Type::Int64 | Type::Sfixed64 =>
simd_json::OwnedValue::from(value as i64),
+ Type::Sint32 => {
+ let value = value as u32;
+ simd_json::OwnedValue::from(((value >> 1) as i32) ^
-((value & 1) as i32))
}
- Type::Int64 | Type::Sint64 | Type::Sfixed64 => {
- simd_json::OwnedValue::from(value as i64)
+ Type::Sint64 => {
+ simd_json::OwnedValue::from(((value >> 1) as i64) ^
-((value & 1) as i64))
}
Type::Uint32 | Type::Fixed32 =>
simd_json::OwnedValue::from(value as u32),
Type::Uint64 | Type::Fixed64 =>
simd_json::OwnedValue::from(value),
_ => simd_json::OwnedValue::from(value),
};
Ok((json_value, new_cursor))
}
+ 1 | 5 => {
+ let (value, new_cursor) = Self::parse_fixed_integer(data,
cursor, wire_type)?;
+ let json_value = match (wire_type, field_desc.r#type()) {
+ (1, Type::Fixed64) => simd_json::OwnedValue::from(value),
+ (1, Type::Sfixed64) => simd_json::OwnedValue::from(value
as i64),
+ (5, Type::Fixed32) => simd_json::OwnedValue::from(value as
u32),
+ (5, Type::Sfixed32) => simd_json::OwnedValue::from(value
as u32 as i32),
+ _ =>
simd_json::OwnedValue::String("unsupported_wire_type".into()),
Review Comment:
`Type::Float` (wire type 5) and `Type::Double` (wire type 1) are absent from
the `(wire_type, field_desc.r#type())` match, so both fall to this arm and
decode to the literal string `"unsupported_wire_type"`. The same PR teaches
`ProtoStreamEncoder` and `ProtoConvert` to write those two types as
little-endian fixed32 and fixed64, so a JSON to protobuf to JSON round trip now
loses exactly the two types the encoder just learned to emit.
##########
core/connectors/sdk/src/decoders/proto.rs:
##########
@@ -83,26 +83,32 @@ impl ProtoStreamDecoder {
}
pub fn update_config(&mut self, config: ProtoConfig, reload_schema: bool)
-> Result<(), Error> {
- self.config = config;
- if reload_schema
- && (self.config.schema_path.is_some() ||
self.config.descriptor_set.is_some())
- {
- self.load_schema()
- } else {
- Ok(())
+ let old_config = std::mem::replace(&mut self.config, config);
+ if reload_schema && let Err(error) = self.load_schema() {
+ self.config = old_config;
+ return Err(error);
}
+ Ok(())
}
pub fn load_schema(&mut self) -> Result<(), Error> {
let schema_path = self.config.schema_path.clone();
let descriptor_set = self.config.descriptor_set.clone();
- if let Some(path) = schema_path {
- self.compile_schema_internal(&path)?;
+ let old_message_descriptor = self.message_descriptor.take();
Review Comment:
`load_schema` takes the descriptors up front and restores them only on
`Err`, but `compile_schema_internal` returns `Ok(())` after logging "Falling
back to Any wrapper mode" for both an unreadable file and a protox compile
failure. A broken schema keeps the last good descriptor while a missing one
discards it and downgrades output to the Any envelope, both reporting success.
`given_loaded_schemas_when_file_changes_should_preserve_errors_and_clear_fallback`
asserts exactly that split, so it is deliberate. Same shape at
`core/connectors/sdk/src/encoders/proto.rs:120` and
`core/connectors/sdk/src/transforms/proto_convert.rs:114`.
##########
core/connectors/runtime/src/state/http.rs:
##########
@@ -298,7 +298,7 @@ impl HttpStateProvider {
return Ok(LoadResponse::NotFound);
}
Ok(response) => return Ok(LoadResponse::Failure(response)),
- Err(send_error) => TransientFailure::Send(send_error),
+ Err(send_error) =>
TransientFailure::Send(send_error.without_url()),
Review Comment:
The `without_url()` redaction is applied to both `TransientFailure::Send`
arms but not to the `TransientFailure::Read` arm at
`core/connectors/runtime/src/state/http.rs:294`, which stores
`read_error.to_string()` and feeds it verbatim into `describe` at
`core/connectors/runtime/src/state/http.rs:619`. In reqwest 0.13.4, the pinned
workspace version, `Response::bytes()` delegates to `do_bytes`, whose error
path is `crate::error::decode(err).with_url(*self.url)`, and `impl Display for
Error` ends with `if let Some(url) = &self.inner.url { write!(f, " for url
({url})")?; }`. A body-read failure on the load path logs the full state URL
with its query string, the exact leak this change closes for send failures.
`resource_label` is already redacted through `url_label`, so this is the only
remaining vector, and the new
`given_unavailable_state_store_when_booting_should_fail_startup` assertion
exercises only the `Send` path.
--
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]
