You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@iggy.apache.org by "numinnex (via GitHub)" <gi...@apache.org> on 2026/08/13 20:49:32 UTC
[PR] feat(common)!: extensible key-value options for topics, streams, users [iggy]
numinnex opened a new pull request, #3880:
URL: https://github.com/apache/iggy/pull/3880
(no comment)
--
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: commits-unsubscribe@iggy.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] feat(common)!: extensible key-value options for topics, streams, users [iggy]
Posted by "numinnex (via GitHub)" <gi...@apache.org>.
numinnex commented on PR #3880:
URL: https://github.com/apache/iggy/pull/3880#issuecomment-5300897405
/ready
--
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: commits-unsubscribe@iggy.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] feat(common)!: extensible key-value options for topics, streams, users [iggy]
Posted by "spetz (via GitHub)" <gi...@apache.org>.
spetz merged PR #3880:
URL: https://github.com/apache/iggy/pull/3880
--
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: commits-unsubscribe@iggy.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] feat(common)!: extensible key-value options for topics, streams, users [iggy]
Posted by "codecov[bot] (via GitHub)" <gi...@apache.org>.
codecov[bot] commented on PR #3880:
URL: https://github.com/apache/iggy/pull/3880#issuecomment-5290109581
## [Codecov](https://app.codecov.io/gh/apache/iggy/pull/3880?dropdown=coverage&src=pr&el=h1&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) Report
:x: Patch coverage is `80.86643%` with `53 lines` in your changes missing coverage. Please review.
:white_check_mark: Project coverage is 82.24%. Comparing base ([`4eb879e`](https://app.codecov.io/gh/apache/iggy/commit/4eb879e8220a707aace1f249d41073637e9ce1ff?dropdown=coverage&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache)) to head ([`ca150e2`](https://app.codecov.io/gh/apache/iggy/commit/ca150e2a70aacdf94edf2056a6e581662118e19e?dropdown=coverage&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache)).
:warning: Report is 1 commits behind head on master.
| [Files with missing lines](https://app.codecov.io/gh/apache/iggy/pull/3880?dropdown=coverage&src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | Patch % | Lines |
|---|---|---|
| [...nary\_serialization/binary\_response\_deserializer.go](https://app.codecov.io/gh/apache/iggy/pull/3880?src=pr&el=tree&filepath=foreign%2Fgo%2Fbinary_serialization%2Fbinary_response_deserializer.go&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#diff-Zm9yZWlnbi9nby9iaW5hcnlfc2VyaWFsaXphdGlvbi9iaW5hcnlfcmVzcG9uc2VfZGVzZXJpYWxpemVyLmdv) | 58.20% | [25 Missing and 3 partials :warning: ](https://app.codecov.io/gh/apache/iggy/pull/3880?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) |
| [...oreign/node/src/wire/topic/update-topic.command.ts](https://app.codecov.io/gh/apache/iggy/pull/3880?src=pr&el=tree&filepath=foreign%2Fnode%2Fsrc%2Fwire%2Ftopic%2Fupdate-topic.command.ts&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#diff-Zm9yZWlnbi9ub2RlL3NyYy93aXJlL3RvcGljL3VwZGF0ZS10b3BpYy5jb21tYW5kLnRz) | 31.25% | [11 Missing :warning: ](https://app.codecov.io/gh/apache/iggy/pull/3880?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) |
| [foreign/go/contracts/compression\_algorithm.go](https://app.codecov.io/gh/apache/iggy/pull/3880?src=pr&el=tree&filepath=foreign%2Fgo%2Fcontracts%2Fcompression_algorithm.go&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#diff-Zm9yZWlnbi9nby9jb250cmFjdHMvY29tcHJlc3Npb25fYWxnb3JpdGhtLmdv) | 0.00% | [6 Missing :warning: ](https://app.codecov.io/gh/apache/iggy/pull/3880?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) |
| [foreign/go/internal/command/update\_user.go](https://app.codecov.io/gh/apache/iggy/pull/3880?src=pr&el=tree&filepath=foreign%2Fgo%2Finternal%2Fcommand%2Fupdate_user.go&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#diff-Zm9yZWlnbi9nby9pbnRlcm5hbC9jb21tYW5kL3VwZGF0ZV91c2VyLmdv) | 0.00% | [5 Missing :warning: ](https://app.codecov.io/gh/apache/iggy/pull/3880?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) |
| [foreign/go/internal/command/topic.go](https://app.codecov.io/gh/apache/iggy/pull/3880?src=pr&el=tree&filepath=foreign%2Fgo%2Finternal%2Fcommand%2Ftopic.go&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#diff-Zm9yZWlnbi9nby9pbnRlcm5hbC9jb21tYW5kL3RvcGljLmdv) | 95.34% | [1 Missing and 1 partial :warning: ](https://app.codecov.io/gh/apache/iggy/pull/3880?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) |
| [foreign/php/src/client.rs](https://app.codecov.io/gh/apache/iggy/pull/3880?src=pr&el=tree&filepath=foreign%2Fphp%2Fsrc%2Fclient.rs&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#diff-Zm9yZWlnbi9waHAvc3JjL2NsaWVudC5ycw==) | 95.00% | [1 Missing :warning: ](https://app.codecov.io/gh/apache/iggy/pull/3880?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) |
<details><summary>Additional details and impacted files</summary>
```diff
@@ Coverage Diff @@
## master #3880 +/- ##
============================================
- Coverage 82.78% 82.24% -0.55%
+ Complexity 1299 1297 -2
============================================
Files 1199 1200 +1
Lines 161885 160970 -915
Branches 131360 130417 -943
============================================
- Hits 134019 132385 -1634
- Misses 24322 25089 +767
+ Partials 3544 3496 -48
```
| [Components](https://app.codecov.io/gh/apache/iggy/pull/3880/components?src=pr&el=components&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | Coverage Δ | |
|---|---|---|
| [Rust Core](https://app.codecov.io/gh/apache/iggy/pull/3880/components?src=pr&el=component&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `83.32% <ø> (+0.04%)` | :arrow_up: |
| [Java SDK](https://app.codecov.io/gh/apache/iggy/pull/3880/components?src=pr&el=component&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `65.98% <ø> (-0.09%)` | :arrow_down: |
| [C# SDK](https://app.codecov.io/gh/apache/iggy/pull/3880/components?src=pr&el=component&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `75.88% <ø> (-0.10%)` | :arrow_down: |
| [Python SDK](https://app.codecov.io/gh/apache/iggy/pull/3880/components?src=pr&el=component&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `89.98% <ø> (ø)` | |
| [PHP SDK](https://app.codecov.io/gh/apache/iggy/pull/3880/components?src=pr&el=component&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `84.26% <ø> (ø)` | |
| [Node SDK](https://app.codecov.io/gh/apache/iggy/pull/3880/components?src=pr&el=component&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `96.21% <ø> (ø)` | |
| [Go SDK](https://app.codecov.io/gh/apache/iggy/pull/3880/components?src=pr&el=component&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `52.73% <67.46%> (-15.81%)` | :arrow_down: |
| [Files with missing lines](https://app.codecov.io/gh/apache/iggy/pull/3880?dropdown=coverage&src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | Coverage Δ | |
|---|---|---|
| [core/ai/mcp/src/service/mod.rs](https://app.codecov.io/gh/apache/iggy/pull/3880?src=pr&el=tree&filepath=core%2Fai%2Fmcp%2Fsrc%2Fservice%2Fmod.rs&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#diff-Y29yZS9haS9tY3Avc3JjL3NlcnZpY2UvbW9kLnJz) | `60.05% <ø> (-0.35%)` | :arrow_down: |
| [core/binary\_protocol/src/codes.rs](https://app.codecov.io/gh/apache/iggy/pull/3880?src=pr&el=tree&filepath=core%2Fbinary_protocol%2Fsrc%2Fcodes.rs&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#diff-Y29yZS9iaW5hcnlfcHJvdG9jb2wvc3JjL2NvZGVzLnJz) | `100.00% <ø> (ø)` | |
| [core/binary\_protocol/src/dispatch.rs](https://app.codecov.io/gh/apache/iggy/pull/3880?src=pr&el=tree&filepath=core%2Fbinary_protocol%2Fsrc%2Fdispatch.rs&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#diff-Y29yZS9iaW5hcnlfcHJvdG9jb2wvc3JjL2Rpc3BhdGNoLnJz) | `92.88% <ø> (ø)` | |
| [...ary\_protocol/src/requests/streams/create\_stream.rs](https://app.codecov.io/gh/apache/iggy/pull/3880?src=pr&el=tree&filepath=core%2Fbinary_protocol%2Fsrc%2Frequests%2Fstreams%2Fcreate_stream.rs&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#diff-Y29yZS9iaW5hcnlfcHJvdG9jb2wvc3JjL3JlcXVlc3RzL3N0cmVhbXMvY3JlYXRlX3N0cmVhbS5ycw==) | `100.00% <ø> (ø)` | |
| [...ary\_protocol/src/requests/streams/update\_stream.rs](https://app.codecov.io/gh/apache/iggy/pull/3880?src=pr&el=tree&filepath=core%2Fbinary_protocol%2Fsrc%2Frequests%2Fstreams%2Fupdate_stream.rs&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#diff-Y29yZS9iaW5hcnlfcHJvdG9jb2wvc3JjL3JlcXVlc3RzL3N0cmVhbXMvdXBkYXRlX3N0cmVhbS5ycw==) | `100.00% <ø> (ø)` | |
| [...inary\_protocol/src/requests/topics/create\_topic.rs](https://app.codecov.io/gh/apache/iggy/pull/3880?src=pr&el=tree&filepath=core%2Fbinary_protocol%2Fsrc%2Frequests%2Ftopics%2Fcreate_topic.rs&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#diff-Y29yZS9iaW5hcnlfcHJvdG9jb2wvc3JjL3JlcXVlc3RzL3RvcGljcy9jcmVhdGVfdG9waWMucnM=) | `100.00% <ø> (ø)` | |
| [...c/requests/topics/create\_topic\_with\_assignments.rs](https://app.codecov.io/gh/apache/iggy/pull/3880?src=pr&el=tree&filepath=core%2Fbinary_protocol%2Fsrc%2Frequests%2Ftopics%2Fcreate_topic_with_assignments.rs&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#diff-Y29yZS9iaW5hcnlfcHJvdG9jb2wvc3JjL3JlcXVlc3RzL3RvcGljcy9jcmVhdGVfdG9waWNfd2l0aF9hc3NpZ25tZW50cy5ycw==) | `87.50% <ø> (+0.32%)` | :arrow_up: |
| [...inary\_protocol/src/requests/topics/update\_topic.rs](https://app.codecov.io/gh/apache/iggy/pull/3880?src=pr&el=tree&filepath=core%2Fbinary_protocol%2Fsrc%2Frequests%2Ftopics%2Fupdate_topic.rs&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#diff-Y29yZS9iaW5hcnlfcHJvdG9jb2wvc3JjL3JlcXVlc3RzL3RvcGljcy91cGRhdGVfdG9waWMucnM=) | `100.00% <ø> (ø)` | |
| [.../binary\_protocol/src/requests/users/create\_user.rs](https://app.codecov.io/gh/apache/iggy/pull/3880?src=pr&el=tree&filepath=core%2Fbinary_protocol%2Fsrc%2Frequests%2Fusers%2Fcreate_user.rs&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#diff-Y29yZS9iaW5hcnlfcHJvdG9jb2wvc3JjL3JlcXVlc3RzL3VzZXJzL2NyZWF0ZV91c2VyLnJz) | `96.21% <ø> (-0.03%)` | :arrow_down: |
| [.../binary\_protocol/src/requests/users/update\_user.rs](https://app.codecov.io/gh/apache/iggy/pull/3880?src=pr&el=tree&filepath=core%2Fbinary_protocol%2Fsrc%2Frequests%2Fusers%2Fupdate_user.rs&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#diff-Y29yZS9iaW5hcnlfcHJvdG9jb2wvc3JjL3JlcXVlc3RzL3VzZXJzL3VwZGF0ZV91c2VyLnJz) | `100.00% <ø> (ø)` | |
| ... and [96 more](https://app.codecov.io/gh/apache/iggy/pull/3880?src=pr&el=tree-more&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | |
... and [36 files with indirect coverage changes](https://app.codecov.io/gh/apache/iggy/pull/3880/indirect-changes?src=pr&el=tree-more&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache)
</details>
<details><summary> :rocket: New features to boost your workflow: </summary>
- :snowflake: [Test Analytics](https://docs.codecov.com/docs/test-analytics): Detect flaky tests, report on failures, and find test suite problems.
- :package: [JS Bundle Analysis](https://docs.codecov.com/docs/javascript-bundle-analysis): Save yourself from yourself by tracking and limiting bundle sizes in JS merges.
</details>
--
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: commits-unsubscribe@iggy.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] feat(common)!: extensible key-value options for topics, streams, users [iggy]
Posted by "hubcio (via GitHub)" <gi...@apache.org>.
hubcio commented on code in PR #3880:
URL: https://github.com/apache/iggy/pull/3880#discussion_r3782571817
##########
core/configs/src/configs_impl/file_provider.rs:
##########
@@ -75,6 +118,7 @@ impl<P: Provider + Clone> ConfigProvider for FileConfigProvider<P> {
// If the config file exists, merge it into the configuration
if file_exists(&self.file_path) {
info!("Found configuration file at path: '{}'.", self.file_path);
+ reject_relocated_keys(&self.file_path)?;
Review Comment:
this guard only covers the toml file, and only when one exists - it sits inside `if file_exists(...)` and the env provider merges below it unchecked. so `IGGY_SYSTEM_PARTITION_ENFORCE_FSYNC=true` in a pure-env container (the usual k8s shape) warns and boots with fsync off - exactly the failure the doc above says this prevents. and since the partition runtime knobs are create-only, existing topics can never get fsync back after an upgrade. run the relocated-key check over env names too, outside the `file_exists` branch - the harness already has this exact check (`validate_env_var_names`), the server just never runs it. also `[system.message_deduplication]` was deleted but isn't in this list, so `enabled = true` slips through the same way. worth a test too - this fn has none.
##########
foreign/go/binary_serialization/binary_response_deserializer.go:
##########
@@ -57,8 +57,10 @@ func DeserializeStream(payload []byte) (*iggcon.StreamDetails, error) {
if err != nil {
return nil, err
}
- topics := make([]iggcon.Topic, 0)
- for pos < len(payload) {
+ // Count-driven: a topic element carries variable-length options blocks,
+ // so "consume until the buffer ends" no longer delimits it.
+ topics := make([]iggcon.Topic, 0, stream.TopicsCount)
Review Comment:
preallocating from an unvalidated wire u32 - `TopicsCount` at max is a ~400 GB reservation and a go OOM isn't recoverable. `DeserializeTopics` caps the same hint further down; do the same here. (the rust decoders got the same fix via `bounded_capacity`.)
##########
core/common/src/wire_conversions.rs:
##########
@@ -799,6 +813,144 @@ pub(crate) fn user_headers_from_validated_slice(
Ok(headers)
}
+// -- Options conversions --
+
+/// Decode a `DescribeOptions` response into domain option specs.
+///
+/// # Errors
+///
+/// Returns `IggyError::InvalidHeaderKind` when an entry carries an unknown
+/// canonical kind code.
+pub fn option_specs_from_wire(
+ wire: iggy_binary_protocol::responses::system::DescribeOptionsResponse,
+) -> Result<Vec<OptionSpec>, IggyError> {
+ wire.entries
+ .into_iter()
+ .map(|entry| {
+ Ok(OptionSpec {
+ key: entry.key.to_string(),
+ kind: HeaderKind::from_code(entry.kind)?,
+ default_value: entry.default_value.to_vec(),
+ description: entry.description,
+ })
+ })
+ .collect()
+}
+
+/// Decode a [`WireOptions`] block into domain resource options.
+///
+/// Every decoded entry is marked with the given `explicit` flag; the wire
+/// block carries no per-entry provenance. Admission calls this once for the
+/// client block (`explicit == true`) and fills defaults separately.
+///
+/// # Errors
+///
+/// Returns `IggyError::InvalidHeaderKey` / `InvalidHeaderValue` when a kind
+/// code has no domain meaning or a fixed-size kind carries a mismatched
+/// payload. The wire layer accepts unknown kinds for forwarding; interpreting
+/// them into the domain requires known semantics.
+pub fn resource_options_from_wire(
Review Comment:
`user_headers_from_validated_slice` errors on any value kind outside the known range, but the wire layer deliberately forwards unknown value kinds so mixed-version clusters can round-trip them - that's the stated contract in `options.rs`. this fn sits on the apply path, the sdk list-decode path, and the server's own http reads, so the first new `HeaderKind` code means: old replicas reject a commit new ones accept (permanent divergence), `get_topics()` fails the whole list over one entry, and the topic GET errors server-side. skipping entries with unknown kinds here closes all three at once. the comment at the apply site promises leniency the code doesn't deliver. (side note: the doc here says `InvalidHeaderKey`/`InvalidHeaderValue` but the actual error is `InvalidHeaderKind`.)
##########
foreign/go/internal/command/update_user.go:
##########
@@ -50,7 +54,8 @@ func (u *UpdateUser) MarshalBinary() ([]byte, error) {
length += 2
}
- bytes := make([]byte, length+1)
+ optionsBytes := iggcon.GetHeadersBytes(u.Options)
+ bytes := make([]byte, length+1+len(optionsBytes))
Review Comment:
`length` already counts both flag bytes when username and status are set, so the `+1` leaves a trailing `0x00` after the options block - and the server now decodes options to end-of-payload and rejects kind 0, so every `UpdateUser` with both fields set fails. the over-allocation is old but the new wire format makes it fatal; drop the `+1` and size the buffer exactly. (the both-nil case is off by one the other way and panics, but that predates this PR.)
##########
core/metadata/src/stm/stream.rs:
##########
@@ -1659,23 +1685,41 @@ impl StateHandler for CreateTopicWithAssignmentsRequest {
return ApplyReply::err(CreateTopicResult::StreamNotFound);
};
- let replication_factor = if self.request.replication_factor == 0 {
- 1
- } else {
- self.request.replication_factor
+ // Both blocks were validated and resolved at admission, so apply reads
+ // them leniently: a key this build does not know is skipped, not
+ // refused. Refusing would make the verdict depend on the build, and a
+ // replica that predates a key would then be missing a topic its peers
+ // committed.
+ let explicit = TopicCreateOptions::parse_committed(&self.request.options);
+ let derived = TopicCreateOptions::parse_committed(&self.derived_options);
+ let (Ok(explicit_map), Ok(derived_map)) = (
+ resource_options_from_wire(&self.request.options, true),
+ resource_options_from_wire(&self.derived_options, false),
+ ) else {
+ return ApplyReply::err(CreateTopicResult::InvalidOptionValue);
};
+ // Explicit wins on collision. `partitions_count` cannot appear here at
+ // all: it is a fixed field of the command, not an option key.
+ let mut options = derived_map;
Review Comment:
`extend` lets a client-sent literal 0 overwrite the derived entry, while the typed field below normalizes 0 to `None` and correctly resolves to the default. net effect: the resolved default is dropped from the persisted map entirely, so the map stops being the single source of truth its doc claims, and one `GetTopic` response shows the resolved value in the fixed field but 0 in the options block. reachable from the shipped cli: `iggy topic create --set max_topic_size=server_default` lands the sentinel via the raw map. after the merge, overwrite these keys from the resolved typed values.
##########
core/server/src/dispatch.rs:
##########
@@ -1138,6 +1176,31 @@ async fn handle_client_request<B, MJ, S, SB>(
.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)?;
+ TopicCreateOptions::parse(&update_topic.options).map(|_| ())
Review Comment:
update still has no bounds check for `max_topic_size` - create runs `validate_topic_bounds` (twice, here and in `prepare_request`), but there's no `Operation::UpdateTopic` arm in `prepare_request` and no bounds call here, so an update can set a cap below one segment size and the stored map reports it as the topic's setting.
##########
core/metadata/src/stm/stream.rs:
##########
@@ -1659,23 +1685,41 @@ impl StateHandler for CreateTopicWithAssignmentsRequest {
return ApplyReply::err(CreateTopicResult::StreamNotFound);
};
- let replication_factor = if self.request.replication_factor == 0 {
- 1
- } else {
- self.request.replication_factor
+ // Both blocks were validated and resolved at admission, so apply reads
+ // them leniently: a key this build does not know is skipped, not
+ // refused. Refusing would make the verdict depend on the build, and a
+ // replica that predates a key would then be missing a topic its peers
+ // committed.
+ let explicit = TopicCreateOptions::parse_committed(&self.request.options);
+ let derived = TopicCreateOptions::parse_committed(&self.derived_options);
+ let (Ok(explicit_map), Ok(derived_map)) = (
+ resource_options_from_wire(&self.request.options, true),
+ resource_options_from_wire(&self.derived_options, false),
+ ) else {
+ return ApplyReply::err(CreateTopicResult::InvalidOptionValue);
Review Comment:
when this decode fails, the revision bump and the stats-registry insert above have already happened - reconcilers get woken and the registry keeps an orphaned entry for a topic that never got created. deterministic across replicas so nothing diverges, but the "past validation" comment above is no longer true. move the bump and insert below the decode.
##########
core/binary_protocol/src/responses/topics/get_topics.rs:
##########
@@ -16,40 +16,57 @@
// under the License.
use crate::WireError;
-use crate::codec::{WireDecode, WireEncode};
+use crate::codec::{WireDecode, WireEncode, read_u32_le};
use crate::responses::streams::get_stream::TopicHeader;
-use bytes::BytesMut;
+use bytes::{BufMut, BytesMut};
-/// `GetTopics` response: sequential topic headers.
+/// `GetTopics` response: count-prefixed topic headers.
///
/// Wire format:
/// ```text
-/// [TopicHeader]*
+/// [topics_count:u32_le][TopicHeader]*
/// ```
///
-/// Empty payload means zero topics.
+/// The count prefix exists because a topic element carries variable-length
+/// options blocks: without it, a decoder has no framing to detect a short
+/// read against.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct GetTopicsResponse {
pub topics: Vec<TopicHeader>,
}
impl WireEncode for GetTopicsResponse {
fn encoded_size(&self) -> usize {
- self.topics.iter().map(WireEncode::encoded_size).sum()
+ 4 + self
+ .topics
+ .iter()
+ .map(WireEncode::encoded_size)
+ .sum::<usize>()
}
fn encode(&self, buf: &mut BytesMut) {
+ #[allow(clippy::cast_possible_truncation)]
+ buf.put_u32_le(self.topics.len() as u32);
for topic in &self.topics {
topic.encode(buf);
}
}
}
+/// Smallest a topic header can encode as: the fixed ids, timestamps, sizes and
+/// counts, plus a name length and two length-prefixed option blocks.
+const MIN_TOPIC_HEADER_SIZE: usize = 45;
Review Comment:
45 contradicts the comment above - the fixed fields alone are 49 bytes, and with `name_len >= 1` plus two 4-byte option-block prefixes the real minimum is 59 (`TopicHeader::FIXED_SIZE + 1 + 8`). undersizing just loosens the cap, but go (`topicMinimumSize`) and java (`MIN_TOPIC_BYTES`) both say 58 from the same zero-length-name assumption - `WireName` rejects empty names, so all three want 59.
##########
core/binary_protocol/src/requests/topics/create_topic_with_assignments.rs:
##########
@@ -72,9 +88,27 @@ impl WireDecode for CreateTopicWithAssignmentsRequest {
}
let request = CreateTopicRequest::decode_from(&buf[request_start..request_end])?;
- let partitions_count = read_u32_le(buf, request_end)? as usize;
- let mut offset = request_end + 4;
- let mut partitions = Vec::with_capacity(partitions_count);
+
+ let derived_size = read_u32_le(buf, request_end)? as usize;
+ let derived_start = request_end + 4;
+ let derived_end = derived_start + derived_size;
Review Comment:
unchecked `pos + size` (also `request_end` at line 81) - the same 32-bit wrap `decode_options_prefixed` guards with `checked_add`, and this decode runs on journal replay, so a wrap would be a replica panic rather than a client error. the derived-block read here is that helper hand-rolled minus its guard - call it instead.
##########
core/server/src/responses.rs:
##########
@@ -943,13 +978,122 @@ where
})
}
+/// Serve the option catalog for one resource scope. Static specs plus this
+/// node's resolved defaults; streams and users have no catalog keys yet, so
+/// their scopes return empty (every key is rejected at create until one
+/// lands).
+fn build_describe_options_response<B, MJ, S, SB>(
+ shard: &Rc<ShellShard<B, MJ, S, SB>>,
+ body: &[u8],
+) -> Result<DescribeOptionsResponse, IggyError>
+where
+ B: ShellBus,
+ MJ: JournalHandle + 'static,
+ MJ::Target: Journal<MJ::Storage, Entry = Message<PrepareHeader>, Header = PrepareHeader>,
+ S: 'static,
+ SB: SuperblockStore + 'static,
+{
+ let request =
+ DescribeOptionsRequest::decode_from(body).map_err(|_| IggyError::InvalidCommand)?;
+ let entries = match request.scope {
+ OPTIONS_SCOPE_TOPIC => {
+ let metadata = shard.plane.metadata();
+ let (enforce_fsync, messages_required_to_save, size_of_messages, preallocate) =
+ metadata.partition_runtime_defaults();
+ vec![
+ OptionDescriptor {
+ key: WireName::new(topic_option_keys::COMPRESSION_ALGORITHM)
+ .map_err(|_| IggyError::InvalidFormat)?,
+ kind: HeaderKind::String.as_code(),
+ default_value: Bytes::from_static(b"none"),
+ description: "Compression algorithm (none, gzip)".to_string(),
+ },
+ OptionDescriptor {
+ key: WireName::new(topic_option_keys::MESSAGE_EXPIRY)
+ .map_err(|_| IggyError::InvalidFormat)?,
+ kind: HeaderKind::Uint64.as_code(),
+ default_value: Bytes::copy_from_slice(
+ &metadata.default_message_expiry().to_le_bytes(),
+ ),
+ description: "Message expiry in microseconds, or a humantime string \
+ (e.g. 7 days)"
+ .to_string(),
+ },
+ OptionDescriptor {
+ key: WireName::new(topic_option_keys::MAX_TOPIC_SIZE)
+ .map_err(|_| IggyError::InvalidFormat)?,
+ kind: HeaderKind::Uint64.as_code(),
+ default_value: Bytes::copy_from_slice(
+ &metadata.default_max_topic_size().to_le_bytes(),
+ ),
+ description: "Topic size cap in bytes, or a byte-size string (e.g. 1 GiB); \
+ must be at least the segment size"
+ .to_string(),
+ },
+ OptionDescriptor {
+ key: WireName::new(topic_option_keys::SEGMENT_SIZE)
+ .map_err(|_| IggyError::InvalidFormat)?,
+ kind: HeaderKind::Uint64.as_code(),
+ default_value: Bytes::copy_from_slice(
+ &metadata.default_segment_size().to_le_bytes(),
+ ),
+ description: format!(
+ "Segment size in bytes, or a byte-size string (e.g. 128 MiB); a 512-byte \
+ multiple within {}..={}",
+ iggy_common::MIN_TOPIC_SEGMENT_SIZE,
+ metadata.max_topic_segment_size()
+ ),
+ },
+ OptionDescriptor {
+ key: WireName::new(topic_option_keys::ENFORCE_FSYNC)
+ .map_err(|_| IggyError::InvalidFormat)?,
+ kind: HeaderKind::Bool.as_code(),
+ default_value: Bytes::copy_from_slice(&[u8::from(enforce_fsync)]),
+ description: "Whether writes to this topic's partitions fsync".to_string(),
+ },
+ OptionDescriptor {
+ key: WireName::new(topic_option_keys::MESSAGES_REQUIRED_TO_SAVE)
+ .map_err(|_| IggyError::InvalidFormat)?,
+ kind: HeaderKind::Uint32.as_code(),
+ default_value: Bytes::copy_from_slice(&messages_required_to_save.to_le_bytes()),
+ description: "Flush the journal once it holds this many messages; must be \
+ non-zero"
+ .to_string(),
+ },
+ OptionDescriptor {
+ key: WireName::new(topic_option_keys::SIZE_OF_MESSAGES_REQUIRED_TO_SAVE)
+ .map_err(|_| IggyError::InvalidFormat)?,
+ kind: HeaderKind::Uint64.as_code(),
+ default_value: Bytes::copy_from_slice(&size_of_messages.to_le_bytes()),
+ description: "Flush the journal once it holds this many bytes, or a \
+ byte-size string; whichever threshold trips first flushes"
+ .to_string(),
+ },
+ OptionDescriptor {
+ key: WireName::new(topic_option_keys::PREALLOCATE_SEGMENTS)
+ .map_err(|_| IggyError::InvalidFormat)?,
+ kind: HeaderKind::Bool.as_code(),
+ default_value: Bytes::copy_from_slice(&[u8::from(preallocate)]),
+ description: "Reserve each segment's bytes up front where the filesystem \
Review Comment:
this description is the operator's only discovery surface for the option and it doesn't mention the cost: the fallocate runs inline on the shard thread (every rotation, plus a serial re-run per owned partition at boot), and nothing bounds `segment_size * partitions_count`, so one create can reserve partitions x 1 GiB. worth a sentence here and a cap at admission. the inline call itself is right - the old spawn_blocking panicked the shard.
##########
core/common/src/types/options/mod.rs:
##########
@@ -0,0 +1,1295 @@
+// 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.
+
+//! Key-value options attached to streams, topics, and users at creation.
+//!
+//! Options reuse the typed [`HeaderKey`] / [`HeaderValue`] machinery from
+//! message user headers. The map MUST stay a `BTreeMap`: metadata snapshots
+//! require deterministic ordering across replicas, and a hash map would make
+//! snapshot bytes diverge per replica.
+
+use std::collections::BTreeMap;
+use std::str::FromStr;
+
+use iggy_binary_protocol::{WireOptions, WireUserHeaderEntry};
+use serde::{Deserialize, Serialize};
+
+use crate::types::compression::compression_algorithm::CompressionAlgorithm;
+use crate::types::message::{HeaderKey, HeaderKind, HeaderValue};
+use crate::{IggyByteSize, IggyError, IggyExpiry, MaxTopicSize};
+
+/// A single resolved option entry.
+#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
+pub struct OptionValue {
+ /// The effective value, resolved by the admitting primary at creation.
+ pub value: HeaderValue,
+ /// Whether the client explicitly sent this key. Derived entries
+ /// (`explicit == false`) were filled from server defaults at admission
+ /// and would have resolved differently under other server configs.
+ pub explicit: bool,
+}
+
+/// Which provenance classes an encoded options block carries.
+///
+/// Responses split a resource's options into two blocks so `GetTopic` can say
+/// which values the client chose and which admission filled in.
+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+pub enum OptionsProvenance {
+ /// Every entry, whatever its provenance.
+ All,
+ /// Only the keys the client sent.
+ Explicit,
+ /// Only the keys admission resolved from server defaults.
+ Derived,
+}
+
+impl OptionValue {
+ /// Whether this entry belongs in a block encoded for `provenance`.
+ #[must_use]
+ pub const fn matches(&self, provenance: OptionsProvenance) -> bool {
+ match provenance {
+ OptionsProvenance::All => true,
+ OptionsProvenance::Explicit => self.explicit,
+ OptionsProvenance::Derived => !self.explicit,
+ }
+ }
+
+ #[must_use]
+ pub fn explicit(value: HeaderValue) -> Self {
+ Self {
+ value,
+ explicit: true,
+ }
+ }
+
+ #[must_use]
+ pub fn derived(value: HeaderValue) -> Self {
+ Self {
+ value,
+ explicit: false,
+ }
+ }
+}
+
+/// Options attached to a stream, topic, or user, keyed by option name.
+pub type ResourceOptions = BTreeMap<HeaderKey, OptionValue>;
+
+/// JSON form for [`ResourceOptions`].
+///
+/// [`HeaderKey`] serializes as a struct (`kind` + base64 `value`), and JSON
+/// object keys must be strings, so the derived impl fails outright with
+/// "key must be a string" the moment a map is non-empty. Every HTTP response
+/// carrying options goes through here instead, rendering the key as its plain
+/// text. Binary transports are unaffected: they never touch serde.
+pub mod resource_options_json {
+ use super::{HeaderKey, OptionValue, ResourceOptions};
+ use serde::de::Error as _;
+ use serde::{Deserialize, Deserializer, Serialize, Serializer};
+ use std::collections::BTreeMap;
+ use std::str::FromStr;
+
+ /// # Errors
+ ///
+ /// Propagates the serializer's own errors.
+ pub fn serialize<S: Serializer>(
+ options: &ResourceOptions,
+ serializer: S,
+ ) -> Result<S::Ok, S::Error> {
+ let readable: BTreeMap<String, &OptionValue> = options
+ .iter()
+ .map(|(key, value)| (String::from_utf8_lossy(key.as_bytes()).into_owned(), value))
+ .collect();
+ readable.serialize(serializer)
+ }
+
+ /// # Errors
+ ///
+ /// Returns a deserializer error when a key is not a valid option name.
+ pub fn deserialize<'de, D: Deserializer<'de>>(
+ deserializer: D,
+ ) -> Result<ResourceOptions, D::Error> {
+ let readable = BTreeMap::<String, OptionValue>::deserialize(deserializer)?;
+ readable
+ .into_iter()
+ .map(|(key, value)| {
+ HeaderKey::from_str(&key)
+ .map(|key| (key, value))
+ .map_err(D::Error::custom)
+ })
+ .collect()
+ }
+}
+
+/// Topic option catalog: the keys `CreateTopic` accepts.
+pub mod topic_option_keys {
+ /// Compression algorithm name (`none`, `gzip`).
+ pub const COMPRESSION_ALGORITHM: &str = "compression_algorithm";
+ /// Message expiry: `Uint64` micros or a humantime string (`7 days`).
+ pub const MESSAGE_EXPIRY: &str = "message_expiry";
+ /// Topic size cap: `Uint64` bytes or a byte-size string (`1 GiB`).
+ pub const MAX_TOPIC_SIZE: &str = "max_topic_size";
+ /// Segment size: `Uint64` bytes or a byte-size string (`128 MiB`).
+ /// Bounded: a 512-byte multiple within
+ /// [`super::MIN_TOPIC_SEGMENT_SIZE`]..=the server's segment ceiling.
+ pub const SEGMENT_SIZE: &str = "segment_size";
+ /// Whether writes to this topic's partitions fsync: `Bool`, or the
+ /// strings `true` / `false`.
+ pub const ENFORCE_FSYNC: &str = "enforce_fsync";
+ /// Flush the journal once it holds this many messages: `Uint32`.
+ /// Must be non-zero.
+ pub const MESSAGES_REQUIRED_TO_SAVE: &str = "messages_required_to_save";
+ /// Flush the journal once it holds this many bytes: `Uint64` or a
+ /// byte-size string. Paired with [`Self::MESSAGES_REQUIRED_TO_SAVE`];
+ /// whichever threshold trips first flushes.
+ pub const SIZE_OF_MESSAGES_REQUIRED_TO_SAVE: &str = "size_of_messages_required_to_save";
+ /// Reserve the segment's bytes up front on a filesystem that supports it:
+ /// `Bool`, or the strings `true` / `false`. Pairs with
+ /// [`Self::SEGMENT_SIZE`] -- preallocation reserves exactly that much, so
+ /// the two only make sense decided together.
+ pub const PREALLOCATE_SEGMENTS: &str = "preallocate_segments";
+}
+
+/// Values an absent topic option resolves to at admission.
+///
+/// These are the knobs' single source of truth: they used to live in
+/// `config.toml` (`[system.topic]`, `[system.partition]`, `[system.segment]`),
+/// which meant every one of them had two homes and an operator could not tell
+/// which won. A topic carries whatever it was created with; anything the
+/// client did not send resolves to the constant here and is persisted as a
+/// derived entry, so the effective value is always visible on `GetTopic`.
+///
+/// Each value matches what the shipped `config.toml` carried, so removing the
+/// keys changed no behavior for a topic created without options.
+pub const DEFAULT_PARTITIONS_COUNT: u32 = 1;
+/// `MaxTopicSize::Unlimited` (was `[system.topic] max_size = "unlimited"`).
+pub const DEFAULT_MAX_TOPIC_SIZE: u64 = u64::MAX;
+/// `IggyExpiry::NeverExpire` (was `[system.topic] message_expiry = "none"`).
+pub const DEFAULT_MESSAGE_EXPIRY: u64 = u64::MAX;
+/// 1 GiB (was `[system.segment] size = "1 GiB"`).
+pub const DEFAULT_SEGMENT_SIZE: u64 = 1024 * 1024 * 1024;
+/// Was `[system.partition] enforce_fsync = false`.
+pub const DEFAULT_ENFORCE_FSYNC: bool = false;
+/// Was `[system.partition] messages_required_to_save = 1024`.
+pub const DEFAULT_MESSAGES_REQUIRED_TO_SAVE: u32 = 1024;
+/// Opt-in, unlike the `[system.segment] preallocate = true` this replaced.
+///
+/// That default was never actually in force: the reservation ran through
+/// `compio::spawn_blocking`, which panics the shard because shard executors
+/// disable the blocking pool, so any deployment that worked at all had
+/// preallocation off. With the call fixed to run inline it reserves real
+/// extents, and `FALLOC_FL_KEEP_SIZE` against the 1 GiB default segment size
+/// means 1 GiB of disk per partition the moment it is created -- a full test
+/// sweep reserved 393 GB before this was flipped. A topic that wants the
+/// latency benefit asks for it with `preallocate_segments`.
+pub const DEFAULT_PREALLOCATE_SEGMENTS: bool = false;
+/// 1 MiB (was `[system.partition] size_of_messages_required_to_save`).
+pub const DEFAULT_SIZE_OF_MESSAGES_REQUIRED_TO_SAVE: u64 = 1024 * 1024;
+
+/// Every runtime knob at its default, for a partition built with no resolved
+/// topic options (simulator, unit tests).
+impl Default for TopicRuntimeDefaults {
+ fn default() -> Self {
+ Self {
+ segment_size: IggyByteSize::from(DEFAULT_SEGMENT_SIZE),
+ enforce_fsync: DEFAULT_ENFORCE_FSYNC,
+ messages_required_to_save: DEFAULT_MESSAGES_REQUIRED_TO_SAVE,
+ size_of_messages_required_to_save: IggyByteSize::from(
+ DEFAULT_SIZE_OF_MESSAGES_REQUIRED_TO_SAVE,
+ ),
+ preallocate_segments: DEFAULT_PREALLOCATE_SEGMENTS,
+ }
+ }
+}
+
+/// Smallest per-topic segment size. Segments far below the shipped default
+/// explode the per-partition segment count, which state transfer bounds via
+/// its manifest entry cap; a partition that crosses it becomes unservable
+/// for transfer, so the floor is enforced at admission.
+pub const MIN_TOPIC_SEGMENT_SIZE: u64 = 1024 * 1024;
+
+/// Validate an explicit per-topic `segment_size` against its bounds.
+///
+/// `ceiling` is node-derived: the smaller of the global segment maximum and
+/// the state-transfer artifact budget minus one bus frame (a segment may
+/// close one whole batch past its cap; an artifact ceiling below that
+/// refuses a legal segment and livelocks the partition's rejoin).
+///
+/// # Errors
+///
+/// Returns `IggyError::InvalidOptionValue("segment_size")` when the value is
+/// below [`MIN_TOPIC_SEGMENT_SIZE`], above `ceiling`, or not a 512-byte
+/// multiple.
+pub fn validate_topic_segment_size(size_bytes: u64, ceiling: u64) -> Result<(), IggyError> {
+ if size_bytes < MIN_TOPIC_SEGMENT_SIZE
+ || size_bytes > ceiling
+ || !size_bytes.is_multiple_of(512)
+ {
+ return Err(IggyError::InvalidOptionValue(
+ topic_option_keys::SEGMENT_SIZE.to_string(),
+ ));
+ }
+ Ok(())
+}
+
+/// Resource whose option catalog `DescribeOptions` serves.
+#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
+#[serde(rename_all = "snake_case")]
+pub enum OptionsScope {
+ Topic,
+ Stream,
+ User,
+}
+
+impl OptionsScope {
+ #[must_use]
+ pub fn as_code(&self) -> u8 {
+ match self {
+ Self::Topic => 1,
+ Self::Stream => 2,
+ Self::User => 3,
+ }
+ }
+
+ /// # Errors
+ ///
+ /// Returns `IggyError::InvalidCommand` for an unknown scope code.
+ pub fn from_code(code: u8) -> Result<Self, IggyError> {
+ match code {
+ 1 => Ok(Self::Topic),
+ 2 => Ok(Self::Stream),
+ 3 => Ok(Self::User),
+ _ => Err(IggyError::InvalidCommand),
+ }
+ }
+}
+
+impl std::fmt::Display for OptionsScope {
+ fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+ match self {
+ Self::Topic => write!(f, "topic"),
+ Self::Stream => write!(f, "stream"),
+ Self::User => write!(f, "user"),
+ }
+ }
+}
+
+impl FromStr for OptionsScope {
+ type Err = IggyError;
+
+ fn from_str(scope: &str) -> Result<Self, Self::Err> {
+ match scope {
+ "topic" => Ok(Self::Topic),
+ "stream" => Ok(Self::Stream),
+ "user" => Ok(Self::User),
+ _ => Err(IggyError::InvalidCommand),
+ }
+ }
+}
+
+/// One entry of a resource's option catalog, as served by `DescribeOptions`.
+#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
+pub struct OptionSpec {
+ /// Option key accepted by the create command.
+ pub key: String,
+ /// Canonical kind the server persists the value under. A `String` value
+ /// parsed via the server-config rules is always accepted too.
+ pub kind: HeaderKind,
+ /// This node's current resolved default in the canonical kind's
+ /// encoding; empty when the key has no server default.
+ #[serde(default)]
+ pub default_value: Vec<u8>,
+ /// Human-readable description.
+ #[serde(default)]
+ pub description: String,
+}
+
+/// Every key `CreateTopic` currently accepts. Unknown keys are rejected at
+/// the edge, never skipped: a silently ignored knob would hand the client
+/// server defaults without it ever learning.
+pub const TOPIC_OPTION_KEYS: &[&str] = &[
+ topic_option_keys::COMPRESSION_ALGORITHM,
+ topic_option_keys::MESSAGE_EXPIRY,
+ topic_option_keys::MAX_TOPIC_SIZE,
+ topic_option_keys::SEGMENT_SIZE,
+ topic_option_keys::ENFORCE_FSYNC,
+ topic_option_keys::MESSAGES_REQUIRED_TO_SAVE,
+ topic_option_keys::SIZE_OF_MESSAGES_REQUIRED_TO_SAVE,
+ topic_option_keys::PREALLOCATE_SEGMENTS,
+];
+
+/// The subset of [`TOPIC_OPTION_KEYS`] an `UpdateTopic` options block may
+/// carry.
+///
+/// These three used to be fixed fields of the update command, which meant one
+/// setting had two homes and an update always rewrote all three whether the
+/// caller meant to or not. As options they are patched: a key the client did
+/// not send keeps its current value.
+///
+/// The partition runtime knobs (`segment_size`, `enforce_fsync`, both flush
+/// thresholds, `preallocate_segments`) stay out, and not only because nothing
+/// re-pushes them to a live partition. They describe how a partition's storage
+/// was laid down: changing `segment_size` mid-segment leaves one segment sized
+/// by the old cap and the next by the new one, and `preallocate_segments` can
+/// only act on a file not yet opened. A topic gets them at creation and keeps
+/// them, so its segments stay uniform.
+pub const UPDATABLE_TOPIC_OPTION_KEYS: &[&str] = &[
+ topic_option_keys::COMPRESSION_ALGORITHM,
+ topic_option_keys::MESSAGE_EXPIRY,
+ topic_option_keys::MAX_TOPIC_SIZE,
+];
+
+/// Keys an `UpdateStream` options block may carry. Empty because streams have
+/// no catalog keys yet: the block exists so the first one costs a catalog
+/// entry instead of another wire change, and until then every key is rejected
+/// by name rather than stored and ignored.
+pub const UPDATABLE_STREAM_OPTION_KEYS: &[&str] = &[];
+
+/// Keys an `UpdateUser` options block may carry. Empty for the same reason as
+/// [`UPDATABLE_STREAM_OPTION_KEYS`].
+pub const UPDATABLE_USER_OPTION_KEYS: &[&str] = &[];
+
+/// Longest key echoed back in an option error. A rejected key is attacker-sized
+/// by definition, and the error rides the wire back to the client.
+const ERROR_KEY_PREVIEW_LEN: usize = 64;
+
+fn key_preview(key: &str) -> String {
+ key.chars().take(ERROR_KEY_PREVIEW_LEN).collect()
+}
+
+/// Build an options map from string key-values, all marked explicit.
+///
+/// Shared by the update-options types: their keys are all client-sent, so none
+/// of the derived-provenance handling that create needs applies. Callers that
+/// also have typed fields insert those afterwards, so a typed value wins on
+/// collision and keeps its canonical kind.
+///
+/// # Errors
+///
+/// `UnsupportedOptionKey` when a key is empty or over 255 bytes,
+/// `InvalidOptionValue` when a value is. Both bounds come from the
+/// header-field codec these entries ride.
+fn raw_options_map(raw: &BTreeMap<String, String>) -> Result<ResourceOptions, IggyError> {
+ let mut options = ResourceOptions::new();
+ for (key, value) in raw {
+ let header_key = HeaderKey::from_str(key)
+ .map_err(|_| IggyError::UnsupportedOptionKey(key_preview(key)))?;
+ let header_value = HeaderValue::try_from(value.as_str())
+ .map_err(|_| IggyError::InvalidOptionValue(key_preview(key)))?;
+ options.insert(header_key, OptionValue::explicit(header_value));
+ }
+ Ok(options)
+}
+
+/// Encode string key-values into an options block, all marked explicit.
+///
+/// # Errors
+///
+/// See [`raw_options_map`].
+fn raw_options_to_wire(raw: &BTreeMap<String, String>) -> Result<WireOptions, IggyError> {
+ crate::wire_conversions::resource_options_to_wire(
+ &raw_options_map(raw)?,
+ OptionsProvenance::All,
+ )
+}
+
+/// The options an `UpdateStream` may carry.
+///
+/// Absent keys leave the stream's existing values alone -- an update patches
+/// the option map, it does not replace it. See [`TopicUpdateOptions`] for why.
+#[derive(Debug, Clone, Default, PartialEq)]
+pub struct StreamUpdateOptions {
+ /// Keys sent as `String` values, checked against
+ /// [`UPDATABLE_STREAM_OPTION_KEYS`] server-side. That list is empty, so
+ /// every key is currently refused by name; the field exists so the first
+ /// updatable stream key costs a catalog entry, not a wire change.
+ pub raw: BTreeMap<String, String>,
+}
+
+impl StreamUpdateOptions {
+ /// Encode the present keys into an options block.
+ ///
+ /// # Errors
+ ///
+ /// See [`raw_options_to_wire`].
+ pub fn to_wire(&self) -> Result<WireOptions, IggyError> {
+ raw_options_to_wire(&self.raw)
+ }
+}
+
+/// The options an `UpdateUser` may carry. Same patch semantics as
+/// [`StreamUpdateOptions`].
+#[derive(Debug, Clone, Default, PartialEq)]
+pub struct UserUpdateOptions {
+ /// Keys sent as `String` values, checked against
+ /// [`UPDATABLE_USER_OPTION_KEYS`] server-side. That list is empty, so every
+ /// key is currently refused by name; the field exists so the first
+ /// updatable user key costs a catalog entry, not a wire change.
+ pub raw: BTreeMap<String, String>,
+}
+
+impl UserUpdateOptions {
+ /// Encode the present keys into an options block.
+ ///
+ /// # Errors
+ ///
+ /// See [`raw_options_to_wire`].
+ pub fn to_wire(&self) -> Result<WireOptions, IggyError> {
+ raw_options_to_wire(&self.raw)
+ }
+}
+
+/// This node's configured fallbacks for the runtime knobs, used to fill the
+/// derived-options block for keys a client did not send.
+#[derive(Debug, Clone, Copy, PartialEq)]
+pub struct TopicRuntimeDefaults {
+ pub segment_size: IggyByteSize,
+ pub enforce_fsync: bool,
+ pub messages_required_to_save: u32,
+ pub size_of_messages_required_to_save: IggyByteSize,
+ pub preallocate_segments: bool,
+}
+
+/// A topic's resolved runtime knobs, as carried from the metadata plane to
+/// each of its partitions. `None` means "keep the shard-wide configured
+/// value": topics created without an options block (simulator, unit tests)
+/// have no resolved values to carry.
+#[derive(Debug, Clone, Copy, Default, PartialEq)]
+pub struct TopicRuntimeOptions {
+ pub segment_size: Option<IggyByteSize>,
+ pub enforce_fsync: Option<bool>,
+ pub messages_required_to_save: Option<u32>,
+ pub size_of_messages_required_to_save: Option<IggyByteSize>,
+ pub preallocate_segments: Option<bool>,
+}
+
+impl TopicRuntimeOptions {
+ /// Derive the runtime knobs from a topic's persisted options map.
+ ///
+ /// Degrades per key, not per map. An entry this build cannot interpret
+ /// leaves its own knob unset and every other knob intact, so one key a
+ /// newer node wrote cannot silently drop a topic's `enforce_fsync` or
+ /// reset its segment size along with it.
+ #[must_use]
+ pub fn from_resource_options(options: &ResourceOptions) -> Self {
+ let parsed = TopicCreateOptions::from_resource_options(options);
+ Self {
+ segment_size: parsed.segment_size,
+ enforce_fsync: parsed.enforce_fsync,
+ messages_required_to_save: parsed.messages_required_to_save,
+ size_of_messages_required_to_save: parsed.size_of_messages_required_to_save,
+ preallocate_segments: parsed.preallocate_segments,
+ }
+ }
+}
+
+/// The options an `UpdateTopic` may carry, as a type that cannot express a
+/// key the update path rejects (see [`UPDATABLE_TOPIC_OPTION_KEYS`]).
+///
+/// Separate from [`TopicCreateOptions`] on purpose: reusing that struct would
+/// let a caller set `segment_size` on an update, encode it, and only learn at
+/// the server that it was refused. Absent keys leave the topic's existing
+/// value alone -- an update patches the option map, it does not replace it.
+#[derive(Debug, Clone, Default, PartialEq)]
+pub struct TopicUpdateOptions {
+ /// `None` leaves the topic's current algorithm alone.
+ pub compression_algorithm: Option<CompressionAlgorithm>,
+ /// `None` leaves the topic's current expiry alone.
+ pub message_expiry: Option<IggyExpiry>,
+ /// `None` leaves the topic's current cap alone.
+ pub max_topic_size: Option<MaxTopicSize>,
+ /// Keys sent as `String` values, checked against
+ /// [`UPDATABLE_TOPIC_OPTION_KEYS`] server-side. Lets a client reach an
+ /// updatable key added to the catalog after this build shipped.
+ pub raw: BTreeMap<String, String>,
+}
+
+impl TopicUpdateOptions {
+ /// Encode the present keys into an options block, in canonical kinds.
+ ///
+ /// A typed field is inserted after the raw entries, so it wins on collision
+ /// and keeps its canonical kind.
+ ///
+ /// # Errors
+ ///
+ /// See [`raw_options_map`].
+ pub fn to_wire(&self) -> Result<WireOptions, IggyError> {
+ let mut options = raw_options_map(&self.raw)?;
+ if let Some(compression_algorithm) = self.compression_algorithm {
+ options.insert(
+ HeaderKey::from_str(topic_option_keys::COMPRESSION_ALGORITHM)
+ .expect("catalog key is a valid header key"),
+ OptionValue::explicit(
+ HeaderValue::try_from(compression_algorithm.to_string().as_str())
+ .expect("compression name fits a header value"),
+ ),
+ );
+ }
+ if let Some(message_expiry) = self.message_expiry {
+ options.insert(
+ HeaderKey::from_str(topic_option_keys::MESSAGE_EXPIRY)
+ .expect("catalog key is a valid header key"),
+ OptionValue::explicit(HeaderValue::from(u64::from(message_expiry))),
+ );
+ }
+ if let Some(max_topic_size) = self.max_topic_size {
+ options.insert(
+ HeaderKey::from_str(topic_option_keys::MAX_TOPIC_SIZE)
+ .expect("catalog key is a valid header key"),
+ OptionValue::explicit(HeaderValue::from(u64::from(max_topic_size))),
+ );
+ }
+ crate::wire_conversions::resource_options_to_wire(&options, OptionsProvenance::All)
+ }
+}
+
+/// Typed view of the known topic option keys, parsed from a wire block.
+///
+/// `None` means the key was absent, which always means "resolve from server
+/// defaults at admission". Values that parse to their type's `ServerDefault`
+/// sentinel are normalized to `None` for the same reason.
+#[derive(Debug, Clone, Default, PartialEq)]
+pub struct TopicCreateOptions {
+ /// Partitions to allocate. NOT an option key: it fills the `CreateTopic`
+ /// command's own fixed field, because it is an argument to the operation
+ /// rather than a property of the topic (admission consumes it to compute
+ /// assignments, and a stored count would go stale on the first
+ /// `CreatePartitions`). Carried here so callers pass one bundle.
+ /// `None` means [`DEFAULT_PARTITIONS_COUNT`].
+ pub partitions_count: Option<u32>,
+ pub compression_algorithm: Option<CompressionAlgorithm>,
+ pub message_expiry: Option<IggyExpiry>,
+ pub max_topic_size: Option<MaxTopicSize>,
+ /// Per-topic segment size; `None` resolves against `[system.segment]
+ /// size` at admission. `0` is normalized to `None`.
+ pub segment_size: Option<IggyByteSize>,
+ /// Per-topic fsync enforcement; `None` resolves against
+ /// `[system.partition] enforce_fsync`.
+ pub enforce_fsync: Option<bool>,
+ /// Per-topic message-count flush threshold; `None` resolves against
+ /// `[system.partition] messages_required_to_save`. `0` is rejected.
+ pub messages_required_to_save: Option<u32>,
+ /// Per-topic byte flush threshold; `None` resolves against
+ /// [`DEFAULT_SIZE_OF_MESSAGES_REQUIRED_TO_SAVE`].
+ pub size_of_messages_required_to_save: Option<IggyByteSize>,
+ /// Whether this topic's segments reserve their bytes on open; `None`
+ /// resolves against [`DEFAULT_PREALLOCATE_SEGMENTS`].
+ pub preallocate_segments: Option<bool>,
+ /// String-valued keys with no typed field above, parsed server-side
+ /// through each key's `FromStr`. Lets a client reach a key added to the
+ /// server catalog after this build shipped. Outbound only: a typed field
+ /// wins on collision in [`Self::to_wire`], and [`Self::parse`] never
+ /// populates it.
+ pub raw: BTreeMap<String, String>,
+}
+
+impl TopicCreateOptions {
+ /// Parse a wire options block against the topic catalog.
+ ///
+ /// # Errors
+ ///
+ /// `UnsupportedOptionKey` for a key outside [`TOPIC_OPTION_KEYS`];
+ /// `InvalidOptionValue` when a value has the wrong kind or fails its
+ /// type's `FromStr`.
+ pub fn parse(options: &WireOptions) -> Result<Self, IggyError> {
+ let mut parsed = Self::default();
+ for entry in options {
+ // Wire validation already enforced UTF-8 string keys.
+ let key = String::from_utf8_lossy(entry.key);
+ parsed.absorb(&entry, &key, UnknownEntry::Reject)?;
+ }
+ Ok(parsed)
+ }
+
+ /// Parse a block that is already COMMITTED, skipping what this build
+ /// cannot interpret.
+ ///
+ /// Admission gated the block against the catalog on the primary, once.
+ /// Re-running that gate at apply would make the verdict depend on the
+ /// build running it, so a replica that predates a key would reject an
+ /// operation its peers accepted and diverge from the group permanently.
+ /// The catalog belongs at the edge; apply only reads what it knows.
+ #[must_use]
+ pub fn parse_committed(options: &WireOptions) -> Self {
+ let mut parsed = Self::default();
+ for entry in options {
+ let key = String::from_utf8_lossy(entry.key);
+ // Infallible: `Skip` turns every parse error into a no-op.
+ let _ = parsed.absorb(&entry, &key, UnknownEntry::Skip);
+ }
+ parsed
+ }
+
+ /// Fold one catalog entry into `self`, applying `unknown` to an entry this
+ /// build cannot interpret. Shared by [`Self::parse`] (wire block) and
+ /// [`Self::from_resource_options`] (persisted map) so both read the
+ /// identical key set, kinds, and value bounds, and differ only in what
+ /// they do with an entry outside it.
+ fn absorb(
+ &mut self,
+ entry: &WireUserHeaderEntry<'_>,
+ key: &str,
+ unknown: UnknownEntry,
+ ) -> Result<(), IggyError> {
+ // Every arm assigns only after its parse succeeds, so a skipped entry
+ // leaves `self` exactly as it found it.
+ match self.absorb_strict(entry, key) {
+ Ok(()) => Ok(()),
+ Err(error) => match unknown {
+ UnknownEntry::Reject => Err(error),
+ UnknownEntry::Skip => Ok(()),
+ },
+ }
+ }
+
+ fn absorb_strict(
+ &mut self,
+ entry: &WireUserHeaderEntry<'_>,
+ key: &str,
+ ) -> Result<(), IggyError> {
+ let parsed = self;
+ {
+ match key {
+ topic_option_keys::COMPRESSION_ALGORITHM => {
+ let value = parse_compression(entry, key)?;
+ parsed.compression_algorithm = Some(value);
+ }
+ topic_option_keys::MESSAGE_EXPIRY => {
+ let expiry = IggyExpiry::from(parse_u64_or(entry, key, IggyExpiry::from_str)?);
+ parsed.message_expiry = (expiry != IggyExpiry::ServerDefault).then_some(expiry);
+ }
+ topic_option_keys::MAX_TOPIC_SIZE => {
+ let size =
+ MaxTopicSize::from(parse_u64_or(entry, key, MaxTopicSize::from_str)?);
+ parsed.max_topic_size = (size != MaxTopicSize::ServerDefault).then_some(size);
+ }
+ topic_option_keys::SEGMENT_SIZE => {
+ let size = parse_byte_size(entry, key)?;
+ parsed.segment_size = (size != 0).then_some(IggyByteSize::from(size));
+ }
+ topic_option_keys::ENFORCE_FSYNC => {
+ parsed.enforce_fsync = Some(parse_bool(entry, key)?);
+ }
+ topic_option_keys::MESSAGES_REQUIRED_TO_SAVE => {
Review Comment:
only zero is rejected for the two flush thresholds - no upper bound anywhere, and `validate_topic_bounds` doesn't cover them. maxed out, the flush gate never trips and committed messages sit in the journal (which doesn't survive a crash) up to a full segment per partition. these used to be operator-only config; now anyone with create_topic can set them. add ceilings.
##########
core/binary_protocol/src/responses/system/describe_options.rs:
##########
@@ -0,0 +1,202 @@
+// 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::WireError;
+use crate::codec::{WireDecode, WireEncode, read_u8, read_u32_le};
+use crate::primitives::identifier::WireName;
+use bytes::{BufMut, Bytes, BytesMut};
+
+/// One catalog entry in a `DescribeOptions` response.
+///
+/// Wire format:
+/// ```text
+/// [key_len:u8][key:N][kind:u8]
+/// [default_len:u32][default bytes][description_len:u32][description]
+/// ```
+///
+/// `kind` is the canonical header kind code the server persists the value
+/// under; clients may also send a `String` value parsed via the same rules as
+/// the server config file. `default` is the server's current resolved default
+/// in the canonical kind's encoding, empty when the key has no server default.
+#[derive(Debug, Clone, PartialEq, Eq)]
+pub struct OptionDescriptor {
+ pub key: WireName,
+ pub kind: u8,
+ pub default_value: Bytes,
+ pub description: String,
+}
+
+impl WireEncode for OptionDescriptor {
+ fn encoded_size(&self) -> usize {
+ self.key.encoded_size() + 1 + 4 + self.default_value.len() + 4 + self.description.len()
+ }
+
+ fn encode(&self, buf: &mut BytesMut) {
+ self.key.encode(buf);
+ buf.put_u8(self.kind);
+ #[allow(clippy::cast_possible_truncation)]
+ buf.put_u32_le(self.default_value.len() as u32);
+ buf.put_slice(&self.default_value);
+ #[allow(clippy::cast_possible_truncation)]
+ buf.put_u32_le(self.description.len() as u32);
+ buf.put_slice(self.description.as_bytes());
+ }
+}
+
+impl WireDecode for OptionDescriptor {
+ fn decode(buf: &[u8]) -> Result<(Self, usize), WireError> {
+ let (key, mut pos) = WireName::decode(buf)?;
+ let kind = read_u8(buf, pos)?;
+ pos += 1;
+ let default_len = read_u32_le(buf, pos)? as usize;
+ pos += 4;
+ let default_end = pos + default_len;
Review Comment:
`pos + default_len` unchecked (same at line 79) - on a 32-bit target a wire u32 wraps, the `buf.len()` guard passes and the slice panics. `decode_options_prefixed` guards the identical pattern with `checked_add` and a comment explaining why. no 32-bit target ships today, so consistency rather than a live bug - but mirror the guard.
##########
foreign/go/contracts/compression_algorithm.go:
##########
@@ -23,3 +23,12 @@ const (
CompressionAlgorithmNone CompressionAlgorithm = 1
CompressionAlgorithmGzip CompressionAlgorithm = 2
)
+
+func (c CompressionAlgorithm) String() string {
+ switch c {
+ case CompressionAlgorithmGzip:
+ return "gzip"
+ default:
Review Comment:
`String()` returns "none" for every non-gzip value including invalid codes, and the gates in `topic.go` are `!= "none"` - so an out-of-range algorithm is silently dropped client-side and the topic gets the server default. before, the raw byte went on the wire and the server rejected it. switch on the enum and error on unknown codes.
##########
core/metadata/src/impls/metadata.rs:
##########
@@ -758,6 +760,22 @@ pub struct IggyMetadata<C, J, S, M, SB = PingPongSuperblock> {
/// set from server config at bootstrap ([`Self::set_default_message_expiry`]).
/// Defaults to never-expire, matching the shipped server config.
default_message_expiry: Cell<u64>,
+ /// Resolved byte value an absent `segment_size` option takes at
+ /// admission. Same resolution discipline as the two defaults above; set
+ /// from `system.segment.size` at bootstrap
+ /// ([`Self::set_default_segment_size`]). Defaults to 1 GiB, matching the
+ /// shipped server config.
+ default_segment_size: Cell<u64>,
Review Comment:
all eight of these cells (and their setters/getters) only ever transport compile-time constants now - bootstrap seeds each with the same `iggy_common::DEFAULT_*` the `Cell::new` already holds, and the segment ceiling collapses to a constant too since config validation refuses boot unless `transfer_artifact_bytes_max - max_message_size >= SEGMENT_MAX_SIZE_BYTES`. reading the consts at the use sites drops ~110 lines. the docs here still say "set from server config at bootstrap", which stopped being true when those config keys were removed.
##########
core/server/src/bootstrap.rs:
##########
@@ -1127,8 +1129,28 @@ async fn shard_main(
metadata.seed_checkpoint_ref(checkpoint_seed.0, checkpoint_seed.1);
// Shard 0's copy resolves the `ServerDefault` sentinels (max topic size and
// message expiry) at create admission; responses echo stored values verbatim.
- metadata.set_default_max_topic_size(config.system.topic.max_size.as_bytes_u64());
- metadata.set_default_message_expiry(u64::from(config.system.topic.message_expiry));
+ metadata.set_default_max_topic_size(iggy_common::DEFAULT_MAX_TOPIC_SIZE);
+ metadata.set_default_message_expiry(iggy_common::DEFAULT_MESSAGE_EXPIRY);
+ metadata.set_default_segment_size(iggy_common::DEFAULT_SEGMENT_SIZE);
+ // Per-topic segment ceiling: the smaller of the global segment maximum
+ // and the state-transfer artifact budget minus one bus frame (a segment
+ // may close one whole batch past its cap; an artifact ceiling below that
+ // refuses a legal segment and livelocks the partition's rejoin).
+ metadata.set_max_topic_segment_size(
+ configs::validators::SEGMENT_MAX_SIZE_BYTES.min(
Review Comment:
this `.min()` is dead - config validation refuses boot unless `transfer_artifact_bytes_max >= SEGMENT_MAX_SIZE_BYTES + max_message_size`, so the subtraction is always >= the cap and this always yields `SEGMENT_MAX_SIZE_BYTES`. pass the constant and keep a comment pointing at the validator, since this is the only runtime expression of that coupling.
##########
core/cli/src/args/topic.rs:
##########
@@ -117,16 +117,21 @@ pub(crate) struct TopicCreateArgs {
/// Can't be lower than segment size in the config.
#[arg(short, long, default_value = "server_default", verbatim_doc_comment)]
pub(crate) max_topic_size: MaxTopicSize,
- /// Replication factor for the topic
- #[arg(short, long, default_value = "1")]
- pub(crate) replication_factor: u8,
/// Message expiry time in human-readable format like "unlimited" or "15days 2min 2s"
///
/// "server_default" or skipping parameter makes CLI to use server default (from current server config) expiry time
#[arg(default_value = "server_default", value_parser = clap::value_parser!(IggyExpiry), verbatim_doc_comment)]
pub(crate) message_expiry: Vec<IggyExpiry>,
+ /// Additional topic option as key=value, repeatable
+ ///
+ /// Values are sent as strings and parsed server-side through each option's
+ /// own FromStr (e.g. --set segment_size=128MiB). The server rejects keys it
+ /// does not support; GET /options/topic lists the ones it accepts.
Review Comment:
this help text points binary-transport cli users at an http endpoint. the sdk has `describe_options` but no cli subcommand calls it - an `iggy options topic` command would make this text honest. the same string is pinned in the cli help snapshot test, so both need the edit.
##########
foreign/java/java-sdk/src/main/java/org/apache/iggy/serde/BytesDeserializer.java:
##########
@@ -134,11 +141,12 @@ public static Topic readTopic(ByteBuf response) {
var messageExpiry = readU64AsBigInteger(response);
var compressionAlgorithmCode = response.readByte();
var maxTopicSize = readU64AsBigInteger(response);
- var replicationFactor = response.readByte();
var size = readU64AsBigInteger(response);
var messagesCount = readU64AsBigInteger(response);
var nameLength = response.readByte();
var name = response.readCharSequence(nameLength, StandardCharsets.UTF_8).toString();
+ skipOptionsBlock(response, "topic explicit options");
Review Comment:
java skips the options blocks everywhere (also 96, 149, 523) and the records carry no options field, so the java client can't read segment_size, enforce_fsync or anything else this PR adds - and the tcp client can only set the three legacy keys. fine as a phased rollout, but file an issue so it isn't silent; today java is blind to the feature in both directions.
##########
foreign/csharp/Iggy_SDK/Utils/CommandCodes.cs:
##########
@@ -23,6 +23,7 @@ internal static class CommandCodes
internal const int GET_STATS_CODE = 10;
internal const int GET_SNAPSHOT_CODE = 11;
internal const int GET_CLUSTER_METADATA_CODE = 12;
+ internal const int DESCRIBE_OPTIONS_CODE = 13;
Review Comment:
dead constant - c# has no describe_options call (same for `DescribeOptions` in node's `command.code.ts`). wire up the call or drop the constants until it exists.
##########
core/metadata/src/stm/snapshot.rs:
##########
@@ -562,6 +573,60 @@ mod tests {
);
}
+ #[test]
Review Comment:
pins topic and stream shapes, but `UserSnapshot` also gained `options` in this PR and isn't pinned - add it so the next user-snapshot edit trips the test the same way.
##########
core/binary_protocol/src/codec.rs:
##########
@@ -263,6 +263,21 @@ pub fn capped_capacity(count: usize, remaining: usize, min_item_size: usize) ->
count.min(remaining / min_item_size)
}
+/// Capacity to preallocate for `count` elements that still have to be decoded
+/// out of `remaining` bytes, each at least `min_element_size` bytes on the
+/// wire.
+///
+/// A count read off the wire is chosen by the peer, and `Vec::with_capacity`
+/// reserves for it before a single element has been bounds-checked. A hostile
+/// or corrupt frame claiming `u32::MAX` elements therefore aborts the process
+/// on allocation failure instead of returning a decode error. The bytes left
+/// to decode are the honest ceiling on how many elements can exist.
+#[must_use]
+#[inline]
+pub fn bounded_capacity(count: usize, remaining: usize, min_element_size: usize) -> usize {
Review Comment:
this duplicates `capped_capacity` above - same expression, differing only when `min_item_size == 0`, where the old one returns the raw wire count unbounded (and a test pins that). one helper should survive, with the `.max(1)` body - the zero branch is the exact hole this fn's doc warns about. fold them and invert the zero-size test to pin the bounded behavior.
##########
core/cli/src/commands/binary_topics/update_topic.rs:
##########
@@ -21,39 +21,47 @@ use async_trait::async_trait;
use core::fmt;
use iggy_common::Client;
use iggy_common::update_topic::UpdateTopic;
-use iggy_common::{CompressionAlgorithm, Identifier, IggyExpiry, MaxTopicSize};
+use iggy_common::{CompressionAlgorithm, Identifier, IggyExpiry, MaxTopicSize, TopicUpdateOptions};
+use std::collections::BTreeMap;
use tracing::{Level, event};
pub struct UpdateTopicCmd {
update_topic: UpdateTopic,
+ compression_algorithm: CompressionAlgorithm,
message_expiry: IggyExpiry,
max_topic_size: MaxTopicSize,
- replication_factor: u8,
+ options: TopicUpdateOptions,
}
impl UpdateTopicCmd {
+ #[allow(clippy::too_many_arguments)]
Review Comment:
the arg count is 6, under clippy's threshold of 7 - this allow is dead, remove it.
##########
core/common/src/wire_conversions.rs:
##########
@@ -799,6 +813,144 @@ pub(crate) fn user_headers_from_validated_slice(
Ok(headers)
}
+// -- Options conversions --
+
+/// Decode a `DescribeOptions` response into domain option specs.
+///
+/// # Errors
+///
+/// Returns `IggyError::InvalidHeaderKind` when an entry carries an unknown
+/// canonical kind code.
+pub fn option_specs_from_wire(
+ wire: iggy_binary_protocol::responses::system::DescribeOptionsResponse,
+) -> Result<Vec<OptionSpec>, IggyError> {
+ wire.entries
+ .into_iter()
+ .map(|entry| {
+ Ok(OptionSpec {
+ key: entry.key.to_string(),
+ kind: HeaderKind::from_code(entry.kind)?,
+ default_value: entry.default_value.to_vec(),
+ description: entry.description,
+ })
+ })
+ .collect()
+}
+
+/// Decode a [`WireOptions`] block into domain resource options.
+///
+/// Every decoded entry is marked with the given `explicit` flag; the wire
+/// block carries no per-entry provenance. Admission calls this once for the
+/// client block (`explicit == true`) and fills defaults separately.
+///
+/// # Errors
+///
+/// Returns `IggyError::InvalidHeaderKey` / `InvalidHeaderValue` when a kind
+/// code has no domain meaning or a fixed-size kind carries a mismatched
+/// payload. The wire layer accepts unknown kinds for forwarding; interpreting
+/// them into the domain requires known semantics.
+pub fn resource_options_from_wire(
+ wire: &iggy_binary_protocol::WireOptions,
+ explicit: bool,
+) -> Result<ResourceOptions, IggyError> {
+ let headers = user_headers_from_validated_slice(wire.as_bytes())?;
+ Ok(headers
+ .into_iter()
+ .map(|(key, value)| (key, OptionValue { value, explicit }))
+ .collect())
+}
+
+/// Merge response-side `(explicit, derived)` wire blocks back into domain
+/// resource options, restoring per-key provenance. Explicit wins on a key
+/// collision (which a well-formed response never produces).
+///
+/// # Errors
+///
+/// Same contract as [`resource_options_from_wire`].
+pub fn resource_options_from_wire_split(
+ explicit: &iggy_binary_protocol::WireOptions,
+ derived: &iggy_binary_protocol::WireOptions,
+) -> Result<ResourceOptions, IggyError> {
+ let mut options = resource_options_from_wire(derived, false)?;
+ options.extend(resource_options_from_wire(explicit, true)?);
+ Ok(options)
+}
+
+/// Split domain resource options into `(explicit, derived)` wire blocks,
+/// the response-side layout that preserves per-key provenance.
+///
+/// # Errors
+///
+/// Same contract as [`resource_options_to_wire`].
+pub fn resource_options_to_wire_split(
+ options: &ResourceOptions,
+) -> Result<(WireOptions, WireOptions), IggyError> {
+ Ok((
+ resource_options_to_wire(options, OptionsProvenance::Explicit)?,
+ resource_options_to_wire(options, OptionsProvenance::Derived)?,
+ ))
+}
+
+/// Fixed per-entry cost of the TLV encoding: a kind byte and a `u32` length
+/// for the key and for the value.
+const OPTION_ENTRY_OVERHEAD: usize = 2 * (1 + 4);
+
+/// Encode domain resource options into a [`WireOptions`] block, keeping only
+/// the entries matching `provenance`. Encoding just the explicit ones is what
+/// lets a client round-trip a create without pinning server defaults.
+///
+/// # Errors
+///
+/// Returns `IggyError::OptionsBlockTooLarge` when the selected entries exceed
+/// `MAX_OPTIONS` or `MAX_OPTIONS_BYTES`. [`WireOptions::from_validated`] skips
+/// revalidation, so the two caps have to hold here: a block written past them
+/// is one the receiving peer's `decode_options_prefixed` refuses, which would
+/// make the resource permanently unreadable rather than merely oversized.
+pub fn resource_options_to_wire(
+ options: &ResourceOptions,
+ provenance: OptionsProvenance,
+) -> Result<WireOptions, IggyError> {
+ let entries: Vec<(&HeaderKey, &OptionValue)> = options
+ .iter()
+ .filter(|(_, option)| option.matches(provenance))
+ .collect();
+ if entries.is_empty() {
+ return Ok(WireOptions::empty());
+ }
+ if entries.len() > MAX_OPTIONS as usize {
+ return Err(IggyError::OptionsBlockTooLarge(format!(
+ "{} entries, maximum {MAX_OPTIONS}",
+ entries.len()
+ )));
+ }
+ let size: usize = entries
+ .iter()
+ .map(|(key, option)| {
+ OPTION_ENTRY_OVERHEAD + key.as_bytes().len() + option.value.as_bytes().len()
+ })
+ .sum();
+ if size > MAX_OPTIONS_BYTES {
+ return Err(IggyError::OptionsBlockTooLarge(format!(
+ "{size} bytes, maximum {MAX_OPTIONS_BYTES}"
+ )));
+ }
+ let mut buf = BytesMut::with_capacity(size);
+ for (key, option) in entries {
+ buf.put_u8(key.kind().as_code());
+ #[allow(clippy::cast_possible_truncation)]
+ buf.put_u32_le(key.as_bytes().len() as u32);
+ buf.put_slice(key.as_bytes());
+ buf.put_u8(option.value.kind().as_code());
+ #[allow(clippy::cast_possible_truncation)]
+ buf.put_u32_le(option.value.as_bytes().len() as u32);
+ buf.put_slice(option.value.as_bytes());
+ }
+ // Structural validity holds by construction: entries come from valid
Review Comment:
this holds because every producer went through `validate_options`, not because of anything `HeaderKey` guarantees - and `from_validated` skips the check that would catch a violation, which would produce a block the receiving peer refuses. a `debug_assert!(key.kind() == HeaderKind::String)` in the loop would make it self-enforcing.
##########
core/binary_protocol/src/responses/system/describe_options.rs:
##########
@@ -0,0 +1,202 @@
+// 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::WireError;
+use crate::codec::{WireDecode, WireEncode, read_u8, read_u32_le};
+use crate::primitives::identifier::WireName;
+use bytes::{BufMut, Bytes, BytesMut};
+
+/// One catalog entry in a `DescribeOptions` response.
+///
+/// Wire format:
+/// ```text
+/// [key_len:u8][key:N][kind:u8]
+/// [default_len:u32][default bytes][description_len:u32][description]
+/// ```
+///
+/// `kind` is the canonical header kind code the server persists the value
+/// under; clients may also send a `String` value parsed via the same rules as
+/// the server config file. `default` is the server's current resolved default
+/// in the canonical kind's encoding, empty when the key has no server default.
+#[derive(Debug, Clone, PartialEq, Eq)]
+pub struct OptionDescriptor {
+ pub key: WireName,
+ pub kind: u8,
+ pub default_value: Bytes,
+ pub description: String,
+}
+
+impl WireEncode for OptionDescriptor {
+ fn encoded_size(&self) -> usize {
+ self.key.encoded_size() + 1 + 4 + self.default_value.len() + 4 + self.description.len()
+ }
+
+ fn encode(&self, buf: &mut BytesMut) {
+ self.key.encode(buf);
+ buf.put_u8(self.kind);
+ #[allow(clippy::cast_possible_truncation)]
+ buf.put_u32_le(self.default_value.len() as u32);
+ buf.put_slice(&self.default_value);
+ #[allow(clippy::cast_possible_truncation)]
+ buf.put_u32_le(self.description.len() as u32);
+ buf.put_slice(self.description.as_bytes());
+ }
+}
+
+impl WireDecode for OptionDescriptor {
+ fn decode(buf: &[u8]) -> Result<(Self, usize), WireError> {
+ let (key, mut pos) = WireName::decode(buf)?;
+ let kind = read_u8(buf, pos)?;
+ pos += 1;
+ let default_len = read_u32_le(buf, pos)? as usize;
+ pos += 4;
+ let default_end = pos + default_len;
+ if buf.len() < default_end {
+ return Err(WireError::UnexpectedEof {
+ offset: pos,
+ need: default_len,
+ have: buf.len().saturating_sub(pos),
+ });
+ }
+ let default_value = Bytes::copy_from_slice(&buf[pos..default_end]);
+ pos = default_end;
+ let description_len = read_u32_le(buf, pos)? as usize;
+ pos += 4;
+ let description_end = pos + description_len;
+ if buf.len() < description_end {
+ return Err(WireError::UnexpectedEof {
+ offset: pos,
+ need: description_len,
+ have: buf.len().saturating_sub(pos),
+ });
+ }
+ let description = String::from_utf8_lossy(&buf[pos..description_end]).into_owned();
+ pos = description_end;
+
+ Ok((
+ Self {
+ key,
+ kind,
+ default_value,
+ description,
+ },
+ pos,
+ ))
+ }
+}
+
+/// `DescribeOptions` response: count-prefixed catalog entries.
+///
+/// Wire format: `[count:u32][OptionDescriptor]*`
+#[derive(Debug, Clone, PartialEq, Eq)]
+pub struct DescribeOptionsResponse {
+ pub entries: Vec<OptionDescriptor>,
+}
+
+impl WireEncode for DescribeOptionsResponse {
+ fn encoded_size(&self) -> usize {
+ 4 + self
+ .entries
+ .iter()
+ .map(WireEncode::encoded_size)
+ .sum::<usize>()
+ }
+
+ fn encode(&self, buf: &mut BytesMut) {
+ #[allow(clippy::cast_possible_truncation)]
+ buf.put_u32_le(self.entries.len() as u32);
+ for entry in &self.entries {
+ entry.encode(buf);
+ }
+ }
+}
+
+/// Smallest a descriptor can encode as: a name, a kind byte, a
+/// length-prefixed default and a length-prefixed description.
+const MIN_DESCRIPTOR_SIZE: usize = 4 + 1 + 1 + 4 + 4;
Review Comment:
counts `WireName` as 4 bytes but its minimum is 2 (`1 + len`, len >= 1), so the floor is 11, not 14. overcounting only tightens the cap so it's safe, just wrong.
##########
core/binary_protocol/src/responses/topics/get_topics.rs:
##########
@@ -16,40 +16,57 @@
// under the License.
use crate::WireError;
-use crate::codec::{WireDecode, WireEncode};
+use crate::codec::{WireDecode, WireEncode, read_u32_le};
use crate::responses::streams::get_stream::TopicHeader;
-use bytes::BytesMut;
+use bytes::{BufMut, BytesMut};
-/// `GetTopics` response: sequential topic headers.
+/// `GetTopics` response: count-prefixed topic headers.
///
/// Wire format:
/// ```text
-/// [TopicHeader]*
+/// [topics_count:u32_le][TopicHeader]*
/// ```
///
-/// Empty payload means zero topics.
+/// The count prefix exists because a topic element carries variable-length
Review Comment:
the stated reason doesn't hold - `get_streams`/`get_users` keep run-to-end loops over elements that also carry variable-length options blocks, and they do detect short reads (the element decoder errors). the prefix is justified for pre-sizing; say that instead. also worth a rule somewhere: three list responses now ship with two framings and every sdk special-cases topics.
##########
core/common/src/types/options/mod.rs:
##########
@@ -0,0 +1,1295 @@
+// 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.
+
+//! Key-value options attached to streams, topics, and users at creation.
+//!
+//! Options reuse the typed [`HeaderKey`] / [`HeaderValue`] machinery from
+//! message user headers. The map MUST stay a `BTreeMap`: metadata snapshots
+//! require deterministic ordering across replicas, and a hash map would make
+//! snapshot bytes diverge per replica.
+
+use std::collections::BTreeMap;
+use std::str::FromStr;
+
+use iggy_binary_protocol::{WireOptions, WireUserHeaderEntry};
+use serde::{Deserialize, Serialize};
+
+use crate::types::compression::compression_algorithm::CompressionAlgorithm;
+use crate::types::message::{HeaderKey, HeaderKind, HeaderValue};
+use crate::{IggyByteSize, IggyError, IggyExpiry, MaxTopicSize};
+
+/// A single resolved option entry.
+#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
+pub struct OptionValue {
+ /// The effective value, resolved by the admitting primary at creation.
+ pub value: HeaderValue,
+ /// Whether the client explicitly sent this key. Derived entries
+ /// (`explicit == false`) were filled from server defaults at admission
+ /// and would have resolved differently under other server configs.
+ pub explicit: bool,
+}
+
+/// Which provenance classes an encoded options block carries.
+///
+/// Responses split a resource's options into two blocks so `GetTopic` can say
+/// which values the client chose and which admission filled in.
+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+pub enum OptionsProvenance {
+ /// Every entry, whatever its provenance.
+ All,
+ /// Only the keys the client sent.
+ Explicit,
+ /// Only the keys admission resolved from server defaults.
+ Derived,
+}
+
+impl OptionValue {
+ /// Whether this entry belongs in a block encoded for `provenance`.
+ #[must_use]
+ pub const fn matches(&self, provenance: OptionsProvenance) -> bool {
+ match provenance {
+ OptionsProvenance::All => true,
+ OptionsProvenance::Explicit => self.explicit,
+ OptionsProvenance::Derived => !self.explicit,
+ }
+ }
+
+ #[must_use]
+ pub fn explicit(value: HeaderValue) -> Self {
+ Self {
+ value,
+ explicit: true,
+ }
+ }
+
+ #[must_use]
+ pub fn derived(value: HeaderValue) -> Self {
+ Self {
+ value,
+ explicit: false,
+ }
+ }
+}
+
+/// Options attached to a stream, topic, or user, keyed by option name.
+pub type ResourceOptions = BTreeMap<HeaderKey, OptionValue>;
+
+/// JSON form for [`ResourceOptions`].
+///
+/// [`HeaderKey`] serializes as a struct (`kind` + base64 `value`), and JSON
+/// object keys must be strings, so the derived impl fails outright with
+/// "key must be a string" the moment a map is non-empty. Every HTTP response
+/// carrying options goes through here instead, rendering the key as its plain
+/// text. Binary transports are unaffected: they never touch serde.
+pub mod resource_options_json {
+ use super::{HeaderKey, OptionValue, ResourceOptions};
+ use serde::de::Error as _;
+ use serde::{Deserialize, Deserializer, Serialize, Serializer};
+ use std::collections::BTreeMap;
+ use std::str::FromStr;
+
+ /// # Errors
+ ///
+ /// Propagates the serializer's own errors.
+ pub fn serialize<S: Serializer>(
+ options: &ResourceOptions,
+ serializer: S,
+ ) -> Result<S::Ok, S::Error> {
+ let readable: BTreeMap<String, &OptionValue> = options
+ .iter()
+ .map(|(key, value)| (String::from_utf8_lossy(key.as_bytes()).into_owned(), value))
+ .collect();
+ readable.serialize(serializer)
+ }
+
+ /// # Errors
+ ///
+ /// Returns a deserializer error when a key is not a valid option name.
+ pub fn deserialize<'de, D: Deserializer<'de>>(
+ deserializer: D,
+ ) -> Result<ResourceOptions, D::Error> {
+ let readable = BTreeMap::<String, OptionValue>::deserialize(deserializer)?;
+ readable
+ .into_iter()
+ .map(|(key, value)| {
+ HeaderKey::from_str(&key)
+ .map(|key| (key, value))
+ .map_err(D::Error::custom)
+ })
+ .collect()
+ }
+}
+
+/// Topic option catalog: the keys `CreateTopic` accepts.
+pub mod topic_option_keys {
+ /// Compression algorithm name (`none`, `gzip`).
+ pub const COMPRESSION_ALGORITHM: &str = "compression_algorithm";
+ /// Message expiry: `Uint64` micros or a humantime string (`7 days`).
+ pub const MESSAGE_EXPIRY: &str = "message_expiry";
+ /// Topic size cap: `Uint64` bytes or a byte-size string (`1 GiB`).
+ pub const MAX_TOPIC_SIZE: &str = "max_topic_size";
+ /// Segment size: `Uint64` bytes or a byte-size string (`128 MiB`).
+ /// Bounded: a 512-byte multiple within
+ /// [`super::MIN_TOPIC_SEGMENT_SIZE`]..=the server's segment ceiling.
+ pub const SEGMENT_SIZE: &str = "segment_size";
+ /// Whether writes to this topic's partitions fsync: `Bool`, or the
+ /// strings `true` / `false`.
+ pub const ENFORCE_FSYNC: &str = "enforce_fsync";
+ /// Flush the journal once it holds this many messages: `Uint32`.
+ /// Must be non-zero.
+ pub const MESSAGES_REQUIRED_TO_SAVE: &str = "messages_required_to_save";
+ /// Flush the journal once it holds this many bytes: `Uint64` or a
+ /// byte-size string. Paired with [`Self::MESSAGES_REQUIRED_TO_SAVE`];
+ /// whichever threshold trips first flushes.
+ pub const SIZE_OF_MESSAGES_REQUIRED_TO_SAVE: &str = "size_of_messages_required_to_save";
+ /// Reserve the segment's bytes up front on a filesystem that supports it:
+ /// `Bool`, or the strings `true` / `false`. Pairs with
+ /// [`Self::SEGMENT_SIZE`] -- preallocation reserves exactly that much, so
+ /// the two only make sense decided together.
+ pub const PREALLOCATE_SEGMENTS: &str = "preallocate_segments";
+}
+
+/// Values an absent topic option resolves to at admission.
+///
+/// These are the knobs' single source of truth: they used to live in
+/// `config.toml` (`[system.topic]`, `[system.partition]`, `[system.segment]`),
+/// which meant every one of them had two homes and an operator could not tell
+/// which won. A topic carries whatever it was created with; anything the
+/// client did not send resolves to the constant here and is persisted as a
+/// derived entry, so the effective value is always visible on `GetTopic`.
+///
+/// Each value matches what the shipped `config.toml` carried, so removing the
+/// keys changed no behavior for a topic created without options.
+pub const DEFAULT_PARTITIONS_COUNT: u32 = 1;
+/// `MaxTopicSize::Unlimited` (was `[system.topic] max_size = "unlimited"`).
+pub const DEFAULT_MAX_TOPIC_SIZE: u64 = u64::MAX;
+/// `IggyExpiry::NeverExpire` (was `[system.topic] message_expiry = "none"`).
+pub const DEFAULT_MESSAGE_EXPIRY: u64 = u64::MAX;
+/// 1 GiB (was `[system.segment] size = "1 GiB"`).
+pub const DEFAULT_SEGMENT_SIZE: u64 = 1024 * 1024 * 1024;
+/// Was `[system.partition] enforce_fsync = false`.
+pub const DEFAULT_ENFORCE_FSYNC: bool = false;
+/// Was `[system.partition] messages_required_to_save = 1024`.
+pub const DEFAULT_MESSAGES_REQUIRED_TO_SAVE: u32 = 1024;
+/// Opt-in, unlike the `[system.segment] preallocate = true` this replaced.
+///
+/// That default was never actually in force: the reservation ran through
+/// `compio::spawn_blocking`, which panics the shard because shard executors
+/// disable the blocking pool, so any deployment that worked at all had
+/// preallocation off. With the call fixed to run inline it reserves real
+/// extents, and `FALLOC_FL_KEEP_SIZE` against the 1 GiB default segment size
+/// means 1 GiB of disk per partition the moment it is created -- a full test
+/// sweep reserved 393 GB before this was flipped. A topic that wants the
+/// latency benefit asks for it with `preallocate_segments`.
+pub const DEFAULT_PREALLOCATE_SEGMENTS: bool = false;
+/// 1 MiB (was `[system.partition] size_of_messages_required_to_save`).
+pub const DEFAULT_SIZE_OF_MESSAGES_REQUIRED_TO_SAVE: u64 = 1024 * 1024;
+
+/// Every runtime knob at its default, for a partition built with no resolved
+/// topic options (simulator, unit tests).
+impl Default for TopicRuntimeDefaults {
+ fn default() -> Self {
+ Self {
+ segment_size: IggyByteSize::from(DEFAULT_SEGMENT_SIZE),
+ enforce_fsync: DEFAULT_ENFORCE_FSYNC,
+ messages_required_to_save: DEFAULT_MESSAGES_REQUIRED_TO_SAVE,
+ size_of_messages_required_to_save: IggyByteSize::from(
+ DEFAULT_SIZE_OF_MESSAGES_REQUIRED_TO_SAVE,
+ ),
+ preallocate_segments: DEFAULT_PREALLOCATE_SEGMENTS,
+ }
+ }
+}
+
+/// Smallest per-topic segment size. Segments far below the shipped default
+/// explode the per-partition segment count, which state transfer bounds via
+/// its manifest entry cap; a partition that crosses it becomes unservable
+/// for transfer, so the floor is enforced at admission.
+pub const MIN_TOPIC_SEGMENT_SIZE: u64 = 1024 * 1024;
+
+/// Validate an explicit per-topic `segment_size` against its bounds.
+///
+/// `ceiling` is node-derived: the smaller of the global segment maximum and
+/// the state-transfer artifact budget minus one bus frame (a segment may
+/// close one whole batch past its cap; an artifact ceiling below that
+/// refuses a legal segment and livelocks the partition's rejoin).
+///
+/// # Errors
+///
+/// Returns `IggyError::InvalidOptionValue("segment_size")` when the value is
+/// below [`MIN_TOPIC_SEGMENT_SIZE`], above `ceiling`, or not a 512-byte
+/// multiple.
+pub fn validate_topic_segment_size(size_bytes: u64, ceiling: u64) -> Result<(), IggyError> {
+ if size_bytes < MIN_TOPIC_SEGMENT_SIZE
+ || size_bytes > ceiling
+ || !size_bytes.is_multiple_of(512)
+ {
+ return Err(IggyError::InvalidOptionValue(
+ topic_option_keys::SEGMENT_SIZE.to_string(),
+ ));
+ }
+ Ok(())
+}
+
+/// Resource whose option catalog `DescribeOptions` serves.
+#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
+#[serde(rename_all = "snake_case")]
+pub enum OptionsScope {
+ Topic,
+ Stream,
+ User,
+}
+
+impl OptionsScope {
+ #[must_use]
+ pub fn as_code(&self) -> u8 {
+ match self {
+ Self::Topic => 1,
+ Self::Stream => 2,
+ Self::User => 3,
+ }
+ }
+
+ /// # Errors
+ ///
+ /// Returns `IggyError::InvalidCommand` for an unknown scope code.
+ pub fn from_code(code: u8) -> Result<Self, IggyError> {
+ match code {
+ 1 => Ok(Self::Topic),
+ 2 => Ok(Self::Stream),
+ 3 => Ok(Self::User),
+ _ => Err(IggyError::InvalidCommand),
+ }
+ }
+}
+
+impl std::fmt::Display for OptionsScope {
+ fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+ match self {
+ Self::Topic => write!(f, "topic"),
+ Self::Stream => write!(f, "stream"),
+ Self::User => write!(f, "user"),
+ }
+ }
+}
+
+impl FromStr for OptionsScope {
+ type Err = IggyError;
+
+ fn from_str(scope: &str) -> Result<Self, Self::Err> {
+ match scope {
+ "topic" => Ok(Self::Topic),
+ "stream" => Ok(Self::Stream),
+ "user" => Ok(Self::User),
+ _ => Err(IggyError::InvalidCommand),
+ }
+ }
+}
+
+/// One entry of a resource's option catalog, as served by `DescribeOptions`.
+#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
+pub struct OptionSpec {
+ /// Option key accepted by the create command.
+ pub key: String,
+ /// Canonical kind the server persists the value under. A `String` value
+ /// parsed via the server-config rules is always accepted too.
+ pub kind: HeaderKind,
+ /// This node's current resolved default in the canonical kind's
+ /// encoding; empty when the key has no server default.
+ #[serde(default)]
+ pub default_value: Vec<u8>,
+ /// Human-readable description.
+ #[serde(default)]
+ pub description: String,
+}
+
+/// Every key `CreateTopic` currently accepts. Unknown keys are rejected at
+/// the edge, never skipped: a silently ignored knob would hand the client
+/// server defaults without it ever learning.
+pub const TOPIC_OPTION_KEYS: &[&str] = &[
+ topic_option_keys::COMPRESSION_ALGORITHM,
+ topic_option_keys::MESSAGE_EXPIRY,
+ topic_option_keys::MAX_TOPIC_SIZE,
+ topic_option_keys::SEGMENT_SIZE,
+ topic_option_keys::ENFORCE_FSYNC,
+ topic_option_keys::MESSAGES_REQUIRED_TO_SAVE,
+ topic_option_keys::SIZE_OF_MESSAGES_REQUIRED_TO_SAVE,
+ topic_option_keys::PREALLOCATE_SEGMENTS,
+];
+
+/// The subset of [`TOPIC_OPTION_KEYS`] an `UpdateTopic` options block may
+/// carry.
+///
+/// These three used to be fixed fields of the update command, which meant one
+/// setting had two homes and an update always rewrote all three whether the
+/// caller meant to or not. As options they are patched: a key the client did
+/// not send keeps its current value.
+///
+/// The partition runtime knobs (`segment_size`, `enforce_fsync`, both flush
+/// thresholds, `preallocate_segments`) stay out, and not only because nothing
+/// re-pushes them to a live partition. They describe how a partition's storage
+/// was laid down: changing `segment_size` mid-segment leaves one segment sized
+/// by the old cap and the next by the new one, and `preallocate_segments` can
+/// only act on a file not yet opened. A topic gets them at creation and keeps
+/// them, so its segments stay uniform.
+pub const UPDATABLE_TOPIC_OPTION_KEYS: &[&str] = &[
+ topic_option_keys::COMPRESSION_ALGORITHM,
+ topic_option_keys::MESSAGE_EXPIRY,
+ topic_option_keys::MAX_TOPIC_SIZE,
+];
+
+/// Keys an `UpdateStream` options block may carry. Empty because streams have
+/// no catalog keys yet: the block exists so the first one costs a catalog
+/// entry instead of another wire change, and until then every key is rejected
+/// by name rather than stored and ignored.
+pub const UPDATABLE_STREAM_OPTION_KEYS: &[&str] = &[];
+
+/// Keys an `UpdateUser` options block may carry. Empty for the same reason as
+/// [`UPDATABLE_STREAM_OPTION_KEYS`].
+pub const UPDATABLE_USER_OPTION_KEYS: &[&str] = &[];
+
+/// Longest key echoed back in an option error. A rejected key is attacker-sized
+/// by definition, and the error rides the wire back to the client.
+const ERROR_KEY_PREVIEW_LEN: usize = 64;
+
+fn key_preview(key: &str) -> String {
+ key.chars().take(ERROR_KEY_PREVIEW_LEN).collect()
+}
+
+/// Build an options map from string key-values, all marked explicit.
+///
+/// Shared by the update-options types: their keys are all client-sent, so none
+/// of the derived-provenance handling that create needs applies. Callers that
+/// also have typed fields insert those afterwards, so a typed value wins on
+/// collision and keeps its canonical kind.
+///
+/// # Errors
+///
+/// `UnsupportedOptionKey` when a key is empty or over 255 bytes,
+/// `InvalidOptionValue` when a value is. Both bounds come from the
+/// header-field codec these entries ride.
+fn raw_options_map(raw: &BTreeMap<String, String>) -> Result<ResourceOptions, IggyError> {
+ let mut options = ResourceOptions::new();
+ for (key, value) in raw {
+ let header_key = HeaderKey::from_str(key)
+ .map_err(|_| IggyError::UnsupportedOptionKey(key_preview(key)))?;
+ let header_value = HeaderValue::try_from(value.as_str())
+ .map_err(|_| IggyError::InvalidOptionValue(key_preview(key)))?;
+ options.insert(header_key, OptionValue::explicit(header_value));
+ }
+ Ok(options)
+}
+
+/// Encode string key-values into an options block, all marked explicit.
+///
+/// # Errors
+///
+/// See [`raw_options_map`].
+fn raw_options_to_wire(raw: &BTreeMap<String, String>) -> Result<WireOptions, IggyError> {
+ crate::wire_conversions::resource_options_to_wire(
+ &raw_options_map(raw)?,
+ OptionsProvenance::All,
+ )
+}
+
+/// The options an `UpdateStream` may carry.
+///
+/// Absent keys leave the stream's existing values alone -- an update patches
+/// the option map, it does not replace it. See [`TopicUpdateOptions`] for why.
+#[derive(Debug, Clone, Default, PartialEq)]
+pub struct StreamUpdateOptions {
+ /// Keys sent as `String` values, checked against
+ /// [`UPDATABLE_STREAM_OPTION_KEYS`] server-side. That list is empty, so
+ /// every key is currently refused by name; the field exists so the first
+ /// updatable stream key costs a catalog entry, not a wire change.
+ pub raw: BTreeMap<String, String>,
+}
+
+impl StreamUpdateOptions {
+ /// Encode the present keys into an options block.
+ ///
+ /// # Errors
+ ///
+ /// See [`raw_options_to_wire`].
+ pub fn to_wire(&self) -> Result<WireOptions, IggyError> {
+ raw_options_to_wire(&self.raw)
+ }
+}
+
+/// The options an `UpdateUser` may carry. Same patch semantics as
+/// [`StreamUpdateOptions`].
+#[derive(Debug, Clone, Default, PartialEq)]
+pub struct UserUpdateOptions {
+ /// Keys sent as `String` values, checked against
+ /// [`UPDATABLE_USER_OPTION_KEYS`] server-side. That list is empty, so every
+ /// key is currently refused by name; the field exists so the first
+ /// updatable user key costs a catalog entry, not a wire change.
+ pub raw: BTreeMap<String, String>,
+}
+
+impl UserUpdateOptions {
+ /// Encode the present keys into an options block.
+ ///
+ /// # Errors
+ ///
+ /// See [`raw_options_to_wire`].
+ pub fn to_wire(&self) -> Result<WireOptions, IggyError> {
+ raw_options_to_wire(&self.raw)
+ }
+}
+
+/// This node's configured fallbacks for the runtime knobs, used to fill the
+/// derived-options block for keys a client did not send.
+#[derive(Debug, Clone, Copy, PartialEq)]
+pub struct TopicRuntimeDefaults {
+ pub segment_size: IggyByteSize,
+ pub enforce_fsync: bool,
+ pub messages_required_to_save: u32,
+ pub size_of_messages_required_to_save: IggyByteSize,
+ pub preallocate_segments: bool,
+}
+
+/// A topic's resolved runtime knobs, as carried from the metadata plane to
+/// each of its partitions. `None` means "keep the shard-wide configured
+/// value": topics created without an options block (simulator, unit tests)
+/// have no resolved values to carry.
+#[derive(Debug, Clone, Copy, Default, PartialEq)]
+pub struct TopicRuntimeOptions {
+ pub segment_size: Option<IggyByteSize>,
+ pub enforce_fsync: Option<bool>,
+ pub messages_required_to_save: Option<u32>,
+ pub size_of_messages_required_to_save: Option<IggyByteSize>,
+ pub preallocate_segments: Option<bool>,
+}
+
+impl TopicRuntimeOptions {
+ /// Derive the runtime knobs from a topic's persisted options map.
+ ///
+ /// Degrades per key, not per map. An entry this build cannot interpret
+ /// leaves its own knob unset and every other knob intact, so one key a
+ /// newer node wrote cannot silently drop a topic's `enforce_fsync` or
+ /// reset its segment size along with it.
+ #[must_use]
+ pub fn from_resource_options(options: &ResourceOptions) -> Self {
+ let parsed = TopicCreateOptions::from_resource_options(options);
+ Self {
+ segment_size: parsed.segment_size,
+ enforce_fsync: parsed.enforce_fsync,
+ messages_required_to_save: parsed.messages_required_to_save,
+ size_of_messages_required_to_save: parsed.size_of_messages_required_to_save,
+ preallocate_segments: parsed.preallocate_segments,
+ }
+ }
+}
+
+/// The options an `UpdateTopic` may carry, as a type that cannot express a
+/// key the update path rejects (see [`UPDATABLE_TOPIC_OPTION_KEYS`]).
+///
+/// Separate from [`TopicCreateOptions`] on purpose: reusing that struct would
+/// let a caller set `segment_size` on an update, encode it, and only learn at
+/// the server that it was refused. Absent keys leave the topic's existing
+/// value alone -- an update patches the option map, it does not replace it.
+#[derive(Debug, Clone, Default, PartialEq)]
+pub struct TopicUpdateOptions {
+ /// `None` leaves the topic's current algorithm alone.
+ pub compression_algorithm: Option<CompressionAlgorithm>,
+ /// `None` leaves the topic's current expiry alone.
+ pub message_expiry: Option<IggyExpiry>,
+ /// `None` leaves the topic's current cap alone.
+ pub max_topic_size: Option<MaxTopicSize>,
+ /// Keys sent as `String` values, checked against
+ /// [`UPDATABLE_TOPIC_OPTION_KEYS`] server-side. Lets a client reach an
+ /// updatable key added to the catalog after this build shipped.
+ pub raw: BTreeMap<String, String>,
+}
+
+impl TopicUpdateOptions {
+ /// Encode the present keys into an options block, in canonical kinds.
+ ///
+ /// A typed field is inserted after the raw entries, so it wins on collision
+ /// and keeps its canonical kind.
+ ///
+ /// # Errors
+ ///
+ /// See [`raw_options_map`].
+ pub fn to_wire(&self) -> Result<WireOptions, IggyError> {
+ let mut options = raw_options_map(&self.raw)?;
+ if let Some(compression_algorithm) = self.compression_algorithm {
+ options.insert(
+ HeaderKey::from_str(topic_option_keys::COMPRESSION_ALGORITHM)
+ .expect("catalog key is a valid header key"),
+ OptionValue::explicit(
+ HeaderValue::try_from(compression_algorithm.to_string().as_str())
+ .expect("compression name fits a header value"),
+ ),
+ );
+ }
+ if let Some(message_expiry) = self.message_expiry {
+ options.insert(
+ HeaderKey::from_str(topic_option_keys::MESSAGE_EXPIRY)
+ .expect("catalog key is a valid header key"),
+ OptionValue::explicit(HeaderValue::from(u64::from(message_expiry))),
+ );
+ }
+ if let Some(max_topic_size) = self.max_topic_size {
+ options.insert(
+ HeaderKey::from_str(topic_option_keys::MAX_TOPIC_SIZE)
+ .expect("catalog key is a valid header key"),
+ OptionValue::explicit(HeaderValue::from(u64::from(max_topic_size))),
+ );
+ }
+ crate::wire_conversions::resource_options_to_wire(&options, OptionsProvenance::All)
+ }
+}
+
+/// Typed view of the known topic option keys, parsed from a wire block.
+///
+/// `None` means the key was absent, which always means "resolve from server
+/// defaults at admission". Values that parse to their type's `ServerDefault`
+/// sentinel are normalized to `None` for the same reason.
+#[derive(Debug, Clone, Default, PartialEq)]
+pub struct TopicCreateOptions {
+ /// Partitions to allocate. NOT an option key: it fills the `CreateTopic`
+ /// command's own fixed field, because it is an argument to the operation
+ /// rather than a property of the topic (admission consumes it to compute
+ /// assignments, and a stored count would go stale on the first
+ /// `CreatePartitions`). Carried here so callers pass one bundle.
+ /// `None` means [`DEFAULT_PARTITIONS_COUNT`].
+ pub partitions_count: Option<u32>,
+ pub compression_algorithm: Option<CompressionAlgorithm>,
+ pub message_expiry: Option<IggyExpiry>,
+ pub max_topic_size: Option<MaxTopicSize>,
+ /// Per-topic segment size; `None` resolves against `[system.segment]
+ /// size` at admission. `0` is normalized to `None`.
+ pub segment_size: Option<IggyByteSize>,
+ /// Per-topic fsync enforcement; `None` resolves against
+ /// `[system.partition] enforce_fsync`.
+ pub enforce_fsync: Option<bool>,
+ /// Per-topic message-count flush threshold; `None` resolves against
+ /// `[system.partition] messages_required_to_save`. `0` is rejected.
+ pub messages_required_to_save: Option<u32>,
+ /// Per-topic byte flush threshold; `None` resolves against
+ /// [`DEFAULT_SIZE_OF_MESSAGES_REQUIRED_TO_SAVE`].
+ pub size_of_messages_required_to_save: Option<IggyByteSize>,
+ /// Whether this topic's segments reserve their bytes on open; `None`
+ /// resolves against [`DEFAULT_PREALLOCATE_SEGMENTS`].
+ pub preallocate_segments: Option<bool>,
+ /// String-valued keys with no typed field above, parsed server-side
+ /// through each key's `FromStr`. Lets a client reach a key added to the
+ /// server catalog after this build shipped. Outbound only: a typed field
+ /// wins on collision in [`Self::to_wire`], and [`Self::parse`] never
+ /// populates it.
+ pub raw: BTreeMap<String, String>,
+}
+
+impl TopicCreateOptions {
+ /// Parse a wire options block against the topic catalog.
+ ///
+ /// # Errors
+ ///
+ /// `UnsupportedOptionKey` for a key outside [`TOPIC_OPTION_KEYS`];
+ /// `InvalidOptionValue` when a value has the wrong kind or fails its
+ /// type's `FromStr`.
+ pub fn parse(options: &WireOptions) -> Result<Self, IggyError> {
+ let mut parsed = Self::default();
+ for entry in options {
+ // Wire validation already enforced UTF-8 string keys.
+ let key = String::from_utf8_lossy(entry.key);
+ parsed.absorb(&entry, &key, UnknownEntry::Reject)?;
+ }
+ Ok(parsed)
+ }
+
+ /// Parse a block that is already COMMITTED, skipping what this build
+ /// cannot interpret.
+ ///
+ /// Admission gated the block against the catalog on the primary, once.
+ /// Re-running that gate at apply would make the verdict depend on the
+ /// build running it, so a replica that predates a key would reject an
+ /// operation its peers accepted and diverge from the group permanently.
+ /// The catalog belongs at the edge; apply only reads what it knows.
+ #[must_use]
+ pub fn parse_committed(options: &WireOptions) -> Self {
+ let mut parsed = Self::default();
+ for entry in options {
+ let key = String::from_utf8_lossy(entry.key);
+ // Infallible: `Skip` turns every parse error into a no-op.
+ let _ = parsed.absorb(&entry, &key, UnknownEntry::Skip);
+ }
+ parsed
+ }
+
+ /// Fold one catalog entry into `self`, applying `unknown` to an entry this
+ /// build cannot interpret. Shared by [`Self::parse`] (wire block) and
+ /// [`Self::from_resource_options`] (persisted map) so both read the
+ /// identical key set, kinds, and value bounds, and differ only in what
+ /// they do with an entry outside it.
+ fn absorb(
+ &mut self,
+ entry: &WireUserHeaderEntry<'_>,
+ key: &str,
+ unknown: UnknownEntry,
+ ) -> Result<(), IggyError> {
+ // Every arm assigns only after its parse succeeds, so a skipped entry
+ // leaves `self` exactly as it found it.
+ match self.absorb_strict(entry, key) {
+ Ok(()) => Ok(()),
+ Err(error) => match unknown {
+ UnknownEntry::Reject => Err(error),
+ UnknownEntry::Skip => Ok(()),
+ },
+ }
+ }
+
+ fn absorb_strict(
+ &mut self,
+ entry: &WireUserHeaderEntry<'_>,
+ key: &str,
+ ) -> Result<(), IggyError> {
+ let parsed = self;
+ {
+ match key {
+ topic_option_keys::COMPRESSION_ALGORITHM => {
+ let value = parse_compression(entry, key)?;
+ parsed.compression_algorithm = Some(value);
+ }
+ topic_option_keys::MESSAGE_EXPIRY => {
+ let expiry = IggyExpiry::from(parse_u64_or(entry, key, IggyExpiry::from_str)?);
+ parsed.message_expiry = (expiry != IggyExpiry::ServerDefault).then_some(expiry);
+ }
+ topic_option_keys::MAX_TOPIC_SIZE => {
+ let size =
+ MaxTopicSize::from(parse_u64_or(entry, key, MaxTopicSize::from_str)?);
+ parsed.max_topic_size = (size != MaxTopicSize::ServerDefault).then_some(size);
+ }
+ topic_option_keys::SEGMENT_SIZE => {
+ let size = parse_byte_size(entry, key)?;
+ parsed.segment_size = (size != 0).then_some(IggyByteSize::from(size));
+ }
+ topic_option_keys::ENFORCE_FSYNC => {
+ parsed.enforce_fsync = Some(parse_bool(entry, key)?);
+ }
+ topic_option_keys::MESSAGES_REQUIRED_TO_SAVE => {
+ let messages = parse_u32(entry, key)?;
+ if messages == 0 {
+ return Err(IggyError::InvalidOptionValue(key.to_string()));
+ }
+ parsed.messages_required_to_save = Some(messages);
+ }
+ topic_option_keys::SIZE_OF_MESSAGES_REQUIRED_TO_SAVE => {
+ let size = parse_byte_size(entry, key)?;
+ parsed.size_of_messages_required_to_save =
+ (size != 0).then_some(IggyByteSize::from(size));
+ }
+ topic_option_keys::PREALLOCATE_SEGMENTS => {
+ parsed.preallocate_segments = Some(parse_bool(entry, key)?);
+ }
+ _ => return Err(IggyError::UnsupportedOptionKey(key.to_string())),
+ }
+ }
+ Ok(())
+ }
+
+ /// Encode the present keys into a client options block, in canonical
+ /// kinds. The client-side counterpart of [`Self::parse`].
+ ///
+ /// A typed field is inserted after the raw entries, so it wins on
+ /// collision and keeps its canonical kind.
+ ///
+ /// # Errors
+ ///
+ /// See [`raw_options_map`].
+ pub fn to_wire(&self) -> Result<WireOptions, IggyError> {
+ let mut options = raw_options_map(&self.raw)?;
+ if let Some(compression_algorithm) = self.compression_algorithm {
+ options.insert(
+ HeaderKey::from_str(topic_option_keys::COMPRESSION_ALGORITHM)
+ .expect("catalog key is a valid header key"),
+ OptionValue::explicit(
+ HeaderValue::try_from(compression_algorithm.to_string().as_str())
+ .expect("compression name fits a header value"),
+ ),
+ );
+ }
+ if let Some(message_expiry) = self.message_expiry {
+ options.insert(
+ HeaderKey::from_str(topic_option_keys::MESSAGE_EXPIRY)
+ .expect("catalog key is a valid header key"),
+ OptionValue::explicit(HeaderValue::from(u64::from(message_expiry))),
+ );
+ }
+ if let Some(max_topic_size) = self.max_topic_size {
+ options.insert(
+ HeaderKey::from_str(topic_option_keys::MAX_TOPIC_SIZE)
+ .expect("catalog key is a valid header key"),
+ OptionValue::explicit(HeaderValue::from(u64::from(max_topic_size))),
+ );
+ }
+ if let Some(segment_size) = self.segment_size {
+ options.insert(
+ HeaderKey::from_str(topic_option_keys::SEGMENT_SIZE)
+ .expect("catalog key is a valid header key"),
+ OptionValue::explicit(HeaderValue::from(segment_size.as_bytes_u64())),
+ );
+ }
+ if let Some(enforce_fsync) = self.enforce_fsync {
+ options.insert(
+ HeaderKey::from_str(topic_option_keys::ENFORCE_FSYNC)
+ .expect("catalog key is a valid header key"),
+ OptionValue::explicit(HeaderValue::from(enforce_fsync)),
+ );
+ }
+ if let Some(messages_required_to_save) = self.messages_required_to_save {
+ options.insert(
+ HeaderKey::from_str(topic_option_keys::MESSAGES_REQUIRED_TO_SAVE)
+ .expect("catalog key is a valid header key"),
+ OptionValue::explicit(HeaderValue::from(messages_required_to_save)),
+ );
+ }
+ if let Some(size_of_messages) = self.size_of_messages_required_to_save {
+ options.insert(
+ HeaderKey::from_str(topic_option_keys::SIZE_OF_MESSAGES_REQUIRED_TO_SAVE)
+ .expect("catalog key is a valid header key"),
+ OptionValue::explicit(HeaderValue::from(size_of_messages.as_bytes_u64())),
+ );
+ }
+ if let Some(preallocate_segments) = self.preallocate_segments {
+ options.insert(
+ HeaderKey::from_str(topic_option_keys::PREALLOCATE_SEGMENTS)
+ .expect("catalog key is a valid header key"),
+ OptionValue::explicit(HeaderValue::from(preallocate_segments)),
+ );
+ }
+ crate::wire_conversions::resource_options_to_wire(&options, OptionsProvenance::All)
+ }
+
+ /// Render the present keys as string key-values, for a transport that
+ /// carries options as a JSON object instead of a TLV block.
+ ///
+ /// `compression_algorithm`, `message_expiry` and `max_topic_size` are left
+ /// out: the JSON body has dedicated fields for those three, and sending
+ /// both would be the two-sources-for-one-setting the update path refuses.
+ /// The rest ride as strings and are parsed server-side by the same
+ /// `FromStr` rules a config file value goes through, so a typed field
+ /// survives the round trip rather than being silently dropped.
+ #[must_use]
+ pub fn to_string_options(&self) -> BTreeMap<String, String> {
+ // Typed fields are inserted over the raw entries, matching the
+ // collision rule `to_wire` applies.
+ let mut options = self.raw.clone();
+ if let Some(segment_size) = self.segment_size {
+ options.insert(
+ topic_option_keys::SEGMENT_SIZE.to_owned(),
+ segment_size.as_bytes_u64().to_string(),
+ );
+ }
+ if let Some(enforce_fsync) = self.enforce_fsync {
+ options.insert(
+ topic_option_keys::ENFORCE_FSYNC.to_owned(),
+ enforce_fsync.to_string(),
+ );
+ }
+ if let Some(messages_required_to_save) = self.messages_required_to_save {
+ options.insert(
+ topic_option_keys::MESSAGES_REQUIRED_TO_SAVE.to_owned(),
+ messages_required_to_save.to_string(),
+ );
+ }
+ if let Some(size_of_messages) = self.size_of_messages_required_to_save {
+ options.insert(
+ topic_option_keys::SIZE_OF_MESSAGES_REQUIRED_TO_SAVE.to_owned(),
+ size_of_messages.as_bytes_u64().to_string(),
+ );
+ }
+ if let Some(preallocate_segments) = self.preallocate_segments {
+ options.insert(
+ topic_option_keys::PREALLOCATE_SEGMENTS.to_owned(),
+ preallocate_segments.to_string(),
+ );
+ }
+ options
+ }
+
+ /// Encode the resolved values for every key the client did NOT send into
+ /// a derived-options wire block, in canonical kinds. `partitions_count`
+ /// is never included: it is consumed at admission.
+ ///
+ /// # Errors
+ ///
+ /// See [`crate::wire_conversions::resource_options_to_wire`].
+ pub fn derived_block(
+ &self,
+ compression_algorithm: CompressionAlgorithm,
+ message_expiry: IggyExpiry,
+ max_topic_size: MaxTopicSize,
+ runtime_defaults: TopicRuntimeDefaults,
+ ) -> Result<WireOptions, IggyError> {
+ let mut derived = ResourceOptions::new();
+ if self.compression_algorithm.is_none() {
+ derived.insert(
+ HeaderKey::from_str(topic_option_keys::COMPRESSION_ALGORITHM)
+ .expect("catalog key is a valid header key"),
+ OptionValue::derived(
+ HeaderValue::try_from(compression_algorithm.to_string().as_str())
+ .expect("compression name fits a header value"),
+ ),
+ );
+ }
+ if self.message_expiry.is_none() {
+ derived.insert(
+ HeaderKey::from_str(topic_option_keys::MESSAGE_EXPIRY)
+ .expect("catalog key is a valid header key"),
+ OptionValue::derived(HeaderValue::from(u64::from(message_expiry))),
+ );
+ }
+ if self.max_topic_size.is_none() {
+ derived.insert(
+ HeaderKey::from_str(topic_option_keys::MAX_TOPIC_SIZE)
+ .expect("catalog key is a valid header key"),
+ OptionValue::derived(HeaderValue::from(u64::from(max_topic_size))),
+ );
+ }
+ if self.segment_size.is_none() {
+ derived.insert(
+ HeaderKey::from_str(topic_option_keys::SEGMENT_SIZE)
+ .expect("catalog key is a valid header key"),
+ OptionValue::derived(HeaderValue::from(
+ runtime_defaults.segment_size.as_bytes_u64(),
+ )),
+ );
+ }
+ if self.enforce_fsync.is_none() {
+ derived.insert(
+ HeaderKey::from_str(topic_option_keys::ENFORCE_FSYNC)
+ .expect("catalog key is a valid header key"),
+ OptionValue::derived(HeaderValue::from(runtime_defaults.enforce_fsync)),
+ );
+ }
+ if self.messages_required_to_save.is_none() {
+ derived.insert(
+ HeaderKey::from_str(topic_option_keys::MESSAGES_REQUIRED_TO_SAVE)
+ .expect("catalog key is a valid header key"),
+ OptionValue::derived(HeaderValue::from(
+ runtime_defaults.messages_required_to_save,
+ )),
+ );
+ }
+ if self.size_of_messages_required_to_save.is_none() {
+ derived.insert(
+ HeaderKey::from_str(topic_option_keys::SIZE_OF_MESSAGES_REQUIRED_TO_SAVE)
+ .expect("catalog key is a valid header key"),
+ OptionValue::derived(HeaderValue::from(
+ runtime_defaults
+ .size_of_messages_required_to_save
+ .as_bytes_u64(),
+ )),
+ );
+ }
+ if self.preallocate_segments.is_none() {
+ derived.insert(
+ HeaderKey::from_str(topic_option_keys::PREALLOCATE_SEGMENTS)
+ .expect("catalog key is a valid header key"),
+ OptionValue::derived(HeaderValue::from(runtime_defaults.preallocate_segments)),
+ );
+ }
+ crate::wire_conversions::resource_options_to_wire(&derived, OptionsProvenance::All)
+ }
+
+ /// Parse a topic's PERSISTED options map (explicit plus admission-derived
+ /// entries) back into typed values.
+ ///
+ /// The persisted map is the single source of truth for a topic's knobs:
+ /// nothing on the STM `Topic` duplicates it per key, so a new key costs
+ /// one catalog entry rather than one field on every stored type.
+ ///
+ /// Infallible by design. A persisted map can hold keys a newer build
+ /// wrote, and refusing the whole map for one of them is what would make
+ /// this replica read a topic differently from the node that stored it.
+ /// Unreadable entries are skipped; their knobs stay unset and resolve
+ /// against shard-wide config exactly as an absent key does.
+ #[must_use]
+ pub fn from_resource_options(options: &ResourceOptions) -> Self {
+ let mut parsed = Self::default();
+ for (key, option) in options {
+ let key = String::from_utf8_lossy(key.as_bytes());
+ let entry = WireUserHeaderEntry {
+ key_kind: iggy_binary_protocol::WireHeaderKind(HeaderKind::String.as_code()),
+ key: key.as_bytes(),
+ value_kind: iggy_binary_protocol::WireHeaderKind(option.value.kind().as_code()),
+ value: option.value.as_bytes(),
+ };
+ // Infallible: `Skip` turns every parse error into a no-op.
+ let _ = parsed.absorb(&entry, &key, UnknownEntry::Skip);
+ }
+ parsed
+ }
+}
+
+/// What a parse does with an entry this build cannot interpret: a key outside
+/// [`TOPIC_OPTION_KEYS`], or a catalog key whose value kind or payload does
+/// not parse.
+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+enum UnknownEntry {
Review Comment:
`UnknownEntry` plus the `absorb` wrapper exist only to turn `absorb_strict`'s err into ok - the call sites already spell the difference (`?` vs `let _ =`), so calling `absorb_strict` directly deletes the enum and ~30 lines with identical behavior. the "assigns only after its parse succeeds" comment should move onto `absorb_strict` - it's the invariant that makes the skip path safe.
##########
core/common/src/http/topics/create_topic.rs:
##########
@@ -47,10 +47,12 @@ pub struct CreateTopic {
/// Max topic size, if `Unlimited` then topic size is unlimited.
/// Can't be lower than segment size in the config.
pub max_topic_size: MaxTopicSize,
- /// Replication factor for the topic.
- pub replication_factor: Option<u8>,
/// Unique topic name, max length is 255 characters.
pub name: String,
+ /// Additional topic options as string key-values, parsed by the server
+ /// with the same rules as its config file. Unknown keys are rejected.
+ #[serde(default)]
+ pub options: BTreeMap<String, String>,
Review Comment:
options go in as a plain string map but come back as `{"value":{"kind":"uint64","value":"<base64>"},"explicit":true}` - an http caller has to base64+LE-decode to read a knob, and a GET body can't be fed back into a POST, so the re-send-only-options round trip documented on the response side is unimplementable over http. render values in their config-string form on the way out.
##########
core/common/src/types/options/mod.rs:
##########
@@ -0,0 +1,1295 @@
+// 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.
+
+//! Key-value options attached to streams, topics, and users at creation.
+//!
+//! Options reuse the typed [`HeaderKey`] / [`HeaderValue`] machinery from
+//! message user headers. The map MUST stay a `BTreeMap`: metadata snapshots
+//! require deterministic ordering across replicas, and a hash map would make
+//! snapshot bytes diverge per replica.
+
+use std::collections::BTreeMap;
+use std::str::FromStr;
+
+use iggy_binary_protocol::{WireOptions, WireUserHeaderEntry};
+use serde::{Deserialize, Serialize};
+
+use crate::types::compression::compression_algorithm::CompressionAlgorithm;
+use crate::types::message::{HeaderKey, HeaderKind, HeaderValue};
+use crate::{IggyByteSize, IggyError, IggyExpiry, MaxTopicSize};
+
+/// A single resolved option entry.
+#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
+pub struct OptionValue {
+ /// The effective value, resolved by the admitting primary at creation.
+ pub value: HeaderValue,
+ /// Whether the client explicitly sent this key. Derived entries
+ /// (`explicit == false`) were filled from server defaults at admission
+ /// and would have resolved differently under other server configs.
+ pub explicit: bool,
+}
+
+/// Which provenance classes an encoded options block carries.
+///
+/// Responses split a resource's options into two blocks so `GetTopic` can say
+/// which values the client chose and which admission filled in.
+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+pub enum OptionsProvenance {
+ /// Every entry, whatever its provenance.
+ All,
+ /// Only the keys the client sent.
+ Explicit,
+ /// Only the keys admission resolved from server defaults.
+ Derived,
+}
+
+impl OptionValue {
+ /// Whether this entry belongs in a block encoded for `provenance`.
+ #[must_use]
+ pub const fn matches(&self, provenance: OptionsProvenance) -> bool {
+ match provenance {
+ OptionsProvenance::All => true,
+ OptionsProvenance::Explicit => self.explicit,
+ OptionsProvenance::Derived => !self.explicit,
+ }
+ }
+
+ #[must_use]
+ pub fn explicit(value: HeaderValue) -> Self {
+ Self {
+ value,
+ explicit: true,
+ }
+ }
+
+ #[must_use]
+ pub fn derived(value: HeaderValue) -> Self {
+ Self {
+ value,
+ explicit: false,
+ }
+ }
+}
+
+/// Options attached to a stream, topic, or user, keyed by option name.
+pub type ResourceOptions = BTreeMap<HeaderKey, OptionValue>;
+
+/// JSON form for [`ResourceOptions`].
+///
+/// [`HeaderKey`] serializes as a struct (`kind` + base64 `value`), and JSON
+/// object keys must be strings, so the derived impl fails outright with
+/// "key must be a string" the moment a map is non-empty. Every HTTP response
+/// carrying options goes through here instead, rendering the key as its plain
+/// text. Binary transports are unaffected: they never touch serde.
+pub mod resource_options_json {
+ use super::{HeaderKey, OptionValue, ResourceOptions};
+ use serde::de::Error as _;
+ use serde::{Deserialize, Deserializer, Serialize, Serializer};
+ use std::collections::BTreeMap;
+ use std::str::FromStr;
+
+ /// # Errors
+ ///
+ /// Propagates the serializer's own errors.
+ pub fn serialize<S: Serializer>(
+ options: &ResourceOptions,
+ serializer: S,
+ ) -> Result<S::Ok, S::Error> {
+ let readable: BTreeMap<String, &OptionValue> = options
+ .iter()
+ .map(|(key, value)| (String::from_utf8_lossy(key.as_bytes()).into_owned(), value))
+ .collect();
+ readable.serialize(serializer)
+ }
+
+ /// # Errors
+ ///
+ /// Returns a deserializer error when a key is not a valid option name.
+ pub fn deserialize<'de, D: Deserializer<'de>>(
+ deserializer: D,
+ ) -> Result<ResourceOptions, D::Error> {
+ let readable = BTreeMap::<String, OptionValue>::deserialize(deserializer)?;
+ readable
+ .into_iter()
+ .map(|(key, value)| {
+ HeaderKey::from_str(&key)
+ .map(|key| (key, value))
+ .map_err(D::Error::custom)
+ })
+ .collect()
+ }
+}
+
+/// Topic option catalog: the keys `CreateTopic` accepts.
+pub mod topic_option_keys {
+ /// Compression algorithm name (`none`, `gzip`).
+ pub const COMPRESSION_ALGORITHM: &str = "compression_algorithm";
+ /// Message expiry: `Uint64` micros or a humantime string (`7 days`).
+ pub const MESSAGE_EXPIRY: &str = "message_expiry";
+ /// Topic size cap: `Uint64` bytes or a byte-size string (`1 GiB`).
+ pub const MAX_TOPIC_SIZE: &str = "max_topic_size";
+ /// Segment size: `Uint64` bytes or a byte-size string (`128 MiB`).
+ /// Bounded: a 512-byte multiple within
+ /// [`super::MIN_TOPIC_SEGMENT_SIZE`]..=the server's segment ceiling.
+ pub const SEGMENT_SIZE: &str = "segment_size";
+ /// Whether writes to this topic's partitions fsync: `Bool`, or the
+ /// strings `true` / `false`.
+ pub const ENFORCE_FSYNC: &str = "enforce_fsync";
+ /// Flush the journal once it holds this many messages: `Uint32`.
+ /// Must be non-zero.
+ pub const MESSAGES_REQUIRED_TO_SAVE: &str = "messages_required_to_save";
+ /// Flush the journal once it holds this many bytes: `Uint64` or a
+ /// byte-size string. Paired with [`Self::MESSAGES_REQUIRED_TO_SAVE`];
+ /// whichever threshold trips first flushes.
+ pub const SIZE_OF_MESSAGES_REQUIRED_TO_SAVE: &str = "size_of_messages_required_to_save";
+ /// Reserve the segment's bytes up front on a filesystem that supports it:
+ /// `Bool`, or the strings `true` / `false`. Pairs with
+ /// [`Self::SEGMENT_SIZE`] -- preallocation reserves exactly that much, so
+ /// the two only make sense decided together.
+ pub const PREALLOCATE_SEGMENTS: &str = "preallocate_segments";
+}
+
+/// Values an absent topic option resolves to at admission.
+///
+/// These are the knobs' single source of truth: they used to live in
+/// `config.toml` (`[system.topic]`, `[system.partition]`, `[system.segment]`),
+/// which meant every one of them had two homes and an operator could not tell
+/// which won. A topic carries whatever it was created with; anything the
+/// client did not send resolves to the constant here and is persisted as a
+/// derived entry, so the effective value is always visible on `GetTopic`.
+///
+/// Each value matches what the shipped `config.toml` carried, so removing the
+/// keys changed no behavior for a topic created without options.
+pub const DEFAULT_PARTITIONS_COUNT: u32 = 1;
+/// `MaxTopicSize::Unlimited` (was `[system.topic] max_size = "unlimited"`).
+pub const DEFAULT_MAX_TOPIC_SIZE: u64 = u64::MAX;
+/// `IggyExpiry::NeverExpire` (was `[system.topic] message_expiry = "none"`).
+pub const DEFAULT_MESSAGE_EXPIRY: u64 = u64::MAX;
+/// 1 GiB (was `[system.segment] size = "1 GiB"`).
+pub const DEFAULT_SEGMENT_SIZE: u64 = 1024 * 1024 * 1024;
+/// Was `[system.partition] enforce_fsync = false`.
+pub const DEFAULT_ENFORCE_FSYNC: bool = false;
+/// Was `[system.partition] messages_required_to_save = 1024`.
+pub const DEFAULT_MESSAGES_REQUIRED_TO_SAVE: u32 = 1024;
+/// Opt-in, unlike the `[system.segment] preallocate = true` this replaced.
+///
+/// That default was never actually in force: the reservation ran through
+/// `compio::spawn_blocking`, which panics the shard because shard executors
+/// disable the blocking pool, so any deployment that worked at all had
+/// preallocation off. With the call fixed to run inline it reserves real
+/// extents, and `FALLOC_FL_KEEP_SIZE` against the 1 GiB default segment size
+/// means 1 GiB of disk per partition the moment it is created -- a full test
+/// sweep reserved 393 GB before this was flipped. A topic that wants the
+/// latency benefit asks for it with `preallocate_segments`.
+pub const DEFAULT_PREALLOCATE_SEGMENTS: bool = false;
+/// 1 MiB (was `[system.partition] size_of_messages_required_to_save`).
+pub const DEFAULT_SIZE_OF_MESSAGES_REQUIRED_TO_SAVE: u64 = 1024 * 1024;
+
+/// Every runtime knob at its default, for a partition built with no resolved
+/// topic options (simulator, unit tests).
+impl Default for TopicRuntimeDefaults {
+ fn default() -> Self {
+ Self {
+ segment_size: IggyByteSize::from(DEFAULT_SEGMENT_SIZE),
+ enforce_fsync: DEFAULT_ENFORCE_FSYNC,
+ messages_required_to_save: DEFAULT_MESSAGES_REQUIRED_TO_SAVE,
+ size_of_messages_required_to_save: IggyByteSize::from(
+ DEFAULT_SIZE_OF_MESSAGES_REQUIRED_TO_SAVE,
+ ),
+ preallocate_segments: DEFAULT_PREALLOCATE_SEGMENTS,
+ }
+ }
+}
+
+/// Smallest per-topic segment size. Segments far below the shipped default
+/// explode the per-partition segment count, which state transfer bounds via
+/// its manifest entry cap; a partition that crosses it becomes unservable
+/// for transfer, so the floor is enforced at admission.
+pub const MIN_TOPIC_SEGMENT_SIZE: u64 = 1024 * 1024;
+
+/// Validate an explicit per-topic `segment_size` against its bounds.
+///
+/// `ceiling` is node-derived: the smaller of the global segment maximum and
+/// the state-transfer artifact budget minus one bus frame (a segment may
+/// close one whole batch past its cap; an artifact ceiling below that
+/// refuses a legal segment and livelocks the partition's rejoin).
+///
+/// # Errors
+///
+/// Returns `IggyError::InvalidOptionValue("segment_size")` when the value is
+/// below [`MIN_TOPIC_SEGMENT_SIZE`], above `ceiling`, or not a 512-byte
+/// multiple.
+pub fn validate_topic_segment_size(size_bytes: u64, ceiling: u64) -> Result<(), IggyError> {
+ if size_bytes < MIN_TOPIC_SEGMENT_SIZE
+ || size_bytes > ceiling
+ || !size_bytes.is_multiple_of(512)
+ {
+ return Err(IggyError::InvalidOptionValue(
+ topic_option_keys::SEGMENT_SIZE.to_string(),
+ ));
+ }
+ Ok(())
+}
+
+/// Resource whose option catalog `DescribeOptions` serves.
+#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
+#[serde(rename_all = "snake_case")]
+pub enum OptionsScope {
+ Topic,
+ Stream,
+ User,
+}
+
+impl OptionsScope {
+ #[must_use]
+ pub fn as_code(&self) -> u8 {
+ match self {
+ Self::Topic => 1,
+ Self::Stream => 2,
+ Self::User => 3,
+ }
+ }
+
+ /// # Errors
+ ///
+ /// Returns `IggyError::InvalidCommand` for an unknown scope code.
+ pub fn from_code(code: u8) -> Result<Self, IggyError> {
+ match code {
+ 1 => Ok(Self::Topic),
+ 2 => Ok(Self::Stream),
+ 3 => Ok(Self::User),
+ _ => Err(IggyError::InvalidCommand),
+ }
+ }
+}
+
+impl std::fmt::Display for OptionsScope {
+ fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+ match self {
+ Self::Topic => write!(f, "topic"),
+ Self::Stream => write!(f, "stream"),
+ Self::User => write!(f, "user"),
+ }
+ }
+}
+
+impl FromStr for OptionsScope {
+ type Err = IggyError;
+
+ fn from_str(scope: &str) -> Result<Self, Self::Err> {
+ match scope {
+ "topic" => Ok(Self::Topic),
+ "stream" => Ok(Self::Stream),
+ "user" => Ok(Self::User),
+ _ => Err(IggyError::InvalidCommand),
+ }
+ }
+}
+
+/// One entry of a resource's option catalog, as served by `DescribeOptions`.
+#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
+pub struct OptionSpec {
+ /// Option key accepted by the create command.
+ pub key: String,
+ /// Canonical kind the server persists the value under. A `String` value
Review Comment:
two contract problems. "canonical kind the server persists the value under" is false for explicit keys - values are stored and echoed with the client's kind byte verbatim, so every cli `--set` lands as a string while the catalog advertises uint64. and `default_value` is documented as "this node's current resolved default" but every value served is a compile-time constant seeded at bootstrap, identical on every node and build. either canonicalize kinds at admission or document that kind is client-chosen, and fix the per-node wording.
##########
foreign/node/src/wire/options.utils.ts:
##########
@@ -0,0 +1,150 @@
+// 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.
+//
+
+import { HeaderKind } from './message/header.type.js';
+import {
+ serializeHeaders,
+ deserializeHeaderValue,
+ HeaderKeyFactory,
+ type HeaderValue,
+ type ParsedHeaderValue,
+} from './message/header.utils.js';
+
+/** Maximum number of key-value entries in one options block. */
+export const MAX_OPTIONS = 1024;
+
+/**
+ * Maximum total byte length of an encoded options block.
+ *
+ * Mirrors the Rust `MAX_OPTIONS_BYTES`, which in turn mirrors the user-headers
+ * budget: options ride that codec and inherit its limit.
+ */
+export const MAX_OPTIONS_BYTES = 100 * 1000;
+
+/**
+ * Key and value length bound, inherited from the header-field codec rather
+ * than being an options-specific rule (`serializeHeaders` enforces the same
+ * range on the way out).
+ */
+const MAX_HEADER_FIELD_LENGTH = 255;
+
+/** A resource option entry: UTF-8 string key with a typed value. */
+export type OptionEntry = {
+ key: string,
+ value: HeaderValue
+};
+
+/** Deserialized options block keyed by option name. */
+export type ParsedOptions = Record<string, ParsedHeaderValue>;
+
+/** Result of deserializing a length-prefixed options block. */
+export type OptionsDeserialized = {
+ /** Number of bytes consumed, length prefix included */
+ bytesRead: number,
+ /** Deserialized options */
+ options: ParsedOptions
+};
+
+/**
+ * Serializes resource options into a TLV block.
+ * Reuses the user-headers TLV encoding: each field is
+ * `[kind:u8][len:u32_le][bytes]`, alternating key, value.
+ * Empty options serialize to zero bytes.
+ *
+ * @param options - Option entries to serialize
+ * @returns Serialized options block
+ * @throws Error if an options constraint is violated
+ */
+export const serializeOptions = (options: OptionEntry[]): Buffer => {
+ if (options.length > MAX_OPTIONS)
+ throw new Error(
+ `Options block has ${options.length} entries, exceeds maximum ${MAX_OPTIONS}`);
+
+ // No key-length check here: `serializeHeaders` already bounds every field
+ // to 1..=255, so an options-specific cap would only duplicate it.
+ const block = serializeHeaders(options.map(({ key, value }) =>
+ ({ key: HeaderKeyFactory.String(key), value })));
+
+ if (block.length > MAX_OPTIONS_BYTES)
+ throw new Error(
+ `Options block is ${block.length} bytes, exceeds maximum ${MAX_OPTIONS_BYTES}`);
+
+ return block;
+};
+
+/**
+ * Deserializes a bare options TLV block spanning `[pos, end)`.
+ *
+ * @param p - Buffer containing the options block
+ * @param pos - Starting position of the block
+ * @param end - End position of the block (exclusive)
+ * @returns Deserialized options keyed by option name
+ * @throws Error if a key is not a string or the block is malformed
+ */
+export const deserializeOptions = (
+ p: Buffer, pos = 0, end = p.length
+): ParsedOptions => {
+ const options: ParsedOptions = {};
+ while (pos < end) {
+ const keyKind = p.readUInt8(pos);
+ if (keyKind !== HeaderKind.String)
+ throw new Error(`Option key kind ${keyKind} is not a string`);
+ const keyLength = p.readUInt32LE(pos + 1);
+ if (keyLength < 1 || keyLength > MAX_HEADER_FIELD_LENGTH)
+ throw new Error(
+ `Invalid option key length: ${keyLength}, ` +
+ `must be between 1 and ${MAX_HEADER_FIELD_LENGTH}`);
+ if (pos + 5 + keyLength > end)
+ throw new Error('Option key overruns the block');
+ const key = p.subarray(pos + 5, pos + 5 + keyLength).toString();
+ pos += 5 + keyLength;
+
+ const valueKind = p.readUInt8(pos);
+ const valueLength = p.readUInt32LE(pos + 1);
+ if (pos + 5 + valueLength > end)
+ throw new Error(`Option value for key '${key}' overruns the block`);
+ const valueBytes = p.subarray(pos + 5, pos + 5 + valueLength);
+ pos += 5 + valueLength;
+
+ let value: ParsedHeaderValue;
+ try {
+ value = deserializeHeaderValue(valueKind, valueBytes);
+ } catch {
+ // Unknown value kinds stay raw bytes, mirroring the wire
+ // forward-compatibility contract for options.
+ value = valueBytes;
+ }
+ options[key] = value;
+ }
+ return options;
+};
+
+/**
+ * Deserializes a `u32_le`-length-prefixed options block at `pos`.
+ *
+ * @param p - Buffer containing `[options_len:u32_le][options TLV]`
+ * @param pos - Starting position of the length prefix
+ * @returns Bytes consumed (prefix included) and deserialized options
+ */
+export const deserializePrefixedOptions = (
+ p: Buffer, pos = 0
+): OptionsDeserialized => {
+ const length = p.readUInt32LE(pos);
+ const options = deserializeOptions(p, pos + 4, pos + 4 + length);
Review Comment:
no check that `pos + 4 + length` fits the buffer - `subarray` clamps silently, so a truncated known-kind value throws inside `deserializeHeaderValue`, gets swallowed by the forward-compat catch, and surfaces as a raw buffer instead of the typed value, with `bytesRead` over-reporting. go and rust both hard-check the prefix; c# does now too.
##########
core/common/src/types/options/mod.rs:
##########
@@ -0,0 +1,1295 @@
+// 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.
+
+//! Key-value options attached to streams, topics, and users at creation.
+//!
+//! Options reuse the typed [`HeaderKey`] / [`HeaderValue`] machinery from
+//! message user headers. The map MUST stay a `BTreeMap`: metadata snapshots
+//! require deterministic ordering across replicas, and a hash map would make
+//! snapshot bytes diverge per replica.
+
+use std::collections::BTreeMap;
+use std::str::FromStr;
+
+use iggy_binary_protocol::{WireOptions, WireUserHeaderEntry};
+use serde::{Deserialize, Serialize};
+
+use crate::types::compression::compression_algorithm::CompressionAlgorithm;
+use crate::types::message::{HeaderKey, HeaderKind, HeaderValue};
+use crate::{IggyByteSize, IggyError, IggyExpiry, MaxTopicSize};
+
+/// A single resolved option entry.
+#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
+pub struct OptionValue {
+ /// The effective value, resolved by the admitting primary at creation.
+ pub value: HeaderValue,
+ /// Whether the client explicitly sent this key. Derived entries
+ /// (`explicit == false`) were filled from server defaults at admission
+ /// and would have resolved differently under other server configs.
+ pub explicit: bool,
+}
+
+/// Which provenance classes an encoded options block carries.
+///
+/// Responses split a resource's options into two blocks so `GetTopic` can say
+/// which values the client chose and which admission filled in.
+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+pub enum OptionsProvenance {
+ /// Every entry, whatever its provenance.
+ All,
+ /// Only the keys the client sent.
+ Explicit,
+ /// Only the keys admission resolved from server defaults.
+ Derived,
+}
+
+impl OptionValue {
+ /// Whether this entry belongs in a block encoded for `provenance`.
+ #[must_use]
+ pub const fn matches(&self, provenance: OptionsProvenance) -> bool {
+ match provenance {
+ OptionsProvenance::All => true,
+ OptionsProvenance::Explicit => self.explicit,
+ OptionsProvenance::Derived => !self.explicit,
+ }
+ }
+
+ #[must_use]
+ pub fn explicit(value: HeaderValue) -> Self {
+ Self {
+ value,
+ explicit: true,
+ }
+ }
+
+ #[must_use]
+ pub fn derived(value: HeaderValue) -> Self {
+ Self {
+ value,
+ explicit: false,
+ }
+ }
+}
+
+/// Options attached to a stream, topic, or user, keyed by option name.
+pub type ResourceOptions = BTreeMap<HeaderKey, OptionValue>;
+
+/// JSON form for [`ResourceOptions`].
+///
+/// [`HeaderKey`] serializes as a struct (`kind` + base64 `value`), and JSON
+/// object keys must be strings, so the derived impl fails outright with
+/// "key must be a string" the moment a map is non-empty. Every HTTP response
+/// carrying options goes through here instead, rendering the key as its plain
+/// text. Binary transports are unaffected: they never touch serde.
+pub mod resource_options_json {
+ use super::{HeaderKey, OptionValue, ResourceOptions};
+ use serde::de::Error as _;
+ use serde::{Deserialize, Deserializer, Serialize, Serializer};
+ use std::collections::BTreeMap;
+ use std::str::FromStr;
+
+ /// # Errors
+ ///
+ /// Propagates the serializer's own errors.
+ pub fn serialize<S: Serializer>(
+ options: &ResourceOptions,
+ serializer: S,
+ ) -> Result<S::Ok, S::Error> {
+ let readable: BTreeMap<String, &OptionValue> = options
+ .iter()
+ .map(|(key, value)| (String::from_utf8_lossy(key.as_bytes()).into_owned(), value))
+ .collect();
+ readable.serialize(serializer)
+ }
+
+ /// # Errors
+ ///
+ /// Returns a deserializer error when a key is not a valid option name.
+ pub fn deserialize<'de, D: Deserializer<'de>>(
+ deserializer: D,
+ ) -> Result<ResourceOptions, D::Error> {
+ let readable = BTreeMap::<String, OptionValue>::deserialize(deserializer)?;
+ readable
+ .into_iter()
+ .map(|(key, value)| {
+ HeaderKey::from_str(&key)
+ .map(|key| (key, value))
+ .map_err(D::Error::custom)
+ })
+ .collect()
+ }
+}
+
+/// Topic option catalog: the keys `CreateTopic` accepts.
+pub mod topic_option_keys {
+ /// Compression algorithm name (`none`, `gzip`).
+ pub const COMPRESSION_ALGORITHM: &str = "compression_algorithm";
+ /// Message expiry: `Uint64` micros or a humantime string (`7 days`).
+ pub const MESSAGE_EXPIRY: &str = "message_expiry";
+ /// Topic size cap: `Uint64` bytes or a byte-size string (`1 GiB`).
+ pub const MAX_TOPIC_SIZE: &str = "max_topic_size";
+ /// Segment size: `Uint64` bytes or a byte-size string (`128 MiB`).
+ /// Bounded: a 512-byte multiple within
+ /// [`super::MIN_TOPIC_SEGMENT_SIZE`]..=the server's segment ceiling.
+ pub const SEGMENT_SIZE: &str = "segment_size";
+ /// Whether writes to this topic's partitions fsync: `Bool`, or the
+ /// strings `true` / `false`.
+ pub const ENFORCE_FSYNC: &str = "enforce_fsync";
+ /// Flush the journal once it holds this many messages: `Uint32`.
+ /// Must be non-zero.
+ pub const MESSAGES_REQUIRED_TO_SAVE: &str = "messages_required_to_save";
+ /// Flush the journal once it holds this many bytes: `Uint64` or a
+ /// byte-size string. Paired with [`Self::MESSAGES_REQUIRED_TO_SAVE`];
+ /// whichever threshold trips first flushes.
+ pub const SIZE_OF_MESSAGES_REQUIRED_TO_SAVE: &str = "size_of_messages_required_to_save";
+ /// Reserve the segment's bytes up front on a filesystem that supports it:
+ /// `Bool`, or the strings `true` / `false`. Pairs with
+ /// [`Self::SEGMENT_SIZE`] -- preallocation reserves exactly that much, so
+ /// the two only make sense decided together.
+ pub const PREALLOCATE_SEGMENTS: &str = "preallocate_segments";
+}
+
+/// Values an absent topic option resolves to at admission.
+///
+/// These are the knobs' single source of truth: they used to live in
+/// `config.toml` (`[system.topic]`, `[system.partition]`, `[system.segment]`),
+/// which meant every one of them had two homes and an operator could not tell
+/// which won. A topic carries whatever it was created with; anything the
+/// client did not send resolves to the constant here and is persisted as a
+/// derived entry, so the effective value is always visible on `GetTopic`.
+///
+/// Each value matches what the shipped `config.toml` carried, so removing the
+/// keys changed no behavior for a topic created without options.
+pub const DEFAULT_PARTITIONS_COUNT: u32 = 1;
+/// `MaxTopicSize::Unlimited` (was `[system.topic] max_size = "unlimited"`).
+pub const DEFAULT_MAX_TOPIC_SIZE: u64 = u64::MAX;
+/// `IggyExpiry::NeverExpire` (was `[system.topic] message_expiry = "none"`).
+pub const DEFAULT_MESSAGE_EXPIRY: u64 = u64::MAX;
+/// 1 GiB (was `[system.segment] size = "1 GiB"`).
+pub const DEFAULT_SEGMENT_SIZE: u64 = 1024 * 1024 * 1024;
+/// Was `[system.partition] enforce_fsync = false`.
+pub const DEFAULT_ENFORCE_FSYNC: bool = false;
+/// Was `[system.partition] messages_required_to_save = 1024`.
+pub const DEFAULT_MESSAGES_REQUIRED_TO_SAVE: u32 = 1024;
+/// Opt-in, unlike the `[system.segment] preallocate = true` this replaced.
+///
+/// That default was never actually in force: the reservation ran through
+/// `compio::spawn_blocking`, which panics the shard because shard executors
+/// disable the blocking pool, so any deployment that worked at all had
+/// preallocation off. With the call fixed to run inline it reserves real
+/// extents, and `FALLOC_FL_KEEP_SIZE` against the 1 GiB default segment size
+/// means 1 GiB of disk per partition the moment it is created -- a full test
+/// sweep reserved 393 GB before this was flipped. A topic that wants the
+/// latency benefit asks for it with `preallocate_segments`.
+pub const DEFAULT_PREALLOCATE_SEGMENTS: bool = false;
+/// 1 MiB (was `[system.partition] size_of_messages_required_to_save`).
+pub const DEFAULT_SIZE_OF_MESSAGES_REQUIRED_TO_SAVE: u64 = 1024 * 1024;
+
+/// Every runtime knob at its default, for a partition built with no resolved
+/// topic options (simulator, unit tests).
+impl Default for TopicRuntimeDefaults {
+ fn default() -> Self {
+ Self {
+ segment_size: IggyByteSize::from(DEFAULT_SEGMENT_SIZE),
+ enforce_fsync: DEFAULT_ENFORCE_FSYNC,
+ messages_required_to_save: DEFAULT_MESSAGES_REQUIRED_TO_SAVE,
+ size_of_messages_required_to_save: IggyByteSize::from(
+ DEFAULT_SIZE_OF_MESSAGES_REQUIRED_TO_SAVE,
+ ),
+ preallocate_segments: DEFAULT_PREALLOCATE_SEGMENTS,
+ }
+ }
+}
+
+/// Smallest per-topic segment size. Segments far below the shipped default
+/// explode the per-partition segment count, which state transfer bounds via
+/// its manifest entry cap; a partition that crosses it becomes unservable
+/// for transfer, so the floor is enforced at admission.
+pub const MIN_TOPIC_SEGMENT_SIZE: u64 = 1024 * 1024;
+
+/// Validate an explicit per-topic `segment_size` against its bounds.
+///
+/// `ceiling` is node-derived: the smaller of the global segment maximum and
+/// the state-transfer artifact budget minus one bus frame (a segment may
+/// close one whole batch past its cap; an artifact ceiling below that
+/// refuses a legal segment and livelocks the partition's rejoin).
+///
+/// # Errors
+///
+/// Returns `IggyError::InvalidOptionValue("segment_size")` when the value is
+/// below [`MIN_TOPIC_SEGMENT_SIZE`], above `ceiling`, or not a 512-byte
+/// multiple.
+pub fn validate_topic_segment_size(size_bytes: u64, ceiling: u64) -> Result<(), IggyError> {
+ if size_bytes < MIN_TOPIC_SEGMENT_SIZE
+ || size_bytes > ceiling
+ || !size_bytes.is_multiple_of(512)
+ {
+ return Err(IggyError::InvalidOptionValue(
+ topic_option_keys::SEGMENT_SIZE.to_string(),
+ ));
+ }
+ Ok(())
+}
+
+/// Resource whose option catalog `DescribeOptions` serves.
+#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
+#[serde(rename_all = "snake_case")]
+pub enum OptionsScope {
+ Topic,
+ Stream,
+ User,
+}
+
+impl OptionsScope {
+ #[must_use]
+ pub fn as_code(&self) -> u8 {
+ match self {
+ Self::Topic => 1,
+ Self::Stream => 2,
+ Self::User => 3,
+ }
+ }
+
+ /// # Errors
+ ///
+ /// Returns `IggyError::InvalidCommand` for an unknown scope code.
+ pub fn from_code(code: u8) -> Result<Self, IggyError> {
+ match code {
+ 1 => Ok(Self::Topic),
+ 2 => Ok(Self::Stream),
+ 3 => Ok(Self::User),
+ _ => Err(IggyError::InvalidCommand),
+ }
+ }
+}
+
+impl std::fmt::Display for OptionsScope {
+ fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+ match self {
+ Self::Topic => write!(f, "topic"),
+ Self::Stream => write!(f, "stream"),
+ Self::User => write!(f, "user"),
+ }
+ }
+}
+
+impl FromStr for OptionsScope {
+ type Err = IggyError;
+
+ fn from_str(scope: &str) -> Result<Self, Self::Err> {
+ match scope {
+ "topic" => Ok(Self::Topic),
+ "stream" => Ok(Self::Stream),
+ "user" => Ok(Self::User),
+ _ => Err(IggyError::InvalidCommand),
+ }
+ }
+}
+
+/// One entry of a resource's option catalog, as served by `DescribeOptions`.
+#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
+pub struct OptionSpec {
+ /// Option key accepted by the create command.
+ pub key: String,
+ /// Canonical kind the server persists the value under. A `String` value
+ /// parsed via the server-config rules is always accepted too.
+ pub kind: HeaderKind,
+ /// This node's current resolved default in the canonical kind's
+ /// encoding; empty when the key has no server default.
+ #[serde(default)]
+ pub default_value: Vec<u8>,
+ /// Human-readable description.
+ #[serde(default)]
+ pub description: String,
+}
+
+/// Every key `CreateTopic` currently accepts. Unknown keys are rejected at
+/// the edge, never skipped: a silently ignored knob would hand the client
+/// server defaults without it ever learning.
+pub const TOPIC_OPTION_KEYS: &[&str] = &[
+ topic_option_keys::COMPRESSION_ALGORITHM,
+ topic_option_keys::MESSAGE_EXPIRY,
+ topic_option_keys::MAX_TOPIC_SIZE,
+ topic_option_keys::SEGMENT_SIZE,
+ topic_option_keys::ENFORCE_FSYNC,
+ topic_option_keys::MESSAGES_REQUIRED_TO_SAVE,
+ topic_option_keys::SIZE_OF_MESSAGES_REQUIRED_TO_SAVE,
+ topic_option_keys::PREALLOCATE_SEGMENTS,
+];
+
+/// The subset of [`TOPIC_OPTION_KEYS`] an `UpdateTopic` options block may
+/// carry.
+///
+/// These three used to be fixed fields of the update command, which meant one
+/// setting had two homes and an update always rewrote all three whether the
+/// caller meant to or not. As options they are patched: a key the client did
+/// not send keeps its current value.
+///
+/// The partition runtime knobs (`segment_size`, `enforce_fsync`, both flush
+/// thresholds, `preallocate_segments`) stay out, and not only because nothing
+/// re-pushes them to a live partition. They describe how a partition's storage
+/// was laid down: changing `segment_size` mid-segment leaves one segment sized
+/// by the old cap and the next by the new one, and `preallocate_segments` can
+/// only act on a file not yet opened. A topic gets them at creation and keeps
+/// them, so its segments stay uniform.
+pub const UPDATABLE_TOPIC_OPTION_KEYS: &[&str] = &[
+ topic_option_keys::COMPRESSION_ALGORITHM,
+ topic_option_keys::MESSAGE_EXPIRY,
+ topic_option_keys::MAX_TOPIC_SIZE,
+];
+
+/// Keys an `UpdateStream` options block may carry. Empty because streams have
+/// no catalog keys yet: the block exists so the first one costs a catalog
+/// entry instead of another wire change, and until then every key is rejected
+/// by name rather than stored and ignored.
+pub const UPDATABLE_STREAM_OPTION_KEYS: &[&str] = &[];
+
+/// Keys an `UpdateUser` options block may carry. Empty for the same reason as
+/// [`UPDATABLE_STREAM_OPTION_KEYS`].
+pub const UPDATABLE_USER_OPTION_KEYS: &[&str] = &[];
+
+/// Longest key echoed back in an option error. A rejected key is attacker-sized
+/// by definition, and the error rides the wire back to the client.
Review Comment:
"rides the wire back to the client" only holds for http - binary transports send just the error code and `from_repr` on the client rebuilds the variant with an empty string, so tcp/quic/ws users see "Unsupported option key: " with no key. since key rejection is the discovery mechanism for anyone not calling `describe_options`, carrying the key in the deny reply (or at least scoping this comment to http) matters.
--
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: commits-unsubscribe@iggy.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org