Crate krabka_macros

Expand description

Derive macros for the krabka broker.

krabka-macros turns one annotated declaration into code that the broker would otherwise write out by hand two or three times. It is built on [moxy], which supplies the token, syntax-tree, attribute-parsing and template layers.

§`RegisterMetrics`

#[derive(RegisterMetrics)] goes on a struct of prometheus-client metric handles. It adds two private methods to the struct:

fn unregistered() -> Self constructs every field: with Default::default(), with a histogram over buckets, or with the new expression.
fn register(&self, registry: &mut Registry) registers a clone of every field that is not skip, in field order. The order of the fields is the order of the families in the text exposition.

Each field takes one optional #[metric(...)] attribute:

help = "..." â€” the help text. Every registered field needs one.
name = "..." â€” the registered name. The default is the field name with a trailing _total removed, because prometheus-client appends _total to the name of a counter itself.
buckets = EXPR â€” construct a Histogram over EXPR. On a Family<_, Histogram> field, construct the family so that each of its histograms uses EXPR.
new = EXPR â€” construct the field with EXPR.
skip â€” construct the field, but do not register it.

#[derive(krabka_macros::RegisterMetrics)]
struct Metrics {
    #[metric(help = "Records received")]
    records_total: Counter,
    #[metric(help = "Request latency in seconds", buckets = [0.001, 0.01, 0.1])]
    latency_seconds: Family<ApiLabel, Histogram>,
}

§`human_units`

#[human_units] goes on a file-config struct, above its #[derive(Deserialize, Serialize, JsonSchema)]. Each field typed Option<Time>, Option<ByteSize> or Option<Ratio> (under any path prefix) gets two attributes:

#[serde(default, with = "krabka_units::serde_units::human::option_X")], so that the TOML file writes the value as a string with a unit, such as "5s" or "1MiB". default is left out when the field already has a serde default.
#[schemars(with = "Option<crate::file_config::schema_units::X>")], so that the JSON Schema names the unit in its format.

A field with its own serde with, deserialize_with or serialize_with keeps its codec and gets neither attribute. Every other field passes through unchanged.

#[krabka_macros::human_units]
#[derive(Deserialize, Serialize, JsonSchema)]
struct FileTimeouts {
    request_timeout: Option<Time>,
    segment_bytes: Option<ByteSize>,
}

§`dispatch_table!`

dispatch_table! { ... } goes in the broker’s dispatch registry module. It reads sections of CamelCase Kafka api names, each a label, a colon, and comma-separated entries closed by ;:

krabka_macros::dispatch_table! {
    context: Metadata, AddPartitionsToTxn => crate::txn::handlers::add_partitions_to_txn::handle;
    sync_context: DescribeConfigs;
    typed: ListGroups;
    typed_group: Heartbeat;
    typed_sync: DescribeAcls;
    typed_infallible: ListConfigResources;
    auth: CreateDelegationToken;
    telemetry: PushTelemetry;
}

From each name, such as CreateAcls, it derives the adapter create_acls_adapter, the ApiKey::CreateAcls variant, the request module krabka_protocol::owned::create_acls_request and its CreateAclsRequest type. The handler is crate::handlers::create_acls::handle unless the entry names another with => path.

context and sync_context adapters call handler(broker, version, body, ctx) on the raw body; a sync_context handler returns its result instead of a future. A handler that needs the request’s correlation id reads it from the context.
typed adapters decode the request, await handler(broker, request, version, ctx) for a Result<Response, BrokerError>, and encode the response with crate::handlers::encode_response. typed_group adapters do the same but decode through crate::handlers::decode_group_request.
typed_sync adapters decode the request, call handler(broker, &request, version, ctx) for a Result<Response, BrokerError> without awaiting it, encode the response, and wrap the result in a ready future. typed_infallible adapters do the same for a handler that returns the Response itself.
telemetry adapters pass a TelemetryContext and wrap a synchronous result.
auth entries generate no adapter: the hand-written <name>_adapter must be in scope.
typed_own_span generates the typed adapter without the span below, for a handler that keeps a #[tracing::instrument] of its own.

Every other generated adapter runs the decode, the handler and the encode inside an info span named handle_<snake_name>, with api = "<Name>", version and, for the context, sync_context and telemetry sections, req_bytes = body.len(). On Err it emits, inside that span, the event #[tracing::instrument(err)] would: ERROR with error set to the error’s Display. The expansion calls ::tracing, so the calling crate depends on it, and BrokerError implements Display.

It then emits fn register_dispatch_table(registry: &mut DispatchRegistry), which registers every entry at the request schema’s FLEXIBLE_MIN and panics on a duplicate registration. The expansion names Broker, ApiVersion, CorrelationId, RequestContext, TelemetryContext, BoxFuture, Bytes, BrokerError, ApiKey, DispatchEntry and DispatchRegistry unqualified, so the calling module imports them.

§`throttle_probes!`

throttle_probes! { ... } is an expression for the KIP-219 throttle-echo audit: an array of (API_KEY, probe as Probe), one per entry, keyed by the response module’s generated API_KEY. Its sections are throttled (the response has throttle_time_ms, which the probe sets to SENTINEL), unthrottled (it has none), and legacy_split: Name = version, whose probe encodes the kafka_3_6_2 response below version. The probes call position, ApiVersion, ThrottlePosition, SENTINEL and Probe from the calling module.

let probes: BTreeMap<_, _> = krabka_macros::throttle_probes! {
    throttled: Metadata, ListOffsets;
    unthrottled: SaslHandshake;
    legacy_split: Produce = 3, Fetch = 4;
}
.into_iter()
.collect();

§`RuntimeOverlay`

#[derive(RuntimeOverlay)] goes on a clap argument group whose fields overlay the same-named fields of a config struct. The struct names that config with #[overlay(target = Type)]. The derive adds pub(crate) fn copy_into(&self, target: &mut Type), which assigns every field in field order:

by default, target.f = self.f, so the field must be Copy;
with #[overlay(refined)], target.f = self.f.map(|value| value.into_value()), for an Option of a refined newtype;
with #[overlay(clone)], target.f.clone_from(&self.f);
with #[overlay(skip)], not at all.

A field that the target lacks, or whose type differs, fails to compile.

#[derive(clap::Args, krabka_macros::RuntimeOverlay)]
#[overlay(target = RuntimeFileConfig)]
struct RuntimeArgs {
    cleaner_interval: Option<Time>,
    #[overlay(refined)]
    num_partitions: Option<PositiveI32>,
}

§`krabka_env`

#[krabka_env] goes on a clap argument struct, above its #[derive(clap::Args)]. Each field without an #[arg(...)] of its own gets #[arg(long, env = "KRABKA_<FIELD>", ...)], where <FIELD> is the field name in upper case and the rest follows from the field type:

Field type	Added argument
`Option<Time>`	`value_parser = krabka_units::parse::positive_time`
`Option<ByteSize>`	`value_parser = krabka_units::parse::positive_byte_size`
`Option<Ratio>`	`value_parser = krabka_units::parse::positive_ratio`
`Option<PositiveCount>`	`value_parser = krabka_broker::config_value::parse_positive_count`
`Option<PositiveI16>`, `I32`, `I64`	the matching `parse_positive_*`
`Option<u32>`	`value_parser = clap::value_parser!(u32).range(1..)`
`Option<i64>`	`value_parser = clap::value_parser!(i64).range(0..)`
`Option<bool>`	`action = clap::ArgAction::Set`
`Option<i16>`, `Option<i32>`, `Option<String>`, `Option<PathBuf>`	none

The type is matched as written, so a field of any other type, or of one of these that needs another parser, keeps its own #[arg(...)]. Doc comments stay where they are and remain the help text.

#[krabka_env(prefix = "BENCH_")] puts every variable under that prefix in place of KRABKA_, so the field tls_ca_path reads BENCH_TLS_CA_PATH. A field whose variable is not its name under the prefix keeps its own #[arg(...)].

§`RefinedNewtype`

#[derive(RefinedNewtype)] goes on a tuple struct with one field, the value a refined_type rule checks. It adds new(value) -> Result<Self, E>, which validates through the rule, and a #[must_use] const getter that returns the value. Both take the struct’s visibility. One #[refined(...)] attribute configures it:

rule(<type>) â€” the refined_type rule, such as GreaterU32<0>. Required. It goes in parentheses because a generic type does not parse after =.
getter = name â€” the getter’s name. The default is into_value.
string_error â€” new returns String, the rule’s error text, rather than refined_type::result::Error<T>.
label = "..." â€” with string_error, new returns "<label>: <rule error>".
default = EXPR â€” implement Default as Self::new(EXPR), which panics when EXPR breaks the rule.
from_str â€” implement FromStr with Err = String: the field type’s own parse, then new, each error as its text.
display â€” implement Display as the field’s.
parse_fn = name â€” add a free fn name(&str) -> Result<Self, String> that parses the way from_str does.
quantity = ByteSize, Time or Frequency â€” the field is a whole number of bytes, milliseconds or Hz, and new takes the krabka_units quantity. It refuses a fraction, an infinity, or a count the field type cannot hold with "<label>: must be a whole number of <unit> that fits in <field type>", then hands the count to the rule. It implies string_error, adds TryFrom<quantity> through new, makes from_str parse with krabka_units::parse and display print the quantity’s human() text, and makes default take a quantity.
quantity_getter = name â€” with quantity, add a getter that returns the value as its quantity.

#[derive(Clone, Copy, krabka_macros::RefinedNewtype)]
#[refined(rule(GreaterU32<0>), string_error, label = "fetch miss limit", getter = get,
          default = 3, from_str, display)]
pub struct FetchMissLimit(u32);

#[derive(Clone, Copy, krabka_macros::RefinedNewtype)]
#[refined(rule(GreaterI32<0>), quantity = ByteSize, label = "fetch max", getter = bytes,
          quantity_getter = size, default = mebibytes(8), from_str, display)]
pub struct FetchMax(i32);

§`PrimitiveCmp`

#[derive(PrimitiveCmp)] goes on a tuple struct with one field. It implements PartialEq and PartialOrd between the struct and the field’s type in both directions, so that Seq(3) == 3_u64 and 2_u64 < Seq(3) compile.

§`EnumStr`

#[derive(EnumStr)] goes on an enum whose variants each stand for one fixed text, such as a metric label value or a config value. It adds a #[must_use] const fn as_str(self) -> &'static str with the enum’s visibility. The method takes &self when any variant has fields, which its arm matches with { .. } or (..). One optional #[enum_str(...)] attribute on the enum configures it:

case = "..." â€” how a variant name becomes its text: "snake_case", "kebab-case", "lowercase" or "UPPERCASE". Without it the text is the variant name unchanged.
as_str = name â€” the method’s name.
parse or parse = name â€” add fn parse(&str) -> Option<Self>, under that name, over the unit variants.
all â€” add const ALL: [Self; N], every variant in declaration order. Every variant must be a unit variant.
label_value â€” implement prometheus_client’s EncodeLabelValue as the text. The crate must depend on prometheus-client.

A variant’s own #[enum_str(...)] takes name = "...", its text in place of the cased name, and alias = "..." or alias("...", "..."), further text that parse accepts.

#[derive(Clone, Copy, krabka_macros::EnumStr)]
#[enum_str(case = "snake_case", as_str = config_name, parse = from_config_name)]
pub enum AssignorKind {
    Auto,
    #[enum_str(alias = "highly-available")]
    HighlyAvailable,
}

§`FieldDefaults`

#[derive(FieldDefaults)] goes on a struct with named fields in place of a hand-written impl Default that only fills each field with a fixed expression. Each field takes one optional #[default(EXPR)] attribute and starts as EXPR; a field without one starts as Default::default(). The expression is evaluated each time default() runs, so it may call functions, and it is written out as given, so it must name what it uses in the scope of the struct. The derive adds no bounds: a generic struct writes the bounds its fields need on the struct itself.

The name differs from Default because the standard derive owns the bare name and its own #[default] attribute on enum variants. Do not derive both on one struct.

#[derive(krabka_macros::FieldDefaults)]
pub struct FreezeConfig {
    #[default(Duration::from_secs(30))]
    pub poll_interval: Duration,
    #[default(true)]
    pub enabled: bool,
    pub owners: Vec<String>,
}

§`cli_main!`

cli_main!(path) goes at the top level of an operator binary’s main.rs, in place of its main. It expands to a #[tokio::main] async fn main() that installs a tracing_subscriber::fmt() subscriber filtered by RUST_LOG, or at info when RUST_LOG is unset or does not parse, and writes to stderr. stdout stays the binary’s own, so a report that a script parses, such as krabka-restore --report json, holds no log lines. It then awaits path::run_from_args(std::env::args_os()) and exits the process with the i32 it returns. Any arguments after the path, such as flavor = "multi_thread", go to #[tokio::main(...)] unchanged. The binary depends on tokio with its macros feature and on tracing-subscriber.

krabka_macros::cli_main!(krabka_restore, flavor = "multi_thread");

§Group timings and epoch values

#[derive(GroupTimingConfig)] adds for_group and the test-only assigning_at_once constructor to a membership config. Its named timing fields and bounds remain visible on the struct. Select the stored keys with #[group_timing(session = KEY, heartbeat = KEY, assignment = KEY)]. The implementation uses the broker’s existing group timing parser and bounds.

#[derive(GroupLifecycle)] adds epoch changes, session expiry queries, and member eviction to a group state. Select its existing policy with #[group_lifecycle(protocol = "consumer" | "share" | "streams")]. Consumer and share groups get assignment delay checks; consumer and streams groups get rebalance fencing. A successful streams epoch change marks the group dirty. All removals call the state’s own remove_member method.

#[derive(FlexibleEpochValue)] adds the version-zero epoch codec to a named struct with an i32 epoch followed by an i64 tagged value. #[epoch_value(record = "RecordName")] supplies the decoding error name. A zero tagged value is omitted on encode and restored on decode.

#[flexible_value(record = "KafkaRecord")] goes on an inherent impl with private encode_fields(self or &self, buf: &mut BytesMut) and decode_fields(buf: &mut &[u8]) -> Result<Self, BrokerError> methods. It retains those methods and adds the public framed codecs. The field methods keep wire order explicit; the encoder retains its receiver ownership.

#[derive(KafkaUuid)] adds display, parsing, and string serde to a UUID wrapper. It calls the declaration’s existing encode, decode, and KafkaUuidError helpers. Display retains the formatter’s padding rules.

§Handler admission

#[voter_admission(request = TYPE, api = KEY, gate = FUNCTION, refusals = EXPR)] goes on an ordinary async handle(broker, version, req_bytes, ctx) function. It runs the shared voter prelude before the operation body and returns early for an encoded refusal. The body retains its original tokens and scopes.

share_state_handler!(Request, Response, TopicRow, PartitionRow, serve) declares a share-state handler. A denied request returns an error row for every requested partition in request order. An allowed request calls serve with the share coordinator and a fresh metadata snapshot.

§Storage methods

#[derive(SparseIndex)] adds the existing loader and file operations to a sparse index. #[sparse_index(kind = "offset" | "time")] selects its raw entry layout and I/O target. The loader retains its monotonic-entry checks. #[index_open] adds the same tracing span to other ordinary index loaders.

#[remote_operation] traces an ordinary remote-storage SPI method. Its existing method name selects the span level and fields. It retains the native method body, parameter names, and synchronous signature.

#[model_type] attaches the existing proof and runtime derives to an ordinary struct or enum. proof(...) and runtime(...) replace only the selected derive list, preserving its path order and the declaration tokens.

#[sendfile_cfg] gates one ordinary item with the fetch path’s shared platform predicate. Apply it separately to imports, exports and methods.

#[decode_dispatch(request = TYPE, binding(req))] decodes a request before the handler’s native body. binding(mut req) retains an owned mutable request. decoded_dispatch! accepts the same explicit function, with [request = TYPE, binding(req)] after its parameters and before its return type.

#[extension_wire(name = "SPAN", api = "API", level = "info")] adapts an ordinary operation with broker, owned request, version and context parameters. It emits the raw five-argument entry point and traces both decoding and the native operation. "debug" retains the marker API’s tracing level.

worm_object_set! emits one ordered admission rule for its Model and Runtime declarations. Function headers and count observations stay native; proof projections remain inside the caller’s Pearlite scope.

§Direct runtime and response tables

set_runtime! applies optional fields in table order. Each group names a validator. plain assigns directly; duration validates positive time and converts it to Duration. A mapping selects a different target field, and in supplies the validator’s range argument. An absent field retains the target value. A validation error stops later assignments.

krabka_macros::set_runtime! {
    runtime => config;
    plain: enabled;
    duration: heartbeat_interval;
    in_range: max_size => group.max_size in 1..=1000;
}

response_errors! implements the caller’s error traits for external protocol types. Its four sections are response, code_response, row and resource_row. Each section contains type paths separated by commas inside braces. A resource row also implements ErrorRow, so each resource type is listed once. The traits remain in the caller’s scope.

metric_registry_fixture!(metrics, output, guard; expect("encode")) renders directly into the named bindings. The caller retains the registry lock until it drops guard. Use unwrap() to select that result policy.

broker_config! { pub struct BrokerConfig { ... } } constructs the broker config in one expansion. The fields stay in declaration order with their original tokens, documentation and attributes. It adds Debug and Clone.

#[membership_defaults] goes before FieldDefaults on a membership config. It assigns the shared defaults to the six named session and heartbeat fields. An existing default on one of those fields is an input error.

#[derive(ShareStateValue)] writes scalar fields in declaration order, followed by state batches and the delivery-count tag. The existing batch, version and tag helpers retain their byte layouts and validation. Supply method documentation with #[share_state(encode_doc = "...", decode_doc = "...")].

#[derive(FlexibleArrayValue)] wraps one Vec field in version-zero framing. #[array_value(record = "RecordName", encode = writer, decode = reader)] selects the existing array callbacks and decoding error name.

tagged_wire! and unthrottled_wire! construct exhaustive struct values. Both supply an empty tagged-field set; the second supplies zero throttle. Every remaining field stays required. Struct updates with .. are refused, and field expressions retain their original tokens and evaluation order.

Macros§

acknowledgement_batches
assignment_dirs_fixture
async_write_delegate
authorization_refusal
bench_failures
bench_main
benchmark_latency_fixture
benchmark_scenario_fixture
bind_retry_fixture
bootstrap_parser_fixture
bound_start_fixture
bounded_bfs
bounded_unit_modules
break_glass_golden_id
broker_config
broker_test_setup
capture_layer_fixture
cli_main
compacted_batch
compaction_record_fixture
compression_bytes_fixture
config_alter_handler
connect_cli_client
connection_policy
consumer_fetch_fixture
control_marker_fixture
controller_request
create_topic_fixture
decoded_dispatch
delete_topic_request
described_group_rows
directory_tree_fixture
dispatch_capacity_parser
dispatch_table
duration_cases_fixture
duration_parser_fixture
empty_endpoint_voters
empty_match_acl_filter
ephemeral_port_fixture
epoch_checkpoint_failure
epoch_log_view
epoch_millis_fn
epoch_record_batch_fixture
evicted_local_segments
export_segment_data_fixture
file_region_fixture
finalized_feature_fixture
fixture_constructors
flag_metadata_fixture
format_directories_fixture
forwardable_requests
frame_max_parser
frame_prefix_fixture
freeze_refusal_rules
guard_response
heartbeat_status
hydrate_member_epochs
index_floor_lookup
index_seed_helpers
java_string_producer_source
jvm_checkpoint_dump_fixture
jvm_principal_golden
keyed_record_batch_fixture
legacy_policy_fixture
member_identity_fixtures
member_key_boundaries
metric_registry_fixture
minio_fixture
model_properties
model_runner
model_transition
named_acl_gates
offset_fetch_rows
otlp_exporter
parsed_cli_entrypoint
partition_gate
patterned_bytes_fixture
private_dispatch_table
producer_batch_fixture
projection_identity
quorum_epoch_encoder
quorum_epoch_requests
reassignment_fields
rebuild_reconciliation_group
reconciliation_actions
reconciliation_next_state
record_batch_fixture
record_header_text
record_limit_fixture
record_strings
records_payload_match
refused_fetch_partition_fixture
registration_feature_projection
remote_metadata_missing
remote_segment_check
remote_segment_fixtures
remote_segment_transition_matrix
remote_started_segment
replication_target_guard
request_frame_fixture
request_test_identity
required_topics
response_errors
s3_archive_config_fixture
schedule_commit_changes
scram_client_first
scram_client_proof_fixture
scram_users_fixture
sendfile_match
sendfile_platform
set_runtime
share_consumption_driver
share_dlq_meters
share_first_fetch
share_fixture_schemas
share_group_denial
share_partition_results
share_refusal_matrix
share_state_handler
share_state_request
share_state_results
simulation_actions
simulation_earliest_timer
simulation_heartbeat
simulation_step
simulation_transport
single_partition_produce_fixture
single_replica_partition_fixture
snapshot_member_maps
snapshot_node_fixture
snapshot_topic_fixture
sorted_index_cursor
splitmix64_step
string_mutations
supported_feature_fixture
supported_features_fixture
sweep_loop
tagged_wire
test_authorizer
throttle_probes
timer_hooks
timestamped_batch
token_refusal_result
topic_record_fixture
unix_millis_fixture
unthrottled_wire
validate_archive_rows
vector_request_fixture
voter_endpoints
wal_capture_topic_fixture
wire_codec_helpers
wire_codes
wire_test_driver
worm_object_set
worm_segment_fixture