---
description: Conventions for developing the RustStream core crate (src/, crates/ruststream-macros)
globs: src/**/*.rs,crates/**/*.rs,tests/**/*.rs
alwaysApply: false
---
- Core is broker-agnostic: traits and message types with zero broker dependencies. Keep
broker-specific config and defaults out of core; if a field has no sane default, do not implement
`Default`. Core is fully generic (no object-safety); erasure lives in `runtime`.
- Default visibility `pub(crate)`; make `pub` only what `lib.rs` re-exports. The crate root denies
`missing_docs`: every `pub` item gets a summary, and every one outside `#[doc(hidden)]` a
compiling `# Examples` doctest (a hidden item is macro or SPI machinery no example names); a
fallible call in it is written with `?`, never unwrapped.
- Library code uses `thiserror` (`anyhow` is bins/examples only); public error enums are
`#[non_exhaustive]`. Never panic on user/network input - return `Result`.
- Native `async fn in trait`; `+ Send` at the call site, not the trait. Destructors never panic or
block. Features are strictly additive, never mutually exclusive: `json` (default), `msgpack`,
`cbor`, `memory`, `macros`, `asyncapi`, `metrics`, `logging`, `otel`, `conformance`, `cli`,
`testing`.
- Settlement is honest: every site that settles a delivery accepts `AckError::Unsupported`. A
transport that cannot acknowledge is a supported broker, not a failure, and the conformance
suite passes on it - the redelivery scenario ends where the requeue reports itself unsupported.
- Compile-time rejection is the default. The ladder for catching a MISUSE (not data - decode errors
and broker faults are runtime by nature) is compile error > build-time assert > startup error >
hot-path check, and the last rung is not an option. Prefer an enum over sibling `bool`/`Option`
fields with illegal combinations, a newtype (or `NonZero*`) over adjacent same-typed parameters,
and a typestate - a consuming `self` transition, a phase type parameter, a scope guard - over a
"call X before Y" contract held only by docs. A runtime check survives only where the force is
external (the language, the domain); mark those with a "why" comment. Put
`#[diagnostic::on_unimplemented]` guidance on bounds users will hit.
- Gates: `just check` (`cargo fmt --check`, `clippy --all-targets --all-features -- -D warnings`,
`cargo check` all-features, `--no-default-features` and the codec-free leg
`--no-default-features --features testing,memory,macros`, then `cargo doc` under
`RUSTDOCFLAGS="-D warnings"`), `just test` (all-features, the reduced-feature legs, doctests on
both feature edges), `just ci` for both. SemVer pre-1.0: minor = breaking, patch = additive;
`ruststream` and `ruststream-macros` versions are lockstep via `[workspace.package].version`.
- English only, no emoji, no em-dashes. One logical unit per commit; no `Co-Authored-By`. Do not
publish/tag/release - prepare branches/PRs; the maintainer releases.
## Error type
One `#[non_exhaustive]` enum per crate, variants by source, with `thiserror`.
```rust
use std::error::Error as StdError;
use thiserror::Error;
#[derive(Debug, Error)]
#[non_exhaustive]
pub enum PublishError {
/// The broker rejected the publish.
#[error("publish rejected: {0}")]
Rejected(#[source] Box<dyn StdError + Send + Sync>),
}
```
## Trait with async methods
Use RPITIT; keep `+ Send` on the returned future. Do not force object-safety in core. A broker's
per-message settings are an associated type of the publisher, so a setting never rides a header:
`Options` is `()` for a broker that has none, and `None` on the paths with no call site to adjust
them (a reply, a deferred retry).
```rust
use std::future::Future;
pub trait Publisher: Send + Sync {
type Error: std::error::Error + Send + Sync + 'static;
/// The broker's per-message settings; every field optional.
type Options: Clone + Send + Sync + 'static;
/// How this transport consumes the payload: `Lend` reads it, `Take` keeps it.
type Payload: PayloadForm;
/// Publishes `msg` with the settings the call site adjusted.
///
/// # Errors
///
/// Returns [`Self::Error`] when the broker rejects the message.
fn publish(
&self,
msg: OutgoingFor<'_, Self::Payload>,
options: Option<&Self::Options>,
) -> impl Future<Output = Result<(), Self::Error>> + Send;
/// Headers this handle contributes under every message it sends.
fn base_headers(&self) -> Option<&HeaderMap> {
None
}
}
```
## Public item with a doctest
Every `pub` item outside `#[doc(hidden)]` carries a compiling example. A fallible call in it is
written with `?`, never `unwrap()`; an example with nothing fallible in it, like the one below,
has nothing to propagate.
```rust
/// Creates a name source bound to `name`.
///
/// # Examples
///
/// ```
/// use ruststream::Name;
/// let source = Name::new("orders");
/// # let _ = source;
/// ```
#[must_use]
pub fn new(name: impl Into<Cow<'static, str>>) -> Self {
Self(name.into())
}
```
## Feature-gated default codec
`DefaultCodec` is a feature-selected alias, resolved by priority at compile time. It exists only
when at least one codec feature is on, so every API that falls back to it is gated the same way
(`#[cfg(any(feature = "json", feature = "cbor", feature = "msgpack"))]`); with no codec feature,
every call site names a codec.
```rust
/// The codec used when an `include` / publisher call does not name one explicitly.
#[cfg(feature = "json")]
pub type DefaultCodec = JsonCodec;
#[cfg(all(not(feature = "json"), feature = "cbor"))]
pub type DefaultCodec = CborCodec;
#[cfg(all(not(feature = "json"), not(feature = "cbor"), feature = "msgpack"))]
pub type DefaultCodec = MsgpackCodec;
```
## Conformance self-test
The reference `MemoryBroker` runs the harness in `tests/conformance_self.rs`. The routing suite
reads the publish log back, so it takes the retaining form; the lifecycle ladder takes the plain
one. The closures stay closures: their bounds are higher-ranked, so a bare method path would bind
one concrete lifetime and fail to type-check.
```rust
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn memory_broker_passes_conformance_suite() {
harness::run_suite(|| MemoryBroker::retaining(Retention::Messages(nonzero!(64)))).await;
}
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn memory_broker_passes_lifecycle() {
harness::lifecycle(
MemoryBroker::new,
|name| MemorySource::new(name),
|broker| broker.publisher(),
)
.await;
}
```