krabka_protocol_codegen/
validate.rs

1use thiserror::Error;
2
3use crate::ir::{FieldSpec, FlexibleVersions, MessageSpec, MessageType};
4
5#[derive(Debug, Error, PartialEq, Eq)]
6pub enum ValidateError {
7    #[error("{message}: in {context}")]
8    Unsupported {
9        message: &'static str,
10        context: String,
11    },
12}
13
14/// Field types the generator currently understands. Anything else is a hard
15/// error. This list is verified against the full Kafka 4.2.0 schema corpus of
16/// 197 files. The field-level primitive types found there are bool, bytes,
17/// float64, int16, int32, int64, int8, records, string, uint16, and uuid.
18/// `uint32` and `float32` are listed here for completeness, because they
19/// appear in older or future schemas, but they are not in the 4.2.0 corpus.
20const KNOWN_PRIMITIVE_TYPES: &[&str] = &[
21    "bool", "int8", "int16", "int32", "int64", "uint16", "uint32", "float64", "string", "bytes",
22    "uuid", "records",
23];
24
25/// # Errors
26/// Returns an error when the schema model is invalid or generated Rust cannot be formatted or written.
27pub fn validate(specs: &[MessageSpec]) -> Result<(), ValidateError> {
28    for spec in specs {
29        let ctx = spec.name.clone();
30        if matches!(
31            spec.message_type,
32            MessageType::Request | MessageType::Response
33        ) && spec.api_key.is_none()
34        {
35            return Err(ValidateError::Unsupported {
36                message: "request/response missing apiKey",
37                context: ctx,
38            });
39        }
40        if let Some(message) = latest_version_unstable_error(spec) {
41            return Err(ValidateError::Unsupported {
42                message,
43                context: ctx,
44            });
45        }
46        validate_fields(&spec.fields, spec.flexible_versions, &ctx)?;
47        for cs in &spec.common_structs {
48            validate_fields(
49                &cs.fields,
50                spec.flexible_versions,
51                &format!("{ctx}.{}", cs.name),
52            )?;
53        }
54    }
55    Ok(())
56}
57
58/// Kafka's `MessageSpec` accepts `latestVersionUnstable` only on a request.
59/// The flag marks the highest version in `validVersions`, so this function
60/// also requires a range that has a highest version. An empty range has no
61/// version to mark. An open-ended range has no fixed highest version.
62fn latest_version_unstable_error(spec: &MessageSpec) -> Option<&'static str> {
63    if !spec.latest_version_unstable {
64        return None;
65    }
66    if spec.message_type != MessageType::Request {
67        return Some("latestVersionUnstable on a message that is not a request");
68    }
69    if spec.valid_versions.is_empty() {
70        return Some("latestVersionUnstable with empty validVersions");
71    }
72    if spec.valid_versions.max == i16::MAX {
73        return Some("latestVersionUnstable with open-ended validVersions");
74    }
75    None
76}
77
78fn validate_fields(
79    fields: &[FieldSpec],
80    flexible: FlexibleVersions,
81    ctx: &str,
82) -> Result<(), ValidateError> {
83    for f in fields {
84        let context = format!("{ctx}.{}", f.name);
85        let base = base_type(&f.field_type);
86
87        let known = KNOWN_PRIMITIVE_TYPES.contains(&base)
88            || base.starts_with("[]")   // arrays (nested [] not stripped by base_type)
89            || is_struct_type(base); // struct reference like `MetadataRequestTopic`
90
91        if !known {
92            return Err(ValidateError::Unsupported {
93                message: "unknown field type",
94                context,
95            });
96        }
97
98        if f.tag.is_some() && !is_some_flexible(flexible) {
99            return Err(ValidateError::Unsupported {
100                message: "tagged field on non-flexible message",
101                context,
102            });
103        }
104
105        // Kafka's generator enforces both rules in `FieldSpec`. The JVM
106        // differential sweep depends on them too: it builds the borrowed
107        // tagged fixture at the highest version, where every tag is in range.
108        if f.tag.is_some() {
109            let tagged = f.tagged_versions.unwrap_or(f.versions);
110            if tagged.max != i16::MAX {
111                return Err(ValidateError::Unsupported {
112                    message: "taggedVersions is not open-ended",
113                    context,
114                });
115            }
116            if tagged.min < f.versions.min || f.versions.max != i16::MAX {
117                return Err(ValidateError::Unsupported {
118                    message: "taggedVersions is not a subset of versions",
119                    context,
120                });
121            }
122        }
123
124        if !f.fields.is_empty() {
125            validate_fields(&f.fields, flexible, &context)?; // flexible is Copy
126        }
127    }
128    Ok(())
129}
130
131fn is_some_flexible(f: FlexibleVersions) -> bool {
132    matches!(f, FlexibleVersions::Range(_))
133}
134
135fn base_type(t: &str) -> &str {
136    t.strip_prefix("[]").unwrap_or(t)
137}
138
139fn is_struct_type(t: &str) -> bool {
140    // Kafka schema convention: struct types are PascalCase identifiers.
141    t.chars().next().is_some_and(char::is_uppercase)
142}
143
144#[cfg(test)]
145mod tests {
146    use std::path::PathBuf;
147
148    use super::*;
149    use crate::ir;
150
151    #[test]
152    fn vendored_schemas_validate() {
153        let dir = PathBuf::from(env!("CARGO_MANIFEST_DIR"))
154            .parent()
155            .unwrap()
156            .join("protocol")
157            .join("schemas");
158        let specs = ir::load_dir(&dir).unwrap();
159        // If this test fails, the generator needs an update before we can target
160        // this Kafka release — surface the offending schema clearly.
161        validate(&specs).unwrap_or_else(|e| panic!("validation failed: {e}"));
162    }
163}
krabka_protocol_codegen/validate.rs

krabka_protocol_codegen/
validate.rs
