#![cfg_attr(not(feature = "std"), no_std)]
#[cfg(feature = "alloc")]
extern crate alloc;
#[cfg(feature = "alloc")]
pub mod accept_rule;
pub mod arch;
#[cfg(feature = "alloc")]
pub mod arena;
pub mod kv_decision;
pub mod ring;
#[cfg(feature = "alloc")]
pub mod serving_state;
#[cfg(feature = "alloc")]
pub mod batch;
#[cfg(feature = "std")]
pub mod buffer;
pub mod datagram_batch;
pub mod factory;
#[cfg(target_has_atomic = "64")]
pub mod histogram;
#[cfg(feature = "io-async")]
pub mod io;
pub mod markers;
#[cfg(feature = "park")]
pub mod park;
pub mod per_core;
#[cfg(feature = "std")]
pub mod signal;
pub mod time;
#[cfg(feature = "live")]
pub mod live;
pub mod sized {
include!(concat!(env!("OUT_DIR"), "/proxima_core_sized.rs"));
}
#[cfg(feature = "registry")]
pub use factory::FactoryRegistry;
pub use factory::Named;
#[cfg(feature = "config")]
pub use factory::{Composition, Factory, FactorySpec};
#[cfg(feature = "alloc")]
pub use serving_state::{ServingFsmError, ServingState};
#[cfg(feature = "alloc")]
use alloc::boxed::Box;
#[cfg(feature = "alloc")]
use alloc::string::String;
use core::time::Duration;
use thiserror::Error;
#[cfg(feature = "alloc")]
pub type ErrorSource = Box<dyn core::error::Error + Send + Sync + 'static>;
#[derive(Debug, Error)]
#[cfg(feature = "alloc")]
pub enum DecodeError {
#[error("missing header terminator")]
MissingHeaderTerminator,
#[error("empty status line")]
EmptyStatusLine,
#[error("malformed status line: {raw}")]
MalformedStatusLine { raw: String },
#[error("chunked: size CRLF missing")]
ChunkedSizeMissing,
#[error("chunked: payload short")]
ChunkedPayloadShort,
#[error("not utf-8 ({context}): {source}")]
NotUtf8 {
context: &'static str,
source: ErrorSource,
},
#[error("parse {context}: {source}")]
Parse {
context: &'static str,
source: ErrorSource,
},
#[error("missing required field: {field}")]
MissingField { field: &'static str },
#[error("decode ({context}): {source}")]
Other {
context: &'static str,
source: ErrorSource,
},
}
#[derive(Debug, Error)]
#[cfg(feature = "alloc")]
pub enum EncodeError {
#[error("json encode: {source}")]
Json { source: ErrorSource },
#[error("format convert ({context}): {source}")]
FormatConvert {
context: &'static str,
source: ErrorSource,
},
#[error("render ({context}): {source}")]
Render {
context: &'static str,
source: ErrorSource,
},
#[error("base64: {source}")]
Base64 { source: ErrorSource },
#[error("requires base64 string input")]
RequiresBase64Input,
#[error("encode ({context}): {source}")]
Other {
context: &'static str,
source: ErrorSource,
},
}
#[derive(Debug, Error)]
#[cfg(feature = "alloc")]
pub enum UpstreamError {
#[error("io ({operation}): {source}")]
Io {
operation: &'static str,
source: ErrorSource,
},
#[error("upstream returned status {status}")]
BadStatus { status: u16 },
#[error("upstream cancelled")]
Cancelled,
#[error("protocol ({context}): {source}")]
Protocol {
context: &'static str,
source: ErrorSource,
},
#[error("connect ({context}): {source}")]
Connect {
context: &'static str,
source: ErrorSource,
},
#[error("upstream ({context}): {source}")]
Other {
context: &'static str,
source: ErrorSource,
},
}
#[derive(Debug, Error)]
#[cfg(feature = "alloc")]
pub enum ConfigError {
#[error("{component}: requires field '{field}'")]
RequiredField {
component: &'static str,
field: &'static str,
},
#[error("{component}: requires an inner pipe")]
MissingInnerPipe { component: &'static str },
#[error("no upstreams configured")]
NoUpstreams,
#[error("parse ({context}): {source}")]
Parse {
context: &'static str,
source: ErrorSource,
},
#[error("spawn failed ({context}): {source}")]
SpawnFailed {
context: &'static str,
source: ErrorSource,
},
#[error("config ({context}): {source}")]
Other {
context: &'static str,
source: ErrorSource,
},
}
#[derive(Debug, Error)]
#[cfg(feature = "alloc")]
pub enum BodyError {
#[error("body cancelled")]
Cancelled,
#[error("worker dropped sender ({context})")]
WorkerDropped { context: &'static str },
#[error("worker panicked ({context})")]
WorkerPanic { context: &'static str },
#[error("serialize ({context}): {source}")]
Serialize {
context: &'static str,
source: ErrorSource,
},
#[error("body ({context}): {source}")]
Other {
context: &'static str,
source: ErrorSource,
},
}
#[derive(Debug, Error)]
#[cfg(feature = "alloc")]
pub enum NotFoundError {
#[error("no pipe registered as '{name}'")]
Pipe { name: String },
#[error("{kind} '{name}' not found")]
Resource { kind: &'static str, name: String },
}
#[derive(Debug, Error)]
#[cfg(feature = "alloc")]
pub enum RegistryError {
#[error("no {kind} named '{name}'")]
NotRegistered { kind: &'static str, name: String },
#[error("{kind} '{name}' already registered")]
AlreadyRegistered { kind: &'static str, name: String },
}
#[derive(Debug, Error)]
#[cfg(feature = "alloc")]
pub enum RecordError {
#[error("recording io ({operation}): {source}")]
Io {
operation: &'static str,
source: ErrorSource,
},
#[error("recording codec ({operation}): {source}")]
Codec {
operation: &'static str,
source: ErrorSource,
},
#[error("recording frame exceeds u32::MAX bytes")]
FrameTooLarge,
#[error("recording version mismatch: expected {expected}, got {got}")]
VersionMismatch { expected: u8, got: u8 },
#[error("recording buffer full at capacity {capacity}")]
BufferFull { capacity: usize },
}
#[derive(Debug, Error)]
pub enum ProximaError {
#[error("upstream error: {0}")]
#[cfg(feature = "alloc")]
Upstream(String),
#[error("timeout after {0:?}")]
Timeout(Duration),
#[error("rate limited")]
RateLimited,
#[error("no upstream had data")]
NoData,
#[error("config: {0}")]
#[cfg(feature = "alloc")]
Config(String),
#[error("io: {0}")]
#[cfg(feature = "std")]
Io(#[from] std::io::Error),
#[error("decode: {0}")]
#[cfg(feature = "alloc")]
Decode(String),
#[error("encode: {0}")]
#[cfg(feature = "alloc")]
Encode(String),
#[error("body stream: {0}")]
#[cfg(feature = "alloc")]
Body(String),
#[error("not found: {0}")]
#[cfg(feature = "alloc")]
NotFound(String),
#[error("forbidden: {0}")]
#[cfg(feature = "alloc")]
Forbidden(String),
#[error("registry: {0}")]
#[cfg(feature = "alloc")]
Registry(String),
#[error("retries exhausted after {attempts} attempts: {last}")]
#[cfg(feature = "alloc")]
RetriesExhausted { attempts: usize, last: String },
#[error("recording: {0}")]
#[cfg(feature = "alloc")]
Record(String),
#[error("replay miss: {fingerprint}")]
#[cfg(feature = "alloc")]
ReplayMiss { fingerprint: String },
#[error("decode: {0}")]
#[cfg(feature = "alloc")]
DecodeKind(#[from] DecodeError),
#[error("encode: {0}")]
#[cfg(feature = "alloc")]
EncodeKind(#[from] EncodeError),
#[error("upstream: {0}")]
#[cfg(feature = "alloc")]
UpstreamKind(#[from] UpstreamError),
#[error("config: {0}")]
#[cfg(feature = "alloc")]
ConfigKind(#[from] ConfigError),
#[error("body: {0}")]
#[cfg(feature = "alloc")]
BodyKind(#[from] BodyError),
#[error("not found: {0}")]
#[cfg(feature = "alloc")]
NotFoundKind(#[from] NotFoundError),
#[error("registry: {0}")]
#[cfg(feature = "alloc")]
RegistryKind(#[from] RegistryError),
#[error("recording: {0}")]
#[cfg(feature = "alloc")]
RecordKind(#[from] RecordError),
}
impl ProximaError {
#[must_use]
pub fn is_retryable(&self) -> bool {
match self {
Self::Timeout(_) => true,
Self::RateLimited => false,
#[cfg(feature = "alloc")]
Self::Upstream(_) | Self::Body(_) => true,
#[cfg(feature = "std")]
Self::Io(_) => true,
#[cfg(feature = "alloc")]
Self::UpstreamKind(kind) => !matches!(kind, UpstreamError::BadStatus { .. }),
#[cfg(feature = "alloc")]
Self::BodyKind(kind) => matches!(
kind,
BodyError::WorkerDropped { .. } | BodyError::WorkerPanic { .. }
),
_ => false,
}
}
}
pub type ProximaResult<T> = Result<T, ProximaError>;
#[cfg(test)]
#[cfg(feature = "alloc")]
#[allow(clippy::unwrap_used, clippy::expect_used)]
mod tests {
use super::*;
use alloc::format;
#[test]
fn timeout_is_retryable() {
assert!(ProximaError::Timeout(Duration::from_secs(1)).is_retryable());
}
#[test]
fn rate_limited_is_not_retryable() {
assert!(!ProximaError::RateLimited.is_retryable());
}
#[test]
fn legacy_upstream_is_retryable() {
assert!(ProximaError::Upstream("transient".into()).is_retryable());
}
#[test]
fn typed_upstream_io_is_retryable_but_bad_status_is_not() {
let io = UpstreamError::Io {
operation: "read",
source: Box::new(LightweightTestError("transient")),
};
assert!(ProximaError::from(io).is_retryable());
let bad = UpstreamError::BadStatus { status: 500 };
assert!(!ProximaError::from(bad).is_retryable());
}
#[test]
fn decode_kind_carries_typed_payload() {
let err = DecodeError::MissingField {
field: "pipeline_id",
};
let display = format!("{}", err);
assert!(display.contains("pipeline_id"));
}
#[test]
fn from_decode_error_to_proxima_error_via_from_impl() {
let err: ProximaError = DecodeError::EmptyStatusLine.into();
assert!(matches!(
err,
ProximaError::DecodeKind(DecodeError::EmptyStatusLine)
));
}
#[test]
fn config_required_field_renders_component_and_field() {
let err = ConfigError::RequiredField {
component: "auth",
field: "allow",
};
let display = format!("{}", err);
assert!(display.contains("auth"));
assert!(display.contains("allow"));
}
#[test]
fn registry_not_registered_uses_kind_and_name() {
let err = RegistryError::NotRegistered {
kind: "codec factory",
name: "json".into(),
};
let display = format!("{}", err);
assert!(display.contains("codec factory"));
assert!(display.contains("json"));
}
#[test]
fn record_version_mismatch_carries_numbers() {
let err = RecordError::VersionMismatch {
expected: 3,
got: 2,
};
let display = format!("{}", err);
assert!(display.contains('3'));
assert!(display.contains('2'));
}
#[derive(Debug)]
struct LightweightTestError(&'static str);
impl core::fmt::Display for LightweightTestError {
fn fmt(&self, f: &mut core::fmt::Formatter<'_>) -> core::fmt::Result {
f.write_str(self.0)
}
}
impl core::error::Error for LightweightTestError {}
}