mirror of
https://github.com/block/buzz.git
synced 2026-10-31 14:31:02 +00:00
refactor(relay): establish single-node backend seams
Signed-off-by: npub13n66s06epmqf2kc3v373ez8hj65cuzyvxzjf93vwpervxqn2u7jq2qd9je <8cf5a83f590ec0955b11647d1c88f796a98e088c30a492c58e0e46c3026ae7a4@buzz.block.builderlab.xyz>
This commit is contained in:
committed by
Brother Darryl
parent
ef847adde4
commit
4f7db3e428
@@ -17,6 +17,14 @@ pub type ByteStream = Pin<Box<dyn futures_core::Stream<Item = Result<Bytes, Medi
|
||||
|
||||
/// S3-compatible object storage client.
|
||||
pub struct MediaStorage {
|
||||
backend: MediaBackend,
|
||||
}
|
||||
|
||||
enum MediaBackend {
|
||||
S3(S3MediaStorage),
|
||||
}
|
||||
|
||||
struct S3MediaStorage {
|
||||
bucket: Box<Bucket>,
|
||||
}
|
||||
|
||||
@@ -66,7 +74,15 @@ impl MediaStorage {
|
||||
S3AddressingStyle::Path => bucket.with_path_style(),
|
||||
S3AddressingStyle::Virtual => bucket,
|
||||
};
|
||||
Ok(Self { bucket })
|
||||
Ok(Self {
|
||||
backend: MediaBackend::S3(S3MediaStorage { bucket }),
|
||||
})
|
||||
}
|
||||
|
||||
fn s3(&self) -> &S3MediaStorage {
|
||||
match &self.backend {
|
||||
MediaBackend::S3(storage) => storage,
|
||||
}
|
||||
}
|
||||
|
||||
/// Store an object from a byte slice.
|
||||
@@ -74,7 +90,8 @@ impl MediaStorage {
|
||||
/// Used for images, sidecars, and thumbnails. For large video files use
|
||||
/// [`put_file`] to avoid loading the entire blob into RAM.
|
||||
pub async fn put(&self, key: &str, bytes: &[u8], content_type: &str) -> Result<(), MediaError> {
|
||||
self.bucket
|
||||
self.s3()
|
||||
.bucket
|
||||
.put_object_with_content_type(key, bytes, content_type)
|
||||
.await?;
|
||||
Ok(())
|
||||
@@ -98,7 +115,8 @@ impl MediaStorage {
|
||||
.map_err(|e| MediaError::Io(e.to_string()))?;
|
||||
let mut reader = tokio::io::BufReader::with_capacity(BUF, file);
|
||||
|
||||
self.bucket
|
||||
self.s3()
|
||||
.bucket
|
||||
.put_object_stream_with_content_type(&mut reader, key, content_type)
|
||||
.await?;
|
||||
Ok(())
|
||||
@@ -106,7 +124,7 @@ impl MediaStorage {
|
||||
|
||||
/// Retrieve an object's bytes.
|
||||
pub async fn get(&self, key: &str) -> Result<Vec<u8>, MediaError> {
|
||||
match self.bucket.get_object(key).await {
|
||||
match self.s3().bucket.get_object(key).await {
|
||||
Ok(response) => Ok(response.to_vec()),
|
||||
Err(s3::error::S3Error::HttpFailWithBody(404, _)) => Err(MediaError::NotFound),
|
||||
Err(e) => Err(MediaError::StorageError(e.to_string())),
|
||||
@@ -119,7 +137,12 @@ impl MediaStorage {
|
||||
/// is transferred from S3 — the full object is never loaded into RAM.
|
||||
/// Intended for HTTP 206 range responses on large video blobs.
|
||||
pub async fn get_range(&self, key: &str, start: u64, end: u64) -> Result<Vec<u8>, MediaError> {
|
||||
match self.bucket.get_object_range(key, start, Some(end)).await {
|
||||
match self
|
||||
.s3()
|
||||
.bucket
|
||||
.get_object_range(key, start, Some(end))
|
||||
.await
|
||||
{
|
||||
Ok(response) => Ok(response.to_vec()),
|
||||
Err(s3::error::S3Error::HttpFailWithBody(404, _)) => Err(MediaError::NotFound),
|
||||
Err(e) => Err(MediaError::StorageError(e.to_string())),
|
||||
@@ -133,6 +156,7 @@ impl MediaStorage {
|
||||
/// blobs (video) directly into HTTP responses via `Body::from_stream()`.
|
||||
pub async fn get_stream(&self, key: &str) -> Result<ByteStream, MediaError> {
|
||||
let response = self
|
||||
.s3()
|
||||
.bucket
|
||||
.get_object_stream(key)
|
||||
.await
|
||||
@@ -150,7 +174,7 @@ impl MediaStorage {
|
||||
|
||||
/// Check if an object exists. Returns false on 404.
|
||||
pub async fn head(&self, key: &str) -> Result<bool, MediaError> {
|
||||
match self.bucket.head_object(key).await {
|
||||
match self.s3().bucket.head_object(key).await {
|
||||
Ok(_) => Ok(true),
|
||||
Err(s3::error::S3Error::HttpFailWithBody(404, _)) => Ok(false),
|
||||
Err(e) => Err(MediaError::StorageError(e.to_string())),
|
||||
@@ -159,7 +183,8 @@ impl MediaStorage {
|
||||
|
||||
/// Delete an object. Returns an error on failure — callers decide whether to propagate.
|
||||
pub async fn delete(&self, key: &str) -> Result<(), MediaError> {
|
||||
self.bucket
|
||||
self.s3()
|
||||
.bucket
|
||||
.delete_object(key)
|
||||
.await
|
||||
.map_err(|e| MediaError::StorageError(e.to_string()))?;
|
||||
@@ -168,7 +193,7 @@ impl MediaStorage {
|
||||
|
||||
/// HEAD with metadata — returns Content-Length (size).
|
||||
pub async fn head_with_metadata(&self, key: &str) -> Result<Option<BlobHeadMeta>, MediaError> {
|
||||
match self.bucket.head_object(key).await {
|
||||
match self.s3().bucket.head_object(key).await {
|
||||
Ok((result, _)) => Ok(Some(BlobHeadMeta {
|
||||
size: result.content_length.unwrap_or(0) as u64,
|
||||
})),
|
||||
@@ -199,7 +224,7 @@ impl MediaStorage {
|
||||
sha256: &str,
|
||||
) -> Result<BlobMeta, MediaError> {
|
||||
let key = Self::ctx_sidecar_key(ctx, sha256);
|
||||
let resp = self.bucket.get_object(&key).await?;
|
||||
let resp = self.s3().bucket.get_object(&key).await?;
|
||||
let meta: BlobMeta = serde_json::from_slice(&resp.to_vec())?;
|
||||
Ok(meta)
|
||||
}
|
||||
@@ -248,6 +273,7 @@ impl MediaStorage {
|
||||
max_keys: usize,
|
||||
) -> Result<crate::bucket_index::Page, MediaError> {
|
||||
let (result, _status) = self
|
||||
.s3()
|
||||
.bucket
|
||||
.list_page(
|
||||
String::new(),
|
||||
@@ -307,8 +333,8 @@ mod tests {
|
||||
fn static_keys_build_client_with_configured_region() {
|
||||
let storage = MediaStorage::new(&storage_config("buzz_dev", "buzz_dev_secret"))
|
||||
.expect("static creds should build a client");
|
||||
match storage.bucket.region {
|
||||
Region::Custom { ref region, .. } => assert_eq!(region, "us-west-2"),
|
||||
match &storage.s3().bucket.region {
|
||||
Region::Custom { region, .. } => assert_eq!(region, "us-west-2"),
|
||||
other => panic!("expected Custom region, got {other:?}"),
|
||||
}
|
||||
}
|
||||
@@ -317,15 +343,15 @@ mod tests {
|
||||
fn client_constructor_applies_both_addressing_styles() {
|
||||
let path = MediaStorage::new(&storage_config("buzz_dev", "buzz_dev_secret"))
|
||||
.expect("path-style client");
|
||||
assert!(path.bucket.is_path_style());
|
||||
assert_eq!(path.bucket.url(), "http://localhost:9000/buzz-media");
|
||||
assert!(path.s3().bucket.is_path_style());
|
||||
assert_eq!(path.s3().bucket.url(), "http://localhost:9000/buzz-media");
|
||||
|
||||
let mut virtual_config = storage_config("buzz_dev", "buzz_dev_secret");
|
||||
virtual_config.s3_addressing_style = S3AddressingStyle::Virtual;
|
||||
let virtual_hosted = MediaStorage::new(&virtual_config).expect("virtual-hosted client");
|
||||
assert!(virtual_hosted.bucket.is_subdomain_style());
|
||||
assert!(virtual_hosted.s3().bucket.is_subdomain_style());
|
||||
assert_eq!(
|
||||
virtual_hosted.bucket.url(),
|
||||
virtual_hosted.s3().bucket.url(),
|
||||
"http://buzz-media.localhost:9000"
|
||||
);
|
||||
}
|
||||
|
||||
@@ -96,8 +96,190 @@ impl PubSubConfig {
|
||||
}
|
||||
}
|
||||
|
||||
/// Central pub/sub manager for a Buzz relay instance.
|
||||
/// Backend-neutral pub/sub handle selected once at relay startup.
|
||||
///
|
||||
/// Request handlers use this stable surface; backend selection never occurs in
|
||||
/// the event, presence, or subscription paths.
|
||||
pub struct PubSubManager {
|
||||
backend: PubSubBackend,
|
||||
}
|
||||
|
||||
enum PubSubBackend {
|
||||
Redis(Arc<RedisPubSubManager>),
|
||||
}
|
||||
|
||||
impl PubSubManager {
|
||||
/// Creates the production Redis pub/sub backend.
|
||||
pub async fn new(redis_url: &str, pool: deadpool_redis::Pool) -> Result<Self, PubSubError> {
|
||||
Self::with_config(PubSubConfig::new(redis_url), pool).await
|
||||
}
|
||||
|
||||
/// Creates the production Redis backend with explicit configuration.
|
||||
pub async fn with_config(
|
||||
config: PubSubConfig,
|
||||
pool: deadpool_redis::Pool,
|
||||
) -> Result<Self, PubSubError> {
|
||||
Ok(Self {
|
||||
backend: PubSubBackend::Redis(Arc::new(
|
||||
RedisPubSubManager::with_config(config, pool).await?,
|
||||
)),
|
||||
})
|
||||
}
|
||||
|
||||
/// Runs the backend event subscriber. Process-local backends may implement
|
||||
/// this as a pending task because publication already reaches local receivers.
|
||||
pub async fn run_subscriber(self: Arc<Self>) {
|
||||
match &self.backend {
|
||||
PubSubBackend::Redis(backend) => Arc::clone(backend).run_subscriber().await,
|
||||
}
|
||||
}
|
||||
|
||||
/// Runs the backend cache-invalidation subscriber.
|
||||
pub async fn run_cache_invalidation_subscriber(self: Arc<Self>) {
|
||||
match &self.backend {
|
||||
PubSubBackend::Redis(backend) => {
|
||||
Arc::clone(backend)
|
||||
.run_cache_invalidation_subscriber()
|
||||
.await
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/// Runs the backend connection-control subscriber.
|
||||
pub async fn run_conn_control_subscriber(self: Arc<Self>) {
|
||||
match &self.backend {
|
||||
PubSubBackend::Redis(backend) => {
|
||||
Arc::clone(backend).run_conn_control_subscriber().await
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/// Returns a receiver for locally visible channel events.
|
||||
pub fn subscribe_local(&self) -> broadcast::Receiver<ChannelEvent> {
|
||||
match &self.backend {
|
||||
PubSubBackend::Redis(backend) => backend.subscribe_local(),
|
||||
}
|
||||
}
|
||||
|
||||
/// Retains interest in a scoped event topic.
|
||||
pub async fn retain_topic(&self, ctx: &TenantContext, topic: EventTopic) {
|
||||
match &self.backend {
|
||||
PubSubBackend::Redis(backend) => backend.retain_topic(ctx, topic).await,
|
||||
}
|
||||
}
|
||||
|
||||
/// Releases interest in a scoped event topic.
|
||||
pub async fn release_topic(&self, ctx: &TenantContext, topic: EventTopic) {
|
||||
match &self.backend {
|
||||
PubSubBackend::Redis(backend) => backend.release_topic(ctx, topic).await,
|
||||
}
|
||||
}
|
||||
|
||||
/// Returns the current topic retain count.
|
||||
pub async fn topic_refcount(&self, ctx: &TenantContext, topic: EventTopic) -> usize {
|
||||
match &self.backend {
|
||||
PubSubBackend::Redis(backend) => backend.topic_refcount(ctx, topic).await,
|
||||
}
|
||||
}
|
||||
|
||||
/// Returns a receiver for cache invalidations.
|
||||
pub fn subscribe_cache_invalidations(&self) -> broadcast::Receiver<ScopedCacheInvalidation> {
|
||||
match &self.backend {
|
||||
PubSubBackend::Redis(backend) => backend.subscribe_cache_invalidations(),
|
||||
}
|
||||
}
|
||||
|
||||
/// Returns a receiver for connection-control commands.
|
||||
pub fn subscribe_conn_control(&self) -> broadcast::Receiver<ScopedConnControl> {
|
||||
match &self.backend {
|
||||
PubSubBackend::Redis(backend) => backend.subscribe_conn_control(),
|
||||
}
|
||||
}
|
||||
|
||||
/// Publishes a cache invalidation.
|
||||
pub async fn publish_cache_invalidation(
|
||||
&self,
|
||||
ctx: &TenantContext,
|
||||
invalidation: &CacheInvalidation,
|
||||
) -> Result<i64, PubSubError> {
|
||||
match &self.backend {
|
||||
PubSubBackend::Redis(backend) => {
|
||||
backend.publish_cache_invalidation(ctx, invalidation).await
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/// Publishes a connection-control command.
|
||||
pub async fn publish_conn_control(
|
||||
&self,
|
||||
ctx: &TenantContext,
|
||||
command: &ConnControl,
|
||||
) -> Result<i64, PubSubError> {
|
||||
match &self.backend {
|
||||
PubSubBackend::Redis(backend) => backend.publish_conn_control(ctx, command).await,
|
||||
}
|
||||
}
|
||||
|
||||
/// Publishes an event to the selected backend.
|
||||
pub async fn publish_event(
|
||||
&self,
|
||||
ctx: &TenantContext,
|
||||
topic: EventTopic,
|
||||
event: &nostr::Event,
|
||||
) -> Result<i64, PubSubError> {
|
||||
match &self.backend {
|
||||
PubSubBackend::Redis(backend) => backend.publish_event(ctx, topic, event).await,
|
||||
}
|
||||
}
|
||||
|
||||
/// Sets presence for a principal.
|
||||
pub async fn set_presence(
|
||||
&self,
|
||||
ctx: &TenantContext,
|
||||
pubkey: &PublicKey,
|
||||
status: &str,
|
||||
) -> Result<(), PubSubError> {
|
||||
match &self.backend {
|
||||
PubSubBackend::Redis(backend) => backend.set_presence(ctx, pubkey, status).await,
|
||||
}
|
||||
}
|
||||
|
||||
/// Clears presence for a principal.
|
||||
pub async fn clear_presence(
|
||||
&self,
|
||||
ctx: &TenantContext,
|
||||
pubkey: &PublicKey,
|
||||
) -> Result<(), PubSubError> {
|
||||
match &self.backend {
|
||||
PubSubBackend::Redis(backend) => backend.clear_presence(ctx, pubkey).await,
|
||||
}
|
||||
}
|
||||
|
||||
/// Gets presence for one principal.
|
||||
pub async fn get_presence(
|
||||
&self,
|
||||
ctx: &TenantContext,
|
||||
pubkey: &PublicKey,
|
||||
) -> Result<Option<String>, PubSubError> {
|
||||
match &self.backend {
|
||||
PubSubBackend::Redis(backend) => backend.get_presence(ctx, pubkey).await,
|
||||
}
|
||||
}
|
||||
|
||||
/// Gets presence for multiple principals.
|
||||
pub async fn get_presence_bulk(
|
||||
&self,
|
||||
ctx: &TenantContext,
|
||||
pubkeys: &[PublicKey],
|
||||
) -> Result<HashMap<String, String>, PubSubError> {
|
||||
match &self.backend {
|
||||
PubSubBackend::Redis(backend) => backend.get_presence_bulk(ctx, pubkeys).await,
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/// Redis implementation behind [`PubSubManager`].
|
||||
struct RedisPubSubManager {
|
||||
pool: deadpool_redis::Pool,
|
||||
/// Redis URL used by the reconnect loop to re-establish pub/sub connections.
|
||||
redis_url: String,
|
||||
@@ -112,12 +294,7 @@ pub struct PubSubManager {
|
||||
conn_control_tx: broadcast::Sender<ScopedConnControl>,
|
||||
}
|
||||
|
||||
impl PubSubManager {
|
||||
/// Creates a new `PubSubManager` connected to the given Redis URL.
|
||||
pub async fn new(redis_url: &str, pool: deadpool_redis::Pool) -> Result<Self, PubSubError> {
|
||||
Self::with_config(PubSubConfig::new(redis_url), pool).await
|
||||
}
|
||||
|
||||
impl RedisPubSubManager {
|
||||
/// Creates a new `PubSubManager` using explicit pub/sub configuration.
|
||||
pub async fn with_config(
|
||||
config: PubSubConfig,
|
||||
|
||||
@@ -78,6 +78,60 @@ async fn run_rate_limit(
|
||||
}
|
||||
}
|
||||
|
||||
/// Backend-neutral admission limiter selected once at relay startup.
|
||||
pub enum AdmissionRateLimiter {
|
||||
/// Production Redis fixed-window counters.
|
||||
Redis(RedisRateLimiter),
|
||||
/// Process-local profile policy until bounded in-memory counters are installed.
|
||||
///
|
||||
/// This is explicit rather than a fallback: callers must select it at startup.
|
||||
Permissive,
|
||||
}
|
||||
|
||||
impl AdmissionRateLimiter {
|
||||
/// Creates the production Redis limiter.
|
||||
pub fn redis(pool: deadpool_redis::Pool) -> Self {
|
||||
Self::Redis(RedisRateLimiter::new(pool))
|
||||
}
|
||||
|
||||
/// Creates the explicit permissive policy for a process-local relay.
|
||||
pub fn permissive() -> Self {
|
||||
Self::Permissive
|
||||
}
|
||||
}
|
||||
|
||||
impl RateLimiter for AdmissionRateLimiter {
|
||||
async fn check_and_increment(
|
||||
&self,
|
||||
ctx: &TenantContext,
|
||||
pubkey: &PublicKey,
|
||||
limit_type: LimitType,
|
||||
window_secs: u64,
|
||||
limit: u64,
|
||||
) -> Result<RateLimitResult, AuthError> {
|
||||
match self {
|
||||
Self::Redis(limiter) => {
|
||||
limiter
|
||||
.check_and_increment(ctx, pubkey, limit_type, window_secs, limit)
|
||||
.await
|
||||
}
|
||||
Self::Permissive => Ok(RateLimitResult::allowed(1, limit, window_secs)),
|
||||
}
|
||||
}
|
||||
|
||||
async fn check_ip_connection(
|
||||
&self,
|
||||
ip: &IpAddr,
|
||||
window_secs: u64,
|
||||
limit: u64,
|
||||
) -> Result<RateLimitResult, AuthError> {
|
||||
match self {
|
||||
Self::Redis(limiter) => limiter.check_ip_connection(ip, window_secs, limit).await,
|
||||
Self::Permissive => Ok(RateLimitResult::allowed(1, limit, window_secs)),
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/// Redis-backed rate limiter using fixed-window counters.
|
||||
///
|
||||
/// Pubkey keys are community-scoped via `&TenantContext`:
|
||||
@@ -119,3 +173,36 @@ impl RateLimiter for RedisRateLimiter {
|
||||
run_rate_limit(&self.pool, &key, window_secs, limit).await
|
||||
}
|
||||
}
|
||||
|
||||
#[cfg(test)]
|
||||
mod admission_backend_tests {
|
||||
use super::*;
|
||||
use buzz_core::CommunityId;
|
||||
use nostr::Keys;
|
||||
use std::net::{IpAddr, Ipv4Addr};
|
||||
use uuid::Uuid;
|
||||
|
||||
#[tokio::test]
|
||||
async fn permissive_backend_allows_principal_and_ip_without_redis() {
|
||||
let limiter = AdmissionRateLimiter::permissive();
|
||||
let tenant =
|
||||
TenantContext::resolved(CommunityId::from_uuid(Uuid::from_u128(1)), "local.invalid");
|
||||
let principal = limiter
|
||||
.check_and_increment(
|
||||
&tenant,
|
||||
&Keys::generate().public_key(),
|
||||
LimitType::Messages,
|
||||
60,
|
||||
10,
|
||||
)
|
||||
.await
|
||||
.expect("permissive principal policy");
|
||||
let ip = limiter
|
||||
.check_ip_connection(&IpAddr::V4(Ipv4Addr::LOCALHOST), 60, 10)
|
||||
.await
|
||||
.expect("permissive IP policy");
|
||||
|
||||
assert!(principal.allowed);
|
||||
assert!(ip.allowed);
|
||||
}
|
||||
}
|
||||
|
||||
@@ -50,9 +50,41 @@ pub struct JoinPolicyConfig {
|
||||
/// WebSocket close-frame delivery after the final delayed cancellation.
|
||||
pub const MAX_DRAIN_JITTER_MS: u64 = 20_000;
|
||||
|
||||
/// Relay runtime profile. The default preserves the production service graph;
|
||||
/// `single-node` selects only process-local or filesystem-backed services.
|
||||
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
|
||||
pub enum RelayProfile {
|
||||
/// Postgres, Redis, S3, and Postgres FTS.
|
||||
Production,
|
||||
/// One-process local relay with no external service dependencies.
|
||||
SingleNode,
|
||||
}
|
||||
|
||||
impl RelayProfile {
|
||||
/// Whether this profile must avoid every external service connection.
|
||||
pub fn is_single_node(self) -> bool {
|
||||
matches!(self, Self::SingleNode)
|
||||
}
|
||||
|
||||
fn from_env() -> Result<Self, ConfigError> {
|
||||
match std::env::var("BUZZ_PROFILE") {
|
||||
Err(_) => Ok(Self::Production),
|
||||
Ok(value) if value.trim().is_empty() || value.eq_ignore_ascii_case("production") => {
|
||||
Ok(Self::Production)
|
||||
}
|
||||
Ok(value) if value.eq_ignore_ascii_case("single-node") => Ok(Self::SingleNode),
|
||||
Ok(value) => Err(ConfigError::InvalidValue(format!(
|
||||
"BUZZ_PROFILE must be 'production' or 'single-node', got {value:?}"
|
||||
))),
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/// Relay runtime configuration, loaded from environment variables.
|
||||
#[derive(Debug, Clone)]
|
||||
pub struct Config {
|
||||
/// Service topology selected by `BUZZ_PROFILE`.
|
||||
pub profile: RelayProfile,
|
||||
/// Address the relay HTTP/WebSocket server binds to.
|
||||
pub bind_addr: SocketAddr,
|
||||
/// Postgres database connection URL.
|
||||
@@ -459,6 +491,7 @@ fn inert_env_vars<'a>(names: &[&'a str], lookup: impl Fn(&str) -> Option<String>
|
||||
impl Config {
|
||||
/// Loads configuration from environment variables, falling back to development defaults.
|
||||
pub fn from_env() -> Result<Self, ConfigError> {
|
||||
let profile = RelayProfile::from_env()?;
|
||||
let bind_addr_raw =
|
||||
std::env::var("BUZZ_BIND_ADDR").unwrap_or_else(|_| "0.0.0.0:3000".to_string());
|
||||
let bind_addr = parse_bind_addr(&bind_addr_raw)?;
|
||||
@@ -987,6 +1020,7 @@ impl Config {
|
||||
}
|
||||
|
||||
Ok(Self {
|
||||
profile,
|
||||
bind_addr,
|
||||
database_url,
|
||||
read_database_url,
|
||||
|
||||
@@ -32,10 +32,10 @@ pub(crate) struct Config {
|
||||
|
||||
impl Config {
|
||||
pub(crate) fn from_env() -> anyhow::Result<Option<Self>> {
|
||||
let enabled = std::env::var("BUZZ_LOCAL_MODE")
|
||||
let legacy_spike_enabled = std::env::var("BUZZ_LOCAL_MODE")
|
||||
.ok()
|
||||
.is_some_and(|value| matches!(value.trim(), "1" | "true" | "on"));
|
||||
if !enabled {
|
||||
if !legacy_spike_enabled {
|
||||
return Ok(None);
|
||||
}
|
||||
let bind_addr = std::env::var("BUZZ_BIND_ADDR")
|
||||
|
||||
@@ -159,6 +159,15 @@ async fn main() -> anyhow::Result<()> {
|
||||
"Config loaded"
|
||||
);
|
||||
|
||||
// Phase 1 constructor fence: never fall through to production's eager
|
||||
// Postgres/Redis/S3 graph for a single-node profile. The local backend
|
||||
// constructor replaces this explicit denial as its implementations land.
|
||||
if config.profile.is_single_node() {
|
||||
return Err(anyhow::anyhow!(
|
||||
"BUZZ_PROFILE=single-node local backend bundle is not installed yet"
|
||||
));
|
||||
}
|
||||
|
||||
let usage_interval_secs = usage_metrics_interval_secs();
|
||||
let usage_idle_timeout_secs = usage_metrics_idle_timeout_secs(usage_interval_secs);
|
||||
relay_metrics::install(config.metrics_port, usage_idle_timeout_secs);
|
||||
@@ -474,7 +483,10 @@ async fn main() -> anyhow::Result<()> {
|
||||
// operator who asked for the mesh gets it or gets told why not.
|
||||
if let Some(handle) = buzz_relay::mesh_boot::boot_mesh(
|
||||
&state.config,
|
||||
state.redis_pool.clone(),
|
||||
state
|
||||
.redis_pool
|
||||
.clone()
|
||||
.expect("production profile has Redis"),
|
||||
&state.relay_keypair,
|
||||
Arc::clone(&state.shutting_down),
|
||||
)
|
||||
@@ -1019,11 +1031,13 @@ async fn main() -> anyhow::Result<()> {
|
||||
}
|
||||
}
|
||||
|
||||
let rs = pool_state.redis_pool.status();
|
||||
metrics::gauge!("buzz_redis_pool_available").set(rs.available as f64);
|
||||
metrics::gauge!("buzz_redis_pool_size").set(rs.size as f64);
|
||||
metrics::gauge!("buzz_redis_pool_max").set(rs.max_size as f64);
|
||||
metrics::gauge!("buzz_redis_pool_waiting").set(rs.waiting as f64);
|
||||
if let Some(redis_pool) = &pool_state.redis_pool {
|
||||
let rs = redis_pool.status();
|
||||
metrics::gauge!("buzz_redis_pool_available").set(rs.available as f64);
|
||||
metrics::gauge!("buzz_redis_pool_size").set(rs.size as f64);
|
||||
metrics::gauge!("buzz_redis_pool_max").set(rs.max_size as f64);
|
||||
metrics::gauge!("buzz_redis_pool_waiting").set(rs.waiting as f64);
|
||||
}
|
||||
}
|
||||
});
|
||||
}
|
||||
|
||||
@@ -377,8 +377,11 @@ async fn readiness_handler(State(state): State<Arc<AppState>>) -> impl IntoRespo
|
||||
|
||||
let check = async {
|
||||
let (pg_ok, redis_ok) = tokio::join!(state.db.ping(), async {
|
||||
state.redis_pool.get().await.is_ok()
|
||||
},);
|
||||
match &state.redis_pool {
|
||||
Some(pool) => pool.get().await.is_ok(),
|
||||
None => true,
|
||||
}
|
||||
});
|
||||
(pg_ok, redis_ok)
|
||||
};
|
||||
|
||||
|
||||
@@ -24,7 +24,7 @@ use buzz_db::Db;
|
||||
use buzz_media::MediaStorage;
|
||||
use buzz_pubsub::cache_invalidation::CacheInvalidation;
|
||||
use buzz_pubsub::conn_control::ConnControl;
|
||||
use buzz_pubsub::rate_limiter::RedisRateLimiter;
|
||||
use buzz_pubsub::rate_limiter::AdmissionRateLimiter;
|
||||
use buzz_pubsub::{PubSubManager, RedisNip98ReplayGuard};
|
||||
use buzz_search::SearchService;
|
||||
use buzz_workflow::WorkflowEngine;
|
||||
@@ -575,6 +575,29 @@ impl Default for ConnectionManager {
|
||||
}
|
||||
}
|
||||
|
||||
/// Runtime backends injected at the application-state boundary.
|
||||
///
|
||||
/// Production construction remains in `main.rs`; the single-node profile supplies
|
||||
/// alternate implementations without adding backend checks to request handlers.
|
||||
pub struct AppBackends {
|
||||
/// Database backend.
|
||||
pub db: Db,
|
||||
/// Redis health/command pool; absent for process-local profiles.
|
||||
pub redis_pool: Option<deadpool_redis::Pool>,
|
||||
/// Event and control-plane pub/sub.
|
||||
pub pubsub: Arc<PubSubManager>,
|
||||
/// Full-text search backend.
|
||||
pub search: SearchService,
|
||||
/// Media blob backend.
|
||||
pub media_storage: MediaStorage,
|
||||
/// Git object backend.
|
||||
pub git_store: crate::api::git::store::GitStore,
|
||||
/// NIP-98 replay fence.
|
||||
pub nip98_replay: Arc<dyn Nip98ReplayGuard>,
|
||||
/// Admission rate limiter.
|
||||
pub admission_rate_limiter: Arc<AdmissionRateLimiter>,
|
||||
}
|
||||
|
||||
/// Shared application state, cloned cheaply via inner `Arc` fields.
|
||||
#[derive(Clone)]
|
||||
pub struct AppState {
|
||||
@@ -582,8 +605,8 @@ pub struct AppState {
|
||||
pub config: Arc<Config>,
|
||||
/// Database connection pool.
|
||||
pub db: Db,
|
||||
/// Redis pool for readiness health checks.
|
||||
pub redis_pool: deadpool_redis::Pool,
|
||||
/// Redis pool for readiness health checks; absent in single-node mode.
|
||||
pub redis_pool: Option<deadpool_redis::Pool>,
|
||||
/// Audit event service, absent when audit logging is disabled.
|
||||
pub audit: Option<Arc<AuditService>>,
|
||||
/// Pub/sub manager for broadcasting events to subscribers.
|
||||
@@ -673,7 +696,7 @@ pub struct AppState {
|
||||
/// cross-pod routing.
|
||||
pub nip98_replay: Arc<dyn Nip98ReplayGuard>,
|
||||
/// Shared Redis-backed admission limits for ordinary HTTP and WebSocket work.
|
||||
pub admission_rate_limiter: Arc<RedisRateLimiter>,
|
||||
pub admission_rate_limiter: Arc<AdmissionRateLimiter>,
|
||||
|
||||
/// Per-agent sliding-window rate limiter for observer frames (kind 24200).
|
||||
/// Key: (community_id, agent pubkey bytes). Value: (count, window_start).
|
||||
@@ -738,6 +761,57 @@ impl AppState {
|
||||
relay_keypair: nostr::Keys,
|
||||
media_storage: MediaStorage,
|
||||
) -> (Self, AuditShutdownHandle) {
|
||||
let git_store = crate::api::git::store::GitStore::new(
|
||||
&config.media.s3_endpoint,
|
||||
&config.media.s3_access_key,
|
||||
&config.media.s3_secret_key,
|
||||
&config.media.s3_bucket,
|
||||
&config.media.s3_region,
|
||||
config.media.s3_addressing_style,
|
||||
)
|
||||
.expect("media storage was already constructed with this S3 config");
|
||||
let nip98_replay: Arc<dyn Nip98ReplayGuard> =
|
||||
Arc::new(RedisNip98ReplayGuard::new(redis_pool.clone()));
|
||||
let admission_rate_limiter = Arc::new(AdmissionRateLimiter::redis(redis_pool.clone()));
|
||||
Self::new_with_backends(
|
||||
config,
|
||||
AppBackends {
|
||||
db,
|
||||
redis_pool: Some(redis_pool),
|
||||
pubsub,
|
||||
search,
|
||||
media_storage,
|
||||
git_store,
|
||||
nip98_replay,
|
||||
admission_rate_limiter,
|
||||
},
|
||||
audit,
|
||||
auth,
|
||||
workflow_engine,
|
||||
relay_keypair,
|
||||
)
|
||||
}
|
||||
|
||||
/// Constructs state from an explicit backend bundle.
|
||||
#[allow(clippy::too_many_arguments)]
|
||||
pub fn new_with_backends(
|
||||
config: Config,
|
||||
backends: AppBackends,
|
||||
audit: impl Into<Option<AuditService>>,
|
||||
auth: AuthService,
|
||||
workflow_engine: Arc<WorkflowEngine>,
|
||||
relay_keypair: nostr::Keys,
|
||||
) -> (Self, AuditShutdownHandle) {
|
||||
let AppBackends {
|
||||
db,
|
||||
redis_pool,
|
||||
pubsub,
|
||||
search,
|
||||
media_storage,
|
||||
git_store,
|
||||
nip98_replay,
|
||||
admission_rate_limiter,
|
||||
} = backends;
|
||||
let max_connections = config.max_connections;
|
||||
let max_concurrent_handlers = config.max_concurrent_handlers;
|
||||
let search_arc = Arc::new(search);
|
||||
@@ -783,15 +857,6 @@ impl AppState {
|
||||
|
||||
let git_max_concurrent_ops = config.git_max_concurrent_ops;
|
||||
let media_max_concurrent_uploads = config.media_max_concurrent_uploads;
|
||||
let git_store = crate::api::git::store::GitStore::new(
|
||||
&config.media.s3_endpoint,
|
||||
&config.media.s3_access_key,
|
||||
&config.media.s3_secret_key,
|
||||
&config.media.s3_bucket,
|
||||
&config.media.s3_region,
|
||||
config.media.s3_addressing_style,
|
||||
)
|
||||
.expect("media storage was already constructed with this S3 config");
|
||||
let git_pack_cache = Arc::new(
|
||||
crate::api::git::pack_cache::GitPackCache::new(
|
||||
&config.git_pack_cache_path,
|
||||
@@ -800,9 +865,6 @@ impl AppState {
|
||||
)
|
||||
.expect("git pack cache path must be available"),
|
||||
);
|
||||
let nip98_replay: Arc<dyn Nip98ReplayGuard> =
|
||||
Arc::new(RedisNip98ReplayGuard::new(redis_pool.clone()));
|
||||
let admission_rate_limiter = Arc::new(RedisRateLimiter::new(redis_pool.clone()));
|
||||
let audit_enabled = audit_arc.is_some();
|
||||
let state = Self {
|
||||
config: Arc::new(config),
|
||||
|
||||
@@ -3,6 +3,9 @@ use thiserror::Error;
|
||||
/// Errors produced by the FTS service.
|
||||
#[derive(Debug, Error)]
|
||||
pub enum SearchError {
|
||||
/// Search is intentionally unavailable in the selected runtime profile.
|
||||
#[error("search unsupported by runtime profile")]
|
||||
Unsupported,
|
||||
/// A database error from sqlx.
|
||||
#[error("database error: {0}")]
|
||||
Db(#[from] sqlx::Error),
|
||||
|
||||
@@ -32,23 +32,39 @@ pub use query::{search, ChannelScope, SearchHit, SearchMode, SearchQuery, Search
|
||||
|
||||
use sqlx::PgPool;
|
||||
|
||||
/// Thin handle around a `PgPool` for community-scoped FTS.
|
||||
///
|
||||
/// Holds nothing the pool itself doesn't already own. The whole purpose of
|
||||
/// this type is a stable injection point for the relay's `AppState`.
|
||||
/// Search backend selected at startup.
|
||||
#[derive(Debug, Clone)]
|
||||
enum SearchBackend {
|
||||
Postgres(PgPool),
|
||||
Unsupported,
|
||||
}
|
||||
|
||||
/// Stable search-service injection point.
|
||||
#[derive(Debug, Clone)]
|
||||
pub struct SearchService {
|
||||
pool: PgPool,
|
||||
backend: SearchBackend,
|
||||
}
|
||||
|
||||
impl SearchService {
|
||||
/// Build a search service over an existing Postgres pool.
|
||||
pub fn new(pool: PgPool) -> Self {
|
||||
Self { pool }
|
||||
Self {
|
||||
backend: SearchBackend::Postgres(pool),
|
||||
}
|
||||
}
|
||||
|
||||
/// Build a backend that explicitly reports search as unsupported.
|
||||
pub fn unsupported() -> Self {
|
||||
Self {
|
||||
backend: SearchBackend::Unsupported,
|
||||
}
|
||||
}
|
||||
|
||||
/// Execute a community-scoped FTS query.
|
||||
pub async fn search(&self, query: &SearchQuery) -> Result<SearchResult, SearchError> {
|
||||
query::search(&self.pool, query).await
|
||||
match &self.backend {
|
||||
SearchBackend::Postgres(pool) => query::search(pool, query).await,
|
||||
SearchBackend::Unsupported => Err(SearchError::Unsupported),
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user