A blazing-fast, async, sans-io Apache Pulsar client for Rust.

Status: pre-alpha. The wire protocol layer is feature-rich, the tokio engine is usable end-to-end with supervised reconnect + transparent producer/consumer rebuild, and the moonpool engine carries client/producer/consumer for deterministic-simulation testing. API is unstable. Do not depend on this in production.

Magnetar is a from-scratch Apache Pulsar client driver written in Rust. It mirrors the surface area of the Apache Pulsar Java client and adds two properties that the Java client cannot reach:

1. Sans-io core. The protocol state machine (magnetar-proto) is a pure, quinn-proto-style state machine — handle_bytes in, poll_transmit out, poll_event for semantic events, poll_timeout for timers. Zero I/O dependencies. No tokio. No async. No sockets. It is feed-only.
2. Multiple swappable engines. The same sans-io state machine is driven by a production tokio engine (magnetar-runtime-tokio) and by a deterministic simulation engine (magnetar-runtime-moonpool) for chaos testing of reconnects, partitions, and TLS handshake reorderings under reproducible seeds.

The architecture explicitly bans channels (mpsc, broadcast, watch, oneshot, crossbeam-channel, flume, async-channel, …). The wake-up mechanism is Arc<parking_lot::Mutex<State>> plus tokio::sync::Notify plus core::task::Waker slabs inside the state machine. See ARCHITECTURE.md for the full rationale.

Magnetar is independent of the existing pulsar-rs crate — it shares neither code nor dependencies. The goal is feature-complete parity with the Apache Pulsar Java client.

• Protocol coverage: producer, consumer, reader, partitioned producer, partitioned consumer, multi-topics consumer, pattern (regex) consumer, table view, transactions.
• PIPs implemented or partially wired: PIP-4 (end-to-end encryption), PIP-30 / PIP-292 (in-band AUTH_CHALLENGE refresh), PIP-31 (transactions), PIP-37 (chunking + redelivery backoff), PIP-54 (partial-batch ACK), PIP-87 (AutoConsumeSchema broker lookup), PIP-90 (broker-entry metadata), PIP-121 (cluster failover — ServiceUrlProvider + ControlledClusterFailover
  • AutoClusterFailover), PIP-145 (regex topic discovery), PIP-188 (TOPIC_MIGRATED with supervised reconnect), PIP-313 (force unsubscribe). See Supported PIPs.
• Resilience: supervised reconnect with Connection::reset (Stage 2) + transparent producer / consumer rebuild via rebuild_producers / rebuild_consumers (Stage 3) + memory_limit runtime enforcement (Java MemoryLimitPolicy::FailImmediately) + global publish-bytes accounting via AtomicU64 CAS in Producer::send with release on Drop.
• Observability: cumulative counters + hdrhistogram p50/p99/max latency
  • rolling-window msgs/sec + bytes/sec rates (record_rate_window). OpenTelemetry context propagation (traceparent/tracestate via message properties, feature = "opentelemetry", ADR-0053).
• Transports: TCP, TLS 1.3 (rustls-only — no native-tls, no openssl), binary proxy (proxy_to_broker_url), pluggable DNS (DnsResolver trait + TokioDnsResolver default routed through Transport::connect).
• TLS knobs: tls_trust_certs_file_path, tls_allow_insecure_connection (blanket override), tls_hostname_verification_enable(false) paired with a PEM trust store (chain-on / hostname-off via custom rustls verifier).
• Schemas: bytes, string, JSON, Avro, Protobuf, Protobuf-native, KeyValue, Auto-consume, Auto-produce-bytes, plus the full primitive family — Int8, Int16, Int32, Int64, Float, Double, Bool, Date, Time, Timestamp, LocalDate, LocalTime, Instant, LocalDateTime.
• Compression: LZ4, ZSTD, Snappy, ZLIB.
• Auth providers: token, mTLS (the two stock providers in magnetar-proto::auth), OAuth2 ClientCredentialsFlow (working — fetches
  • caches + auto-refreshes JWTs against a standard OIDC token endpoint), SASL PLAIN (RFC 4616, working), SASL Kerberos / GSSAPI via libgssapi under the auth-sasl-kerberos feature (working — multi-round AUTH_CHALLENGE initiate loop), Athenz with a pre-fetched role token (AthenzProvider::with_role_token, working), and the opt-in Athenz ZTS round-trip (auth-athenz-zts, working via AthenzProvider::with_default_signer or a custom zts::JwtSigner / zts::ZtsClient).
• Trackers: ack grouping, unacked-message tracker (ack timeout + redelivery), negative-ack tracker with MultiplierRedeliveryBackoff (PIP-37), batch-index ACK set (PIP-54).
• Interceptors: ProducerInterceptor + ConsumerInterceptor SPIs.
• Admin REST client: a reqwest-backed admin client lives in magnetar-admin.
• CLI: magnetar binary in magnetar-cli covers admin lookups and stats today; data-plane subcommands (produce / consume / inspect) are in progress.

Magnetar is not yet on crates.io. Use the Git path until the first release:

[dependencies]
magnetar = { git = "https://github.com/CleverCloud/magnetar", branch = "main" }

The default feature set enables the tokio engine. The feature flags catalog:

| experimental-v5-client | no | Enables the PIP-466 V5 wrapper surface (magnetar::v5) over the v4 wire commands. | | scalable-topics | no | Enables the experimental PIP-460 scalable-topic scaffold (topic://, DAG watch, StreamConsumer, topic-info). | | crypto-aws-lc-rs | yes | rustls crypto provider: aws-lc-rs; brings post-quantum hybrid KEX (X25519MLKEM768). See TLS crypto provider. | | crypto-ring | no | rustls crypto provider: ring. | | crypto-openssl | no | rustls crypto provider: rustls-openssl (wraps system OpenSSL via deny.toml carve-out). | | crypto-fips | no | rustls crypto provider: aws-lc-rs FIPS-validated module (requires cmake + C toolchain). |

The workspace ships eleven crates:

xtask is a workspace member but is not published — it hosts build helpers (protoc codegen, e2e driver, dependency audits).

The rustls crypto backend is selected at compile time via four mutually-pluggable Cargo features on the magnetar façade. The wire protocol — TLS 1.3 (default) / TLS 1.2 — is identical across every provider; what differs is the audited / FIPS-validated / post-quantum posture of the underlying primitives.

# Pick a single provider (mutually exclusive at build time).
cargo build -p magnetar --no-default-features --features tokio,crypto-aws-lc-rs
cargo build -p magnetar --no-default-features --features tokio,crypto-ring
cargo build -p magnetar --no-default-features --features tokio,crypto-openssl   # needs system OpenSSL
cargo build -p magnetar --no-default-features --features tokio,crypto-fips      # needs cmake + C toolchain

# The `magnetar` binary (magnetar-cli) mirrors the same cascade — the
# admin REST client (reqwest + rustls) and the data-plane runtime both
# bind to the selected provider. `cargo build -p magnetar-cli` alone
# defaults to `crypto-aws-lc-rs`.
cargo build -p magnetar-cli --no-default-features --features crypto-ring

Under cargo build --workspace --all-features the compile-time cfg cascade resolves to aws-lc-rs (highest priority). Single-provider builds go through cargo run -p xtask -- check-crypto-matrix. A single compile_error! fires if no crypto-* feature is enabled.

The crypto-aws-lc-rs default picks up rustls 0.23's built-in prefer-post-quantum feature, so the wire client negotiates the X25519MLKEM768 hybrid key exchange with brokers that support it.

openssl / openssl-sys are admitted only as transitive deps of rustls-openssl; the rest of ADR-0005 (no native-tls, rustls everywhere) stays in force. See ADR-0035 for the binding decision.

The magnetar binary exposes a sozu / systemd-style identification banner:

$ magnetar --version
magnetar 0.1.0-dev.0 (a1b2c3d4e5f6-dirty)
built 2026-05-26T14:32:11Z · profile=release · rustc=rustc 1.88.0 (…) · target=x86_64-unknown-linux-gnu
features: +default
pulsar wire protocol: v21
os: linux · report bugs at https://github.com/CleverCloud/magnetar

• -V prints a single-line, never-colorized form: magnetar 0.1.0-dev.0 (sha-dirty).
• --version prints the multi-line form above, colorized on a TTY. NO_COLOR=1 or piping suppresses ANSI (https://no-color.org).
• SOURCE_DATE_EPOCH=<unix-seconds> pins the build timestamp for reproducible builds.

Full reference: docs/cli.md.

The high-level PulsarClient builder is the public entry point. It wires the tokio engine to the sans-io state machine and gives you producer / consumer / reader / table-view / partitioned / multi-topics / pattern builders.

use magnetar::{OutgoingMessage, PulsarClient};

# async fn run() -> Result<(), Box<dyn std::error::Error>> {
let client = PulsarClient::builder()
    .service_url("pulsar://localhost:6650")
    .build()
    .await?;

let producer = client
    .producer("persistent://public/default/orders")
    .name("orders-writer")
    .compression(magnetar_proto::types::CompressionKind::Zstd)
    .batching(/* max_messages */ 256, /* max_bytes */ 128 * 1024)
    .create()
    .await?;

producer
    .send(OutgoingMessage::with_payload(b"hello, pulsar".as_slice()).into())
    .await?;

let consumer = client
    .consumer("persistent://public/default/orders")
    .subscription("worker")
    .subscription_type(magnetar_proto::pb::command_subscribe::SubType::Shared)
    .subscribe()
    .await?;

let msg = consumer.receive().await?;
println!("payload: {:?}", msg.payload);
consumer.ack(msg.message_id).await?;
# Ok(()) }

use std::sync::Arc;
use magnetar::{PulsarClient, TypedProducerBuilder, TypedConsumerBuilder};
use magnetar_proto::schema::StringSchema;

# async fn run() -> Result<(), Box<dyn std::error::Error>> {
let client = PulsarClient::builder()
    .service_url("pulsar://localhost:6650")
    .build()
    .await?;

let schema = Arc::new(StringSchema::new());

let producer = client
    .typed_producer("persistent://public/default/notes", schema.clone())
    .create()
    .await?;

producer.new_message().value("a note".to_string()).send().await?;

let consumer = client
    .typed_consumer("persistent://public/default/notes", schema)
    .subscription("transcriber")
    .subscribe()
    .await?;

let msg = consumer.receive().await?;
println!("decoded value: {}", msg.value);
consumer.ack(msg.id).await?;
# Ok(()) }

use magnetar::PulsarClient;
use magnetar_proto::MessageId;

# async fn run() -> Result<(), Box<dyn std::error::Error>> {
let client = PulsarClient::builder()
    .service_url("pulsar://localhost:6650")
    .build()
    .await?;

let reader = client
    .reader("persistent://public/default/events")
    .start_message_id(MessageId::EARLIEST)
    .create()
    .await?;

while let Ok(msg) = reader.receive().await {
    println!("entry {:?}", msg.message_id);
}
# Ok(()) }

use magnetar::{PulsarClient, MessageRoutingMode};

# async fn run() -> Result<(), Box<dyn std::error::Error>> {
let client = PulsarClient::builder()
    .service_url("pulsar://localhost:6650")
    .build()
    .await?;

let p = client
    .partitioned_producer("persistent://public/default/events")
    .routing_mode(MessageRoutingMode::RoundRobin)
    .batching(/* max_messages */ 128, /* max_bytes */ 64 * 1024)
    .create()
    .await?;

p.new_message().key("user-42").value(b"event".as_slice()).send().await?;

let c = client
    .partitioned_consumer("persistent://public/default/events")
    .subscription("workers")
    .subscribe()
    .await?;

let msg = c.receive().await?;
c.ack(msg.topic(), msg.message_id).await?;
# Ok(()) }

use magnetar::PulsarClient;

# async fn run() -> Result<(), Box<dyn std::error::Error>> {
let client = PulsarClient::builder()
    .service_url("pulsar://localhost:6650")
    .build()
    .await?;

let pc = client
    .pattern_consumer()
    .namespace("public/default")
    .pattern("orders-.*")
    .subscription("workers")
    .subscribe(&client)
    .await?;

println!("matched topics: {:?}", pc.topics());

let msg = pc.receive().await?;
pc.ack(msg.topic(), msg.message_id).await?;
# Ok(()) }

use magnetar::PulsarClient;

# async fn run() -> Result<(), Box<dyn std::error::Error>> {
let client = PulsarClient::builder()
    .service_url("pulsar://localhost:6650")
    .build()
    .await?;

let view = client
    .table_view("persistent://public/default/config")
    .subscription("cfg-watcher")
    .create()
    .await?;

view.for_each(|key, value| println!("{key} = {value:?}"));
let last = view.get("api.threshold");
# Ok(()) }

use std::time::Duration;
use magnetar::{PulsarClient, OutgoingMessage};

# async fn run() -> Result<(), Box<dyn std::error::Error>> {
let client = PulsarClient::builder()
    .service_url("pulsar://localhost:6650")
    .build()
    .await?;

// open the transaction-coordinator-backed transaction
let runtime_client = /* obtain magnetar_runtime_tokio::Client */
#   unreachable!();
let txn = runtime_client.new_txn(Duration::from_secs(60)).await?;

let producer = client
    .producer("persistent://public/default/orders")
    .create()
    .await?;

producer
    .send(OutgoingMessage::with_payload(b"line-item".as_slice()).txn(txn.id()).into())
    .await?;

txn.commit().await?;
# Ok(()) }

use std::sync::Arc;
use magnetar::{
    ConsumerInterceptor, IncomingMessage, OutgoingMessage, ProducerInterceptor, PulsarClient,
    send_with_interceptors,
};

#[derive(Debug)]
struct StampSender;
impl ProducerInterceptor for StampSender {
    fn before_send(&self, msg: &mut OutgoingMessage) {
        msg.properties.push(("client".to_owned(), "magnetar".to_owned()));
    }
}

# async fn run() -> Result<(), Box<dyn std::error::Error>> {
let client = PulsarClient::builder()
    .service_url("pulsar://localhost:6650")
    .build()
    .await?;

let producer = client
    .producer("persistent://public/default/orders")
    .create()
    .await?;

let chain: Vec<Arc<dyn ProducerInterceptor>> = vec![Arc::new(StampSender)];
let id = send_with_interceptors(
    &producer,
    OutgoingMessage::with_payload(b"hi".as_slice()),
    &chain,
)
.await?;
println!("acked at {id:?}");
# Ok(()) }

A check (✅) is a working public-API surface backed by code in the workspace. A flag (🟡) means partial — a working subset; check ARCHITECTURE.md for the open gaps. A cross (❌) is a known-missing feature.

• Moonpool engine parity. Java parity is fully satisfied by the tokio engine (ADR-0019); the façade's dependent surfaces are engine-generic through ADR-0026 §D1 extension traits. Producers, consumers, readers, partitioned producer/consumer, multi-topics, pattern consumers, TableView, transactions, and typed schema builders all dispatch through impl<E: Engine> PulsarClient<E> where the selected runtime implements the matching *Api trait. Per-feature, per-engine status lives in the parity matrix above.
• PIP-460 scalable topics ship as an experimental scaffold behind feature = "scalable-topics" (default off); the in-process four-layer test suite is the binding gate. e2e against a real broker waits for an upstream Pulsar release that ships PIP-460 (currently Draft, targeting Pulsar 5.0 LTS).
• PIP-466 V5 surface ships as an experimental, engine-generic wrapper behind feature = "experimental-v5-client" (default off). No wire change — it wraps the v4 surface.
• SASL ships both mechanisms end-to-end: PLAIN (RFC 4616) under the default auth-sasl feature, and Kerberos/GSSAPI via libgssapi under the auth-sasl-kerberos feature. The multi-round AUTH_CHALLENGE exchange threads through AuthProvider::respond_to_challenge. The four sans-io test layers drive a deterministic ScriptedGssapiClient; the e2e layer runs against a Dockerised KDC. See ADR-0029.
• Athenz ships both the pre-fetched role-token path and the opt-in ZTS round-trip (auth-athenz-zts). Production-style ZMS+ZTS+certificate bootstrap remains out of scope for the local fixture.

Magnetar publishes two engines that drive the same sans-io state machine. Pick at compile time via feature flags. The parity matrix above is the per-feature, per-engine status snapshot; this section describes each engine's surface.

• TLS via tokio-rustls (ring backend); no native-tls, no openssl.
• One driver task per connection — see ARCHITECTURE.md §"The driver loop".
• The user-facing futures (Consumer::receive, Producer::send, …) lock the shared state machine, register their Waker in a slab, and wait. The driver picks them up as the matching OpOutcome lands.
• This is what magnetar::PulsarClient wires by default (PulsarClient<TokioEngine>).

• Drives the same sans-io state machine as the tokio engine over moonpool_core::Providers (a bundle of NetworkProvider, TimeProvider, TaskProvider, RandomProvider, StorageProvider). Plug TokioProviders for production-style runs against a real broker, or a moonpool-sim provider bundle for reproducible chaos under a seed.
• TLS uses a local rustls::ClientConnection adapter (tls.rs) that drives read_tls / process_new_packets / write_tls over the moonpool byte pipe — the handshake stays deterministic under chaos.
• See docs/moonpool-engine.md for the engine's surface, supervised reconnect, chaos test pack, and the tokio ↔ moonpool differential equivalence harness.

• Pulsar 4.0+ (LTS). The CONNECT frame advertises ProtocolVersion::V21 and the connection falls back to whichever lower version the broker reports on CONNECTED.
• The end-to-end suite runs against apachepulsar/pulsar:4.0.4.

Magnetar targets a feature-complete Apache Pulsar driver with full Java-client parity, driven by two interchangeable engines (production tokio + deterministic-simulation moonpool) over the same sans-io protocol state machine (ADR-0010, ADR-0019). Java parity is satisfied on the tokio engine; the moonpool engine is on the same trajectory and the differential harness enforces tokio ↔ moonpool EventStream equivalence.

The bulk of the parity matrix above ships on main, including:

• PIP-180 shadow topic (ADR-0033).
• PIP-33 replicated subscriptions (ADR-0034).
• SASL Kerberos / GSSAPI (ADR-0029).
• Pluggable rustls crypto provider (aws-lc-rs / ring / openssl / fips — ADR-0035).
• Daily 128-random-seed moonpool sweep (ADR-0036).
• Anti-thrash supervised reconnect policy (opt-in, ADR-0028).

Known open work is narrow and tracked in docs/follow-ups.md: PIP-460 e2e waits for a Pulsar 5.0 RC that pins the scalable-topic wire commands, the moonpool git dependency waits for a release containing upstream PR #113, and a few simulation / test-harness gaps remain.

API is unstable until the first tagged release — do not depend on this in production yet.

The whole workspace builds against stable Rust 1.88.

# Build / lint / format
cargo build --workspace --all-features
cargo clippy --workspace --all-features -- -D warnings
cargo +nightly fmt --check

# Unit + integration tests (no broker needed)
cargo test --workspace

# Dependency audits
cargo deny check

# Docs
RUSTDOCFLAGS="-D warnings --cfg tokio_unstable --cfg tracing_unstable" \
  cargo doc --workspace --all-features --no-deps

End-to-end tests against a real broker run as part of cargo test --workspace --all-features (ADR-0046 — no --features e2e, no #[ignore]). Docker is the only prerequisite; the suite spins pulsar:4.0.4 via testcontainers-rs.

Additional xtask checks specific to the sans-io invariants:

cargo run -p xtask -- check-no-channels   # greps src/** for banned channel crates
cargo run -p xtask -- check-no-io-deps    # magnetar-proto must not depend on any I/O crate
cargo run -p xtask -- codegen --check     # asserts proto codegen has no drift

Apache-2.0 — see LICENSE and NOTICE. The project vendors a verbatim copy of the Apache Pulsar wire protocol definition (PulsarApi.proto, PulsarMarkers.proto), released by the Apache Software Foundation under Apache-2.0.

See GUIDELINES.md and CONTRIBUTING.md for project conventions before sending a patch.