feat(fuzz): fuzz storage services via nisshi-sans-io's new arbitrary feature

.gitignore

@@ -31,3 +31,4 @@ tmp
 .ua/intermediate/
 .ua/diff-overlay.json
 .ua/.trash-*/
+fuzz/artifacts/fuzz_alter_user_scram_credentials_storage/

CHANGELOG.md

@@ -12,6 +12,10 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ## [Unreleased]
 
+### Added
+
+- `nisshi-sans-io` has an `arbitrary` feature that derives `arbitrary::Arbitrary` on every generated message type. The dev-only `fuzz` crate uses it to run a fuzzed request through each non-transaction storage service, and it gains a target for each of them.
+
 ### Changed
 
 - The S3 and in-memory (DynoStore) storage engine stores each consumer
@@ -180,3 +184,8 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
   `WSAGetLastError`, so `raw_os_error()` there is `WSAEAFNOSUPPORT` (10047),
   a different number, and a Windows host without IPv6 support failed to bind
   `[::]` instead of falling back to `0.0.0.0`.
+- A `Fetch` request with an invalid `isolation_level`, `max_wait_ms`, `min_bytes` or `max_bytes` gets `INVALID_REQUEST` on the response and on every requested partition, instead of an internal error. `FetchService` takes a `ValidatedFetchRequest`, which parses these fields once.
+- A `CreateTopics` topic with manual `assignments` and an explicit `num_partitions` or `replication_factor` gets `INVALID_REQUEST`, as in Kafka, instead of creating `num_partitions` partitions: a single request could ask the broker for millions and exhaust its memory, because the 10,000 partition cap counted only the assignments. `CreateTopicsService` takes a `ValidatedCreateTopicsRequest`, built by `CreateTopicsValidationLayer`.
+- A `ListOffsets` request with an invalid `isolation_level` gets `INVALID_REQUEST`, and one with an invalid timestamp gets `INVALID_TIMESTAMP` on that partition, instead of an internal error.
+- An `AlterUserScramCredentials` request with an unrecognized SASL mechanism gets `UNSUPPORTED_SASL_MECHANISM` on that item, instead of an internal error.
+- A `Fetch` or `Metadata` request for a topic with neither a name nor a topic id no longer panics.

CLAUDE.md

@@ -88,6 +88,8 @@ Cargo workspace (members, including the `fuzz` harness, are listed in the root `
 
 `nisshi-sans-io/build.rs` reads ~185 official Kafka JSON message descriptors from `nisshi-sans-io/message/*.json` and generates typed Rust structs for every request/response pair. **Do not manually edit generated files.** The message JSON files are from upstream Apache Kafka.
 
+The generator also emits `#[derive(arbitrary::Arbitrary)]` on every generated type, gated behind the `arbitrary` feature flag (`dep:arbitrary`, see `fn arbitrary_field_attribute` in `build.rs`) — this is what lets the `fuzz` crate construct arbitrary `Request`/`Response` values directly from fuzzer bytes instead of fuzzing only the wire decoder. A handful of fields need a hand-written `#[arbitrary(with = ...)]`/`#[arbitrary(default)]` attribute to stay sound (e.g. raw-byte fields route through `nisshi-sans-io/src/arbitrary_support.rs` rather than deriving); see the doc comment on `arbitrary_field_attribute` for the full list and why each one is needed.
+
 ### Service Layer Pattern (`nisshi-service`)
 
 Uses `rama` crate for Layer/Service composition:
@@ -144,6 +146,31 @@ Lake features: `parquet`, `iceberg`, `delta` - enable writing schema-backed topi
 - `nisshi-smoke-test` runs the real Kafka CLI tools (in a `$KAFKA_IMAGE` container) against a broker, and stays out of `just test`; run it with `just smoke <engine>`. Its containers use the host network, so on macOS it needs Docker Desktop with host networking turned on. Its tests talk to the shared broker through `KafkaCli::shared()`, or start their own with `Broker::isolated()`, and each uses its own topic and group names (`nisshi_smoke_test::unique::name`). A test that applies only to some engines is gated with `#[cfg(feature = "<engine>")]` or `#[cfg(any(...))]` on the engines' features, never with `cfg(not(...))`, because `just clippy` turns every feature on and would never compile it. The test files are grouped in folders under `tests/smoke/`: `main.rs` declares each folder's module, and `<folder>.rs` declares each file in it, which `every_test_file_is_declared` checks. S3 runs only `topic_admin::topic_lifecycle`, selected by the `smoke-s3` profile in `.config/nextest.toml`. A test that fails because of an open bug is marked `#[ignore = "<what is broken>"]`, or `#[cfg_attr(feature = "<engine>", ignore = "...")]` if it fails only on some engines. CI skips it, local runs still run it (`run.sh` passes `--run-ignored all` outside CI), and the change that fixes the bug removes the attribute
 - Single-file test targets with specific feature requirements (e.g. `nisshi-schema`'s `berg`) use `required-features` in their `Cargo.toml`
 
+## Fuzz Testing (`fuzz`)
+
+The `fuzz` crate (`cargo-fuzz` + `libfuzzer-sys`) is a dev-only workspace member, excluded from `just test-workspace`. CI runs its targets through `fuzz.yml` (see CI Pipeline below), and you can run any of them locally. It requires the nightly toolchain, which `just` targets invoke explicitly (`cargo +nightly fuzz ...`); you don't need to switch your default toolchain.
+
+```shell
+just cargo-fuzz +args        # thin wrapper: cargo +nightly fuzz {{args}}
+just fuzz-request-decode     # fuzz_request_decode, 60s
+just fuzz-member-metadata    # fuzz_member_metadata, 60s
+just fuzz-generate-seed      # (re)generate the corpus seed files, see below
+just fuzz-ci <tier> [target]  # the targets of a tier in fuzz/ci-targets.json, each for its budget
+
+cargo +nightly fuzz list                             # list all targets
+cargo +nightly fuzz run <target> -- -max_total_time=60   # fuzz one target
+cargo +nightly fuzz run <target> fuzz/artifacts/<target>/<crash-file>  # replay a saved crash
+```
+
+Three target shapes, all in `fuzz/fuzz_targets/`:
+- **`fuzz_request_decode` / `fuzz_response_decode` / `fuzz_varint` / `fuzz_deflated_batch` / `fuzz_member_metadata`** — fuzz raw bytes straight through the wire decoder/encoder, no `Storage` involved.
+- **`fuzz_<api>_storage`** (the majority, ~23 targets, one per Kafka API like `fuzz_fetch_storage`, `fuzz_create_topics_storage`) — build an arbitrary typed request (via `nisshi-sans-io`'s `arbitrary` feature) and run it end-to-end through the real `*Service` against an in-memory `StorageContainer` (storage starts empty every execution), to catch panics/invariant violations in service and storage logic rather than the decoder. One known exception: `fuzz_delete_records_storage` is left out of every tier in `fuzz/ci-targets.json`, so CI never runs it because `delete_records` is currently an unconditional `todo!()` in both `nisshi-storage-dynostore` and the libSQL backend of `nisshi-storage-sql`, so it panics on its first execution — a known pre-existing gap, kept for parity and ready to start finding real bugs once those stubs are filled in.
+- **`generate_seeds`** (plain `fn main`, not a fuzz target) — writes hand-built binary seed files into `fuzz/corpus/<target>/`; rerun it after adding a new seed case.
+
+`fuzz/src/lib.rs` provides the shared `check_request`/`check_response` helpers used by most storage targets: they encode a value at its API's latest version, decode it, re-encode the decoded value, and decode again — then assert the two decodes are equal. They deliberately do **not** assert the first decode equals the original fuzzed value: a sans-io codec is not required to be lossless against an arbitrary in-memory value (e.g. every generated sequence field is `Option<Vec<T>>` regardless of wire nullability, so a mandatory array left at `None` is wire-equivalent to `Some(vec![])` but not `PartialEq`). The round-trip property that must hold is that re-encoding an already-decoded (version-normalized) value decodes back to exactly the same thing. `normalize_request` returns that decoded value as a typed request, so a target can serve the service what a client could actually send instead of the raw fuzzed value (`fuzz_list_offsets_storage` does).
+
+`fuzz/corpus/`, `fuzz/artifacts/` (crash reproducers) and `fuzz/.gitignore`'s `target`/`coverage` are all untracked — rebuild the corpus locally with `just fuzz-generate-seed` plus whatever `cargo +nightly fuzz run` accumulates. If you fix a bug found by fuzzing, replay every existing artifact for that target (`cargo +nightly fuzz run <target> fuzz/artifacts/<target>/<crash-file>`) to confirm the fix, not just the one input that was being worked on.
+
 ## CI Pipeline
 
 GitHub Actions (`.github/workflows/ci.yml`) runs in two tiers, gated by `ci-gate`, the single required check that fans in every other job:
@@ -170,6 +197,7 @@ On a pull_request, a `changes` job in `ci.yml`, `codeql.yml`, `dependencies.yml`
 | `etc/schema/` | Sample schemas: `.avsc` (Avro), `.json` (JSON Schema), `.proto` (Protobuf) |
 | `nisshi-sans-io/message/` | Kafka JSON protocol descriptors (upstream, ~185 files) |
 | `nisshi-sans-io/build.rs` | Code generator: JSON descriptors -> Rust types |
+| `fuzz/` | `cargo-fuzz` targets exercising the sans-io codec and storage services (see Fuzz Testing) |
 
 ## Lint Configuration
 

Cargo.toml

@@ -77,6 +77,7 @@
 [workspace.dependencies]
 anyhow = "1.0.94"
 apache-avro = "0.21.0"
+arbitrary = { version = "1", features = ["derive"] }
 arrow = { version = "58" }
 async-trait = "0.1.86"
 backoff = {version = "0.4.0", features = ["tokio"]}
@@ -84,7 +85,7 @@
 cached = "4.0.1"
 chrono = "0.4"
 clap = { version = "4.5.57", features = ["derive", "env"] }
 condtype = "1.3.0"
 console = "0.16.2"
 convert_case = "0.12.0"
 crc-fast = "1.8.0"
@@ -163,7 +164,7 @@
 snap = "1.1.1"
 socket2 = "0.6.5"
 syn = { version = "3.0.3", features = ["extra-traits", "full"] }
 nisshi = { version = "0.7.0-pre.3", path = "nisshi", default-features = false }
 nisshi-auth = { version = "0.7.0-pre.3", path = "nisshi-auth", default-features = false }
 nisshi-broker = { version = "0.7.0-pre.3", path = "nisshi-broker", default-features = false }
 nisshi-cat = { version = "0.7.0-pre.3", path = "nisshi-cat", default-features = false }
@@ -194,7 +195,7 @@
 tokio-util = { version = "0.7", features = ["full"] }
 traceforge = "0.2"
 tracing = "0.1"
 tracing-core = { version = "0.1" }
 tracing-opentelemetry = "0.34.0"
 tracing-subscriber = { version = "0.3", features = ["env-filter", "json"] }
 turso = "0.8.2"

fuzz/Cargo.toml

@@ -11,7 +11,13 @@ cargo-fuzz = true
 [dependencies]
 bytes.workspace = true
 libfuzzer-sys.workspace = true
-nisshi-sans-io.workspace = true
+nisshi-sans-io = { workspace = true, features = ["arbitrary"] }
+nisshi-service.workspace = true
+nisshi-storage-dynostore.workspace = true
+nisshi-storage.workspace = true
+rama.workspace = true
+tokio.workspace = true
+url.workspace = true
 
 [[bin]]
 name = "generate_seeds"
@@ -58,3 +64,159 @@ path = "fuzz_targets/fuzz_varint.rs"
 test = false
 doc = false
 bench = false
+
+[[bin]]
+name = "fuzz_create_topics_storage"
+path = "fuzz_targets/fuzz_create_topics_storage.rs"
+test = false
+doc = false
+bench = false
+
+[[bin]]
+name = "fuzz_alter_user_scram_credentials_storage"
+path = "fuzz_targets/fuzz_alter_user_scram_credentials_storage.rs"
+test = false
+doc = false
+bench = false
+
+[[bin]]
+name = "fuzz_describe_user_scram_credentials_storage"
+path = "fuzz_targets/fuzz_describe_user_scram_credentials_storage.rs"
+test = false
+doc = false
+bench = false
+
+[[bin]]
+name = "fuzz_create_acls_storage"
+path = "fuzz_targets/fuzz_create_acls_storage.rs"
+test = false
+doc = false
+bench = false
+
+[[bin]]
+name = "fuzz_describe_acls_storage"
+path = "fuzz_targets/fuzz_describe_acls_storage.rs"
+test = false
+doc = false
+bench = false
+
+[[bin]]
+name = "fuzz_consumer_group_describe_storage"
+path = "fuzz_targets/fuzz_consumer_group_describe_storage.rs"
+test = false
+doc = false
+bench = false
+
+[[bin]]
+name = "fuzz_delete_groups_storage"
+path = "fuzz_targets/fuzz_delete_groups_storage.rs"
+test = false
+doc = false
+bench = false
+
+[[bin]]
+name = "fuzz_delete_records_storage"
+path = "fuzz_targets/fuzz_delete_records_storage.rs"
+test = false
+doc = false
+bench = false
+
+[[bin]]
+name = "fuzz_delete_topics_storage"
+path = "fuzz_targets/fuzz_delete_topics_storage.rs"
+test = false
+doc = false
+bench = false
+
+[[bin]]
+name = "fuzz_describe_cluster_storage"
+path = "fuzz_targets/fuzz_describe_cluster_storage.rs"
+test = false
+doc = false
+bench = false
+
+[[bin]]
+name = "fuzz_describe_configs_storage"
+path = "fuzz_targets/fuzz_describe_configs_storage.rs"
+test = false
+doc = false
+bench = false
+
+[[bin]]
+name = "fuzz_describe_groups_storage"
+path = "fuzz_targets/fuzz_describe_groups_storage.rs"
+test = false
+doc = false
+bench = false
+
+[[bin]]
+name = "fuzz_describe_topic_partitions_storage"
+path = "fuzz_targets/fuzz_describe_topic_partitions_storage.rs"
+test = false
+doc = false
+bench = false
+
+[[bin]]
+name = "fuzz_fetch_storage"
+path = "fuzz_targets/fuzz_fetch_storage.rs"
+test = false
+doc = false
+bench = false
+
+[[bin]]
+name = "fuzz_find_coordinator_storage"
+path = "fuzz_targets/fuzz_find_coordinator_storage.rs"
+test = false
+doc = false
+bench = false
+
+[[bin]]
+name = "fuzz_incremental_alter_configs_storage"
+path = "fuzz_targets/fuzz_incremental_alter_configs_storage.rs"
+test = false
+doc = false
+bench = false
+
+[[bin]]
+name = "fuzz_init_producer_id_storage"
+path = "fuzz_targets/fuzz_init_producer_id_storage.rs"
+test = false
+doc = false
+bench = false
+
+[[bin]]
+name = "fuzz_list_groups_storage"
+path = "fuzz_targets/fuzz_list_groups_storage.rs"
+test = false
+doc = false
+bench = false
+
+[[bin]]
+name = "fuzz_list_offsets_storage"
+path = "fuzz_targets/fuzz_list_offsets_storage.rs"
+test = false
+doc = false
+bench = false
+
+[[bin]]
+name = "fuzz_list_partition_reassignments_storage"
+path = "fuzz_targets/fuzz_list_partition_reassignments_storage.rs"
+test = false
+doc = false
+bench = false
+
+[[bin]]
+name = "fuzz_metadata_storage"
+path = "fuzz_targets/fuzz_metadata_storage.rs"
+test = false
+doc = false
+bench = false
+
+[[bin]]
+name = "fuzz_produce_storage"
+path = "fuzz_targets/fuzz_produce_storage.rs"
+test = false
+doc = false
+bench = false
+
+[lints]

fuzz/README.md

@@ -20,6 +20,8 @@ To move a target between tiers, change its budget, or stop running it in CI, edi
 
 The PR job has a 50 minute timeout and each nightly job 75 minutes. The six `pr` budgets add up to 6 minutes of fuzzing, after a build that takes about 33 minutes from scratch (10 of it dependencies, which the Rust cache saves from `main`) and, measured on a PR run with that cache, about 19 minutes. If the budgets outgrow those, raise `timeout-minutes` in `.github/workflows/fuzz.yml` too.
 
+The `fuzz_<api>_storage` targets run in `nightly` only, 15 minutes each: the `pr` job builds every target and runs the listed ones back to back, and its time is the sum of its budgets. `fuzz_delete_records_storage` is in no tier on purpose: `delete_records` is an unconditional `todo!()` in `nisshi-storage-dynostore` and in the libSQL and turso backends of `nisshi-storage-sql`, so it panics on its first execution. Add it to a tier when those stubs are filled in. The "in neither tier" warning for it is expected until then.
+
 ## Adding a target
 
 1. Add `fuzz_targets/<name>.rs` and a `[[bin]]` entry in `Cargo.toml` (copy an existing one).

fuzz/ci-targets.json

@@ -9,8 +9,29 @@
   },
   "nightly": {
     "fuzz_batch_records": 1800,
+    "fuzz_alter_user_scram_credentials_storage": 900,
+    "fuzz_consumer_group_describe_storage": 900,
+    "fuzz_create_acls_storage": 900,
+    "fuzz_create_topics_storage": 900,
     "fuzz_deflated_batch": 1800,
+    "fuzz_delete_groups_storage": 900,
+    "fuzz_delete_topics_storage": 900,
+    "fuzz_describe_acls_storage": 900,
+    "fuzz_describe_cluster_storage": 900,
+    "fuzz_describe_configs_storage": 900,
+    "fuzz_describe_groups_storage": 900,
+    "fuzz_describe_topic_partitions_storage": 900,
+    "fuzz_describe_user_scram_credentials_storage": 900,
+    "fuzz_fetch_storage": 900,
+    "fuzz_find_coordinator_storage": 900,
+    "fuzz_incremental_alter_configs_storage": 900,
+    "fuzz_init_producer_id_storage": 900,
+    "fuzz_list_groups_storage": 900,
+    "fuzz_list_offsets_storage": 900,
+    "fuzz_list_partition_reassignments_storage": 900,
     "fuzz_member_metadata": 1800,
+    "fuzz_metadata_storage": 900,
+    "fuzz_produce_storage": 900,
     "fuzz_request_decode": 1800,
     "fuzz_response_decode": 1800,
     "fuzz_varint": 1800

fuzz/dict/fuzz_alter_user_scram_credentials_storage.dict

@@ -0,0 +1,8 @@
+"a\000"
+"\025\000\000\000\000\000\000\000"
+"\001\001"
+"\001\000\000\000\000\000\000\000"
+"\377\020\000\000"
+"\002\000\000\000\000\000\000\000"
+"3\001\000\000\000\000\000\000"
+"\013\000"