use std::net::SocketAddr;
use std::path::PathBuf;
use std::time::Duration;
use crate::ServerError;
#[derive(Debug, Clone, serde::Deserialize)]
#[serde(deny_unknown_fields)]
pub struct ServerConfig {
pub listen_address: SocketAddr,
pub health_listen_address: SocketAddr,
pub drain_timeout_ms: u64,
pub channels: Vec<ChannelDef>,
pub routing_rules: Vec<RoutingRuleDef>,
pub persistence_path: Option<PathBuf>,
pub cluster: Option<ClusterConfig>,
#[serde(default)]
pub auth: Option<AuthConfig>,
#[serde(default)]
pub services: ServicesConfig,
#[serde(default)]
pub limits: LimitsConfig,
#[serde(default)]
pub websocket: Option<WebSocketConfig>,
#[serde(default)]
pub participant: Option<ParticipantConfig>,
}
impl ServerConfig {
#[must_use]
pub const fn drain_timeout(&self) -> Duration {
Duration::from_millis(self.drain_timeout_ms)
}
}
#[derive(Debug, Clone, serde::Deserialize)]
#[serde(deny_unknown_fields)]
pub struct ChannelDef {
pub name: String,
#[serde(default)]
pub schema_ref: Option<PathBuf>,
pub durable: bool,
#[serde(skip)]
pub loaded_schema: Option<LoadedSchema>,
}
#[derive(Debug, Clone)]
pub struct LoadedSchema {
pub bytes: Vec<u8>,
pub document: serde_json::Value,
}
#[derive(Debug, Clone, serde::Deserialize)]
#[serde(deny_unknown_fields)]
pub struct RoutingRuleDef {
pub source_channel: String,
pub target_channel: String,
pub predicate: Option<String>,
}
pub const DEFAULT_COOKIE: &str = "beamr-cookie";
#[derive(Debug, Clone, serde::Deserialize)]
#[serde(deny_unknown_fields)]
pub struct ClusterConfig {
pub node_name: String,
pub listen_address: SocketAddr,
pub seed_nodes: Vec<SocketAddr>,
#[serde(default = "default_cookie")]
pub cookie: String,
}
fn default_cookie() -> String {
DEFAULT_COOKIE.to_owned()
}
#[derive(Debug, Clone, serde::Deserialize)]
#[serde(deny_unknown_fields)]
pub struct AuthConfig {
pub token: String,
}
#[derive(Debug, Clone, serde::Deserialize)]
#[serde(deny_unknown_fields)]
pub struct WebSocketConfig {
pub listen_address: SocketAddr,
pub path: String,
#[serde(default)]
pub allowed_origins: Vec<String>,
#[serde(default)]
pub ping_interval_ms: Option<u64>,
}
#[derive(Debug, Clone, serde::Deserialize)]
#[serde(deny_unknown_fields)]
pub struct ServicesConfig {
#[serde(default = "default_service_profile")]
pub profile: String,
}
impl Default for ServicesConfig {
fn default() -> Self {
Self {
profile: default_service_profile(),
}
}
}
impl ServicesConfig {
pub fn profile(&self) -> Result<ServiceProfile, ServerError> {
ServiceProfile::parse(&self.profile)
}
}
fn default_service_profile() -> String {
ServiceProfile::FULL.to_owned()
}
#[derive(Debug, Clone, Copy, serde::Deserialize)]
#[serde(deny_unknown_fields)]
pub struct LimitsConfig {
#[serde(default = "default_max_connections")]
pub max_connections: usize,
#[serde(default = "default_max_subscriptions_per_connection")]
pub max_subscriptions_per_connection: usize,
#[serde(default = "default_max_conversations_per_connection")]
pub max_conversations_per_connection: usize,
#[serde(default = "default_max_pending_pushes_per_connection")]
pub max_pending_pushes_per_connection: usize,
#[serde(default = "default_max_pending_conversation_replies_per_connection")]
pub max_pending_conversation_replies_per_connection: usize,
#[serde(default = "default_max_pending_replies_per_conversation")]
pub max_pending_replies_per_conversation: usize,
#[serde(default = "default_max_connection_inbox_bytes")]
pub max_connection_inbox_bytes: usize,
#[serde(default = "default_max_subscription_inbox_depth")]
pub max_subscription_inbox_depth: usize,
}
impl LimitsConfig {
pub const DEFAULT_MAX_CONNECTIONS: usize = 256;
pub const DEFAULT_MAX_SUBSCRIPTIONS_PER_CONNECTION: usize = 32;
pub const DEFAULT_MAX_CONVERSATIONS_PER_CONNECTION: usize = 32;
pub const DEFAULT_MAX_PENDING_PUSHES_PER_CONNECTION: usize = 32;
pub const DEFAULT_MAX_PENDING_CONVERSATION_REPLIES_PER_CONNECTION: usize = 32;
pub const DEFAULT_MAX_PENDING_REPLIES_PER_CONVERSATION: usize = 8;
pub const DEFAULT_MAX_CONNECTION_INBOX_BYTES: usize = 4 * 1024 * 1024;
pub const DEFAULT_MAX_SUBSCRIPTION_INBOX_DEPTH: usize = 256;
pub(crate) fn collect_errors(&self, errors: &mut Vec<String>) {
let checks: [(&str, usize); 8] = [
("max_connections", self.max_connections),
(
"max_subscriptions_per_connection",
self.max_subscriptions_per_connection,
),
(
"max_conversations_per_connection",
self.max_conversations_per_connection,
),
(
"max_pending_pushes_per_connection",
self.max_pending_pushes_per_connection,
),
(
"max_pending_conversation_replies_per_connection",
self.max_pending_conversation_replies_per_connection,
),
(
"max_pending_replies_per_conversation",
self.max_pending_replies_per_conversation,
),
(
"max_connection_inbox_bytes",
self.max_connection_inbox_bytes,
),
(
"max_subscription_inbox_depth",
self.max_subscription_inbox_depth,
),
];
for (field, value) in checks {
if value == 0 {
errors.push(format!(
"limits.{field}: must be greater than zero (a zero cap would be \
unlimited-by-silence, which §5 forbids)"
));
}
}
}
}
impl Default for LimitsConfig {
fn default() -> Self {
Self {
max_connections: default_max_connections(),
max_subscriptions_per_connection: default_max_subscriptions_per_connection(),
max_conversations_per_connection: default_max_conversations_per_connection(),
max_pending_pushes_per_connection: default_max_pending_pushes_per_connection(),
max_pending_conversation_replies_per_connection:
default_max_pending_conversation_replies_per_connection(),
max_pending_replies_per_conversation: default_max_pending_replies_per_conversation(),
max_connection_inbox_bytes: default_max_connection_inbox_bytes(),
max_subscription_inbox_depth: default_max_subscription_inbox_depth(),
}
}
}
const fn default_max_connections() -> usize {
LimitsConfig::DEFAULT_MAX_CONNECTIONS
}
const fn default_max_subscriptions_per_connection() -> usize {
LimitsConfig::DEFAULT_MAX_SUBSCRIPTIONS_PER_CONNECTION
}
const fn default_max_conversations_per_connection() -> usize {
LimitsConfig::DEFAULT_MAX_CONVERSATIONS_PER_CONNECTION
}
const fn default_max_pending_pushes_per_connection() -> usize {
LimitsConfig::DEFAULT_MAX_PENDING_PUSHES_PER_CONNECTION
}
const fn default_max_pending_conversation_replies_per_connection() -> usize {
LimitsConfig::DEFAULT_MAX_PENDING_CONVERSATION_REPLIES_PER_CONNECTION
}
const fn default_max_pending_replies_per_conversation() -> usize {
LimitsConfig::DEFAULT_MAX_PENDING_REPLIES_PER_CONVERSATION
}
const fn default_max_connection_inbox_bytes() -> usize {
LimitsConfig::DEFAULT_MAX_CONNECTION_INBOX_BYTES
}
const fn default_max_subscription_inbox_depth() -> usize {
LimitsConfig::DEFAULT_MAX_SUBSCRIPTION_INBOX_DEPTH
}
#[derive(Debug, Clone, Copy, serde::Deserialize)]
#[serde(deny_unknown_fields)]
pub struct ParticipantConfig {
pub wire_frame_limit: u64,
pub attach_receipt_ttl_ms: u64,
pub receipt_provenance_ttl_ms: u64,
pub max_live_attach_receipts_server: u64,
pub max_live_attach_receipts_per_participant: u64,
pub max_receipt_provenance_server: u64,
pub max_receipt_provenance_per_conversation: u64,
pub max_receipt_provenance_per_participant: u64,
pub max_retired_identity_slots_server: u64,
pub identity_slots: u64,
pub observer_recovery_max_entries: u64,
pub max_semantic_conversations_per_connection: u64,
pub max_ordinary_record_entries: u64,
pub max_ordinary_record_bytes: u64,
pub max_generated_marker_entries: u64,
pub max_generated_marker_bytes: u64,
pub mandatory_transaction_bound_entries: u64,
pub mandatory_transaction_bound_bytes: u64,
pub full_recovery_claim_entries: u64,
pub full_recovery_claim_bytes: u64,
pub retained_capacity_entries: u64,
pub retained_capacity_bytes: u64,
pub max_retained_record_rows: u64,
pub closure_episode_churn_limit: u64,
}
impl ParticipantConfig {
pub(crate) fn collect_errors(&self, errors: &mut Vec<String>) {
let nonzero: [(&str, u64); 23] = [
("wire_frame_limit", self.wire_frame_limit),
("attach_receipt_ttl_ms", self.attach_receipt_ttl_ms),
("receipt_provenance_ttl_ms", self.receipt_provenance_ttl_ms),
(
"max_live_attach_receipts_server",
self.max_live_attach_receipts_server,
),
(
"max_live_attach_receipts_per_participant",
self.max_live_attach_receipts_per_participant,
),
(
"max_receipt_provenance_server",
self.max_receipt_provenance_server,
),
(
"max_receipt_provenance_per_conversation",
self.max_receipt_provenance_per_conversation,
),
(
"max_receipt_provenance_per_participant",
self.max_receipt_provenance_per_participant,
),
(
"max_retired_identity_slots_server",
self.max_retired_identity_slots_server,
),
("identity_slots", self.identity_slots),
(
"observer_recovery_max_entries",
self.observer_recovery_max_entries,
),
(
"max_semantic_conversations_per_connection",
self.max_semantic_conversations_per_connection,
),
(
"max_ordinary_record_entries",
self.max_ordinary_record_entries,
),
("max_ordinary_record_bytes", self.max_ordinary_record_bytes),
(
"max_generated_marker_entries",
self.max_generated_marker_entries,
),
(
"max_generated_marker_bytes",
self.max_generated_marker_bytes,
),
(
"mandatory_transaction_bound_entries",
self.mandatory_transaction_bound_entries,
),
(
"mandatory_transaction_bound_bytes",
self.mandatory_transaction_bound_bytes,
),
(
"full_recovery_claim_entries",
self.full_recovery_claim_entries,
),
("full_recovery_claim_bytes", self.full_recovery_claim_bytes),
("retained_capacity_entries", self.retained_capacity_entries),
("retained_capacity_bytes", self.retained_capacity_bytes),
("max_retained_record_rows", self.max_retained_record_rows),
];
for (field, value) in nonzero {
if value == 0 {
errors.push(format!("participant.{field}: must be greater than zero"));
}
}
if self.receipt_provenance_ttl_ms < self.attach_receipt_ttl_ms {
errors.push(
"participant.receipt_provenance_ttl_ms: must be at least \
attach_receipt_ttl_ms (provenance cannot expire before the receipt it explains)"
.to_owned(),
);
}
if self.full_recovery_claim_entries != self.mandatory_transaction_bound_entries {
errors.push(
"participant.full_recovery_claim_entries: must equal \
mandatory_transaction_bound_entries"
.to_owned(),
);
}
if self.full_recovery_claim_bytes != self.mandatory_transaction_bound_bytes {
errors.push(
"participant.full_recovery_claim_bytes: must equal \
mandatory_transaction_bound_bytes"
.to_owned(),
);
}
if !(2..=u64::from(u32::MAX)).contains(&self.closure_episode_churn_limit) {
errors.push(
"participant.closure_episode_churn_limit: must be in 2..=u32::MAX".to_owned(),
);
}
self.collect_unit2_derived_errors(errors);
}
fn collect_unit2_derived_errors(&self, errors: &mut Vec<String>) {
if self
.max_retained_record_rows
.checked_mul(self.identity_slots)
.is_none()
{
errors.push("participant.UNIT2_MAX_LIVE_RECIPIENT_OBLIGATIONS: max_retained_record_rows * identity_slots overflows u64".to_owned());
}
}
}
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum ServiceProfile {
Full,
WorkerFrontDoor,
}
impl ServiceProfile {
pub const FULL: &'static str = "full";
pub const WORKER_FRONT_DOOR: &'static str = "worker-front-door";
pub fn parse(value: &str) -> Result<Self, ServerError> {
match value {
Self::FULL => Ok(Self::Full),
Self::WORKER_FRONT_DOOR => Ok(Self::WorkerFrontDoor),
other => Err(ServerError::ConfigValidation {
message: format!(
"services.profile: unknown profile '{other}'; expected \"{}\" or \"{}\"",
Self::FULL,
Self::WORKER_FRONT_DOOR
),
}),
}
}
}