# 🗂️ schemreg
[](https://crates.io/crates/schemreg)
[](https://docs.rs/schemreg)
[](https://github.com/hupe1980/schemreg/actions/workflows/ci.yml)
[](LICENSE)
[](https://releases.rs/docs/1.88.0/)
Async schema registry client for **Confluent Schema Registry** (and Karapace), **Apicurio Registry v3**, and **AWS Glue Schema Registry**, with:
- ⚡ Zero-copy wire-format encode / decode (Confluent 5-byte header, Glue 18-byte header)
- 🚀 Transparent in-memory caching with thundering-herd coalescing
- 🔌 Pluggable backend via the `SchemaRegistryClient` trait
- 🎯 Feature-gated: pull in only what you need
---
## ✨ Features
| *(none)* | 🔧 Core types, wire format helpers, traits, Glue wire format |
| `confluent` | 🌐 Confluent HTTP client, encoder, decoder, TLS via rustls + webpki-roots |
| `apicurio` | 🗂️ Native Apicurio Registry v3 HTTP client (`ApicurioSchemaRegistry`) |
| `glue` | ☁️ AWS Glue SDK client (`AwsGlueSchemaRegistry`), ZLIB compression via flate2 |
| `avro` | 🪶 Avro encode / decode via apache-avro, works with any `SchemaRegistryClient` |
| `json` | 📋 JSON Schema encode / decode, works with any `SchemaRegistryClient` |
| `native-tls-roots` | 🔒 rustls-native-certs (implies `confluent`) |
| `aws-lc-rs` | 🔑 AWS LC crypto backend instead of ring (implies `confluent`) |
---
## 🚀 Quick start
### 🌐 Confluent Schema Registry
```toml
# Cargo.toml
[dependencies]
schemreg = { version = "0.3", features = ["confluent"] }
tokio = { version = "1", features = ["full"] }
bytes = "1"
```
```rust
use std::sync::Arc;
use schemreg::{
CachedSchemaRegistry, EncodeTarget, SchemaType, SubjectNameStrategy,
confluent::{ConfluentSchemaEncoder, ConfluentSchemaRegistry},
decoder::WireFormatDecoder,
traits::SchemaEncoder,
};
use bytes::Bytes;
#[tokio::main]
async fn main() -> Result<(), Box<dyn std::error::Error>> {
// Build the HTTP client.
let registry = ConfluentSchemaRegistry::builder()
.url("http://localhost:8081")
// .basic_auth("user", "password")
.build()?;
// Wrap with a bounded in-memory cache (1 000-entry capacity, FIFO eviction).
let cached = Arc::new(CachedSchemaRegistry::new(registry));
// Producer side: register schema on first send, then reuse cached ID.
let encoder = ConfluentSchemaEncoder::builder()
.registry(&*cached)
.schema(r#"{"type":"record","name":"Order","fields":[{"name":"id","type":"string"}]}"#, SchemaType::Avro)
.strategy(SubjectNameStrategy::TopicName)
.build()?;
let raw_bytes = Bytes::from_static(b"\x04\x08some-avro-payload");
let framed = encoder.encode(raw_bytes, "orders", None, EncodeTarget::Value).await?;
// Consumer side: strip the header (uses the same cached registry for schema lookup).
let decoder = WireFormatDecoder::confluent(Arc::clone(&cached));
let msg = decoder.decode(framed).await?;
println!("Decoded {} bytes", msg.payload.len());
Ok(())
}
```
### ☁️ AWS Glue Schema Registry
```toml
[dependencies]
schemreg = { version = "0.3", features = ["glue"] }
tokio = { version = "1", features = ["full"] }
bytes = "1"
```
```rust
use schemreg::{
GlueCompression, GlueDataFormat, GlueSchemaVersionId,
encode_glue_wire_format, decode_glue_wire_format,
};
fn main() -> Result<(), Box<dyn std::error::Error>> {
let version_id: GlueSchemaVersionId =
"550e8400-e29b-41d4-a716-446655440000".parse()?;
// Encode — adds the 18-byte Glue header.
let framed = encode_glue_wire_format(version_id, b"avro bytes", GlueCompression::None)?;
// Decode — strips the header.
let (id, payload) = decode_glue_wire_format(&framed)?;
assert_eq!(id, version_id);
assert_eq!(payload, b"avro bytes");
Ok(())
}
```
With the `glue` feature and real AWS credentials you can use the high-level `AwsGlueSchemaRegistry` client:
```rust
use aws_config::BehaviorVersion;
use schemreg::glue::AwsGlueSchemaRegistry;
let config = aws_config::load_defaults(BehaviorVersion::latest()).await;
let registry = AwsGlueSchemaRegistry::from_config(&config);
```
---
## 🔬 Wire formats
### 🌐 Confluent Schema Registry
```
Byte offset 0 1 2 3 4 5 …
┌────────┬────────────────────────────┬──────────────────────┐
│ 0x00 │ schema_id (u32 BE) │ payload (N bytes) │
└────────┴────────────────────────────┴──────────────────────┘
magic │←──────── 4 bytes ──────────→│
```
### ☁️ AWS Glue Schema Registry
```
Byte offset 0 1 2 18 18 …
┌────────┬────────┬────────────────────┬──────────────────────┐
│ 0x03 │ comp │ schema_version_id │ payload (N bytes) │
└────────┴────────┴────────────────────┴──────────────────────┘
version │byte │←───── 16 bytes ─────→│
```
- **comp**: `0x00` = none, `0x05` = ZLIB
- **schema_version_id**: 128-bit UUID stored as big-endian bytes
---
## 🔌 Custom backend
Any struct that implements `SchemaRegistryClient` can be used as the backend:
```rust
use std::sync::Arc;
use schemreg::{Schema, SchemaId, SchemaReference, SchemaRegistryClient, SchemaType, SchemaVersion};
use schemreg::error::Result;
struct MyRegistry { /* ... */ }
impl SchemaRegistryClient for MyRegistry {
async fn get_schema_by_id(&self, id: SchemaId) -> Result<Arc<Schema>> {
todo!()
}
async fn get_latest_schema(&self, subject: &str) -> Result<Arc<Schema>> {
todo!()
}
async fn get_schema_by_version(&self, subject: &str, version: SchemaVersion) -> Result<Arc<Schema>> {
todo!()
}
async fn register_schema(
&self, subject: &str, schema: &str,
schema_type: SchemaType, references: &[SchemaReference],
) -> Result<SchemaId> {
todo!()
}
}
// Wrap with cache transparently:
use schemreg::CachedSchemaRegistry;
let cached = CachedSchemaRegistry::new(MyRegistry { /* ... */ });
```
See [`examples/custom_backend.rs`](examples/custom_backend.rs) for a full working example.
---
## ⚡ Cache management
`CachedSchemaRegistry` exposes the `AnySchemaCache` trait for lifecycle control:
```rust
use schemreg::CachedSchemaRegistry;
let cached = CachedSchemaRegistry::with_max_entries(my_registry, 512);
// Pre-warm for known schema IDs (avoids cold-miss latency on startup).
cached.warm_cache([1u32, 2u32, 3u32]).await?;
println!("cached: {}", cached.cache_len());
// Invalidate a single stale entry.
cached.invalidate(2u32);
// Wipe everything.
cached.clear_cache();
```
---
## 🔍 Format-agnostic decoding
`WireFormatDecoder` auto-detects the wire format and dispatches to the correct backend:
```rust
use std::sync::Arc;
use schemreg::decoder::WireFormatDecoder;
let decoder = WireFormatDecoder::new()
.with_confluent(Arc::clone(&cached_confluent_registry))
.with_glue(Arc::clone(&cached_glue_registry)); // requires `glue` feature
let msg = decoder.decode(raw_bytes).await?;
println!("format: {:?}", msg.schema_format); // Avro / Protobuf / Json / Unknown
println!("payload: {} bytes", msg.payload.len());
```
---
## 📖 Examples
| [`confluent_encode_decode`](examples/confluent_encode_decode.rs) | 🌐 Full encode→decode round-trip with an in-memory stub registry |
| [`avro_roundtrip`](examples/avro_roundtrip.rs) | 🪶 End-to-end Avro encode → Confluent wire format → decode |
| [`glue_roundtrip`](examples/glue_roundtrip.rs) | ☁️ Glue wire format encode / decode, optional ZLIB compression |
| [`custom_backend`](examples/custom_backend.rs) | 🔌 Implementing `SchemaRegistryClient` + cache + WireFormatDecoder |
| [`apicurio_roundtrip`](examples/apicurio_roundtrip.rs) | 🗂️ Apicurio Registry v3 encode→decode round-trip with a mock registry |
```bash
cargo run --example confluent_encode_decode --features confluent
cargo run --example avro_roundtrip --features avro
cargo run --example glue_roundtrip --features glue
cargo run --example custom_backend
cargo run --example apicurio_roundtrip --features apicurio
```
---
## 🛠️ Full API surface — `SchemaRegistryClient` trait
All implementations (`ConfluentSchemaRegistry`, `ApicurioSchemaRegistry`, and any custom backend) expose the same methods through the `SchemaRegistryClient` trait. `CachedSchemaRegistry` adds transparent caching and delegates all calls to the inner client.
| `get_schema_by_id(id)` | Fetch schema by its globally unique integer ID |
| `get_latest_schema(subject)` | Fetch the most recently registered version under a subject |
| `get_schema_by_version(subject, version)` | Fetch a specific version under a subject |
| `register_schema(subject, schema, type, refs)` | Register a new schema (idempotent — returns existing ID if already registered) |
| `check_compatibility(subject, schema, type, refs)` | Test whether a schema is compatible with the currently registered version |
| `check_compatible(subject, schema, type)` | Convenience alias for `check_compatibility` with no schema references |
| `delete_subject(subject, permanent)` | Delete all versions of a subject (`permanent = true` for hard delete) |
| `get_subjects()` | List all registered subjects |
| `get_versions(subject)` | List all registered version numbers for a subject |
| `health_check()` | Probe the registry for connectivity (lightweight ping — backend-specific endpoint) |
| `set_compatibility(subject, level)` | Set the per-subject compatibility policy |
| `get_compatibility(subject)` | Get the current compatibility policy for a subject |
### Compatibility levels
`CompatibilityLevel` supports all Confluent / Apicurio policies:
```rust
use schemreg::CompatibilityLevel;
registry.set_compatibility("orders-value", CompatibilityLevel::BackwardTransitive).await?;
let level = registry.get_compatibility("orders-value").await?;
```
Available variants: `Backward`, `BackwardTransitive`, `Forward`, `ForwardTransitive`, `Full`, `FullTransitive`, `None`.
### 12-Factor / environment variable configuration
Both `ConfluentSchemaRegistryBuilder` and `ApicurioSchemaRegistryBuilder` support `from_env()`:
```rust
// Reads SCHEMA_REGISTRY_URL, SCHEMA_REGISTRY_USERNAME/PASSWORD or SCHEMA_REGISTRY_BEARER_TOKEN
let registry = ConfluentSchemaRegistryBuilder::from_env()?.build()?;
// Reads APICURIO_REGISTRY_URL, APICURIO_REGISTRY_USERNAME/PASSWORD or APICURIO_REGISTRY_BEARER_TOKEN
let registry = ApicurioSchemaRegistryBuilder::from_env()?.build()?;
```
---
## 🔄 Retry and resilience
`schemreg` has **built-in retry logic** for all HTTP requests — no extra configuration is needed:
| HTTP 429 (rate limited) | Retried; `Retry-After` header is honored (server-dictated delay) |
| HTTP 5xx (server error) | Retried with exponential backoff |
| Network errors | Retried (connection reset, timeout, DNS) |
| Max retries | 3 attempts; final error propagated |
| Backoff | 100 ms base, doubles per attempt, capped at 60 s |
No configuration is needed. To set independent connection and request timeouts:
```rust
use std::time::Duration;
use schemreg::confluent::ConfluentSchemaRegistryBuilder;
let registry = ConfluentSchemaRegistryBuilder::default()
.url("https://registry.example.com")
.connect_timeout(Duration::from_secs(3)) // TCP connection timeout
.request_timeout(Duration::from_secs(30)) // full request timeout
.pool_max_idle_per_host(10) // max idle connections per host
.build()?;
```
---
## 📄 License
Licensed under either of:
- Apache License, Version 2.0 ([LICENSE-APACHE](LICENSE-APACHE) or <http://www.apache.org/licenses/LICENSE-2.0>)
- MIT License ([LICENSE-MIT](LICENSE-MIT) or <http://opensource.org/licenses/MIT>)
at your option.