From 4b89e5d33f6a83755c780b1f4c4b1ae83bfc9ca0 Mon Sep 17 00:00:00 2001 From: mdecimus Date: Wed, 21 May 2025 17:33:24 +0200 Subject: [PATCH] Add cluster orchestration support for Kafka and Zenoh --- Cargo.lock | 63 ++++++++++++++ README.md | 1 + crates/store/Cargo.toml | 6 +- crates/store/src/backend/kafka/mod.rs | 105 +++++++++++++++++++++++ crates/store/src/backend/kafka/pubsub.rs | 57 ++++++++++++ crates/store/src/backend/mod.rs | 4 + crates/store/src/backend/nats/mod.rs | 6 +- crates/store/src/backend/nats/pubsub.rs | 22 ++--- crates/store/src/backend/zenoh/mod.rs | 38 ++++++++ crates/store/src/backend/zenoh/pubsub.rs | 47 ++++++++++ crates/store/src/config.rs | 22 ++++- crates/store/src/dispatch/pubsub.rs | 28 ++++++ crates/store/src/lib.rs | 6 +- crates/trc/src/event/description.rs | 2 - crates/trc/src/event/level.rs | 1 - crates/trc/src/lib.rs | 1 - crates/trc/src/serializers/binary.rs | 6 +- 17 files changed, 385 insertions(+), 30 deletions(-) create mode 100644 crates/store/src/backend/kafka/mod.rs create mode 100644 crates/store/src/backend/kafka/pubsub.rs create mode 100644 crates/store/src/backend/zenoh/mod.rs create mode 100644 crates/store/src/backend/zenoh/pubsub.rs diff --git a/Cargo.lock b/Cargo.lock index 77cb58e4..0eeb4251 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -1231,6 +1231,15 @@ dependencies = [ "digest 0.10.7", ] +[[package]] +name = "cmake" +version = "0.1.54" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "e7caa3f9de89ddbe2c607f4101924c5abec803763ae9534e4f4d7d8f84aa81f0" +dependencies = [ + "cc", +] + [[package]] name = "colorchoice" version = "1.0.3" @@ -4197,6 +4206,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "8b70e7a7df205e92a1a4cd9aaae7898dac0aa555503cc0a649494d0d60e7651d" dependencies = [ "cc", + "libc", "pkg-config", "vcpkg", ] @@ -4867,6 +4877,27 @@ dependencies = [ "libc", ] +[[package]] +name = "num_enum" +version = "0.7.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "4e613fc340b2220f734a8595782c551f1250e969d87d3be1ae0579e8d4065179" +dependencies = [ + "num_enum_derive", +] + +[[package]] +name = "num_enum_derive" +version = "0.7.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "af1844ef2428cc3e1cb900be36181049ef3d3193c63e43026cfe202983b27a56" +dependencies = [ + "proc-macro-crate", + "proc-macro2", + "quote", + "syn 2.0.101", +] + [[package]] name = "number_prefix" version = "0.4.0" @@ -6050,6 +6081,37 @@ dependencies = [ "yasna", ] +[[package]] +name = "rdkafka" +version = "0.37.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "14b52c81ac3cac39c9639b95c20452076e74b8d9a71bc6fc4d83407af2ea6fff" +dependencies = [ + "futures-channel", + "futures-util", + "libc", + "log", + "rdkafka-sys", + "serde", + "serde_derive", + "serde_json", + "slab", + "tokio", +] + +[[package]] +name = "rdkafka-sys" +version = "4.8.0+2.3.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "ced38182dc436b3d9df0c77976f37a67134df26b050df1f0006688e46fc4c8be" +dependencies = [ + "cmake", + "libc", + "libz-sys", + "num_enum", + "pkg-config", +] + [[package]] name = "redis" version = "0.31.0" @@ -7672,6 +7734,7 @@ dependencies = [ "r2d2", "rand 0.9.1", "rayon", + "rdkafka", "redis", "regex", "reqwest 0.12.15", diff --git a/README.md b/README.md index 6f98a3a9..8950512b 100644 --- a/README.md +++ b/README.md @@ -84,6 +84,7 @@ Key features: - Memory safe (thanks to Rust). - **Scalable and fault-tolerant**: - Designed to handle growth seamlessly, from small setups to large-scale deployments. + - Coordinator-less cluster orchestration or with **Kafka**, **NATS** or **Redis**. - Built with **fault tolerance** and **high availability** in mind, recovers from hardware or software failures with minimal operational impact. - **Kubernetes** support for automated scaling and efficient container orchestration. - Read replicas, sharded blob storage and in-memory data stores for high performance and low latency. diff --git a/crates/store/Cargo.toml b/crates/store/Cargo.toml index f6725c7f..1f896504 100644 --- a/crates/store/Cargo.toml +++ b/crates/store/Cargo.toml @@ -53,6 +53,7 @@ memchr = { version = "2" } rkyv = { version = "0.8.10", features = ["little_endian"] } compact_str = "0.9.0" zenoh = { version = "1.3.4", default-features = false, features = ["auth_pubkey", "transport_multilink", "transport_compression", "transport_quic", "transport_tcp", "transport_tls", "transport_udp"], optional = true } +rdkafka = { version = "0.37.0", features = ["cmake-build"], optional = true } [dev-dependencies] tokio = { version = "1.45", features = ["full"] } @@ -79,9 +80,8 @@ redis = ["dep:redis", "deadpool"] # Pubsub nats = ["async-nats"] -peer-to-peer = ["zenoh"] +zenoh = ["dep:zenoh"] +kafka = ["rdkafka"] enterprise = [] test_mode = [] - - diff --git a/crates/store/src/backend/kafka/mod.rs b/crates/store/src/backend/kafka/mod.rs new file mode 100644 index 00000000..91293f47 --- /dev/null +++ b/crates/store/src/backend/kafka/mod.rs @@ -0,0 +1,105 @@ +/* + * SPDX-FileCopyrightText: 2020 Stalwart Labs Ltd + * + * SPDX-License-Identifier: AGPL-3.0-only OR LicenseRef-SEL + */ + +use rdkafka::{ + ClientConfig, ClientContext, TopicPartitionList, + consumer::{BaseConsumer, ConsumerContext, Rebalance, StreamConsumer}, + error::KafkaResult, + producer::FutureProducer, +}; +use std::{fmt::Debug, time::Duration}; +use utils::config::{Config, utils::AsKey}; + +pub mod pubsub; + +pub(super) type LoggingConsumer = StreamConsumer; + +pub struct KafkaPubSub { + consumer_builder: ClientConfig, + producer: FutureProducer, +} + +impl KafkaPubSub { + pub async fn open(config: &mut Config, prefix: impl AsKey) -> Option { + let prefix = prefix.as_key(); + let brokers = config + .values((&prefix, "brokers")) + .map(|(_, v)| v.to_string()) + .collect::>(); + if brokers.is_empty() { + config.new_build_error((&prefix, "brokers"), "No Kafka brokers specified"); + return None; + } + + let mut consumer_builder = ClientConfig::new(); + + consumer_builder + .set( + "group.id", + config.value_require_non_empty((&prefix, "group-id"))?, + ) + .set( + "bootstrap.servers", + config.value_require_non_empty((&prefix, "brokers"))?, + ) + .set("enable.partition.eof", "false") + .set( + "session.timeout.ms", + config + .property_or_default((&prefix, "timeout.session"), "5s") + .unwrap_or(Duration::from_secs(5)) + .as_millis() + .to_string(), + ) + .set("enable.auto.commit", "true"); + + let producer = ClientConfig::new() + .set( + "bootstrap.servers", + config.value_require_non_empty((&prefix, "brokers"))?, + ) + .set( + "message.timeout.ms", + config + .property_or_default((&prefix, "timeout.message"), "5s") + .unwrap_or(Duration::from_secs(5)) + .as_millis() + .to_string(), + ) + .create() + .map_err(|err| { + config.new_build_error( + (&prefix, "config"), + format!("Failed to create Kafka producer: {}", err), + ); + }) + .ok()?; + + KafkaPubSub { + consumer_builder, + producer, + } + .into() + } +} + +impl Debug for KafkaPubSub { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + f.debug_struct("KafkaPubSub").finish() + } +} + +pub(super) struct CustomContext; + +impl ClientContext for CustomContext {} + +impl ConsumerContext for CustomContext { + fn pre_rebalance(&self, _: &BaseConsumer, _: &Rebalance) {} + + fn post_rebalance(&self, _: &BaseConsumer, _: &Rebalance) {} + + fn commit_callback(&self, _: KafkaResult<()>, _: &TopicPartitionList) {} +} diff --git a/crates/store/src/backend/kafka/pubsub.rs b/crates/store/src/backend/kafka/pubsub.rs new file mode 100644 index 00000000..01f88633 --- /dev/null +++ b/crates/store/src/backend/kafka/pubsub.rs @@ -0,0 +1,57 @@ +/* + * SPDX-FileCopyrightText: 2020 Stalwart Labs Ltd + * + * SPDX-License-Identifier: AGPL-3.0-only OR LicenseRef-SEL + */ + +use std::time::Duration; + +use super::{CustomContext, KafkaPubSub, LoggingConsumer}; +use crate::dispatch::pubsub::{Msg, PubSubStream}; +use rdkafka::{ + Message, + consumer::{CommitMode, Consumer, StreamConsumer}, + producer::FutureRecord, +}; +use trc::{ClusterEvent, Error, EventType}; + +pub struct KafkaPubSubStream { + subs: LoggingConsumer, +} + +impl KafkaPubSub { + pub async fn publish(&self, topic: &'static str, message: Vec) -> trc::Result<()> { + self.producer + .send( + FutureRecord::<(), [u8]>::to(topic).payload(message.as_slice()), + Duration::from_secs(0), + ) + .await + .map(|_| ()) + .map_err(|(err, _)| { + Error::new(EventType::Cluster(ClusterEvent::PublisherError)).reason(err) + }) + } + + pub async fn subscribe(&self, topic: &'static str) -> trc::Result { + let subs: StreamConsumer = self + .consumer_builder + .create_with_context(CustomContext) + .map_err(|err| { + Error::new(EventType::Cluster(ClusterEvent::SubscriberError)).reason(err) + })?; + subs.subscribe(&[topic]).map_err(|err| { + Error::new(EventType::Cluster(ClusterEvent::SubscriberError)).reason(err) + })?; + + Ok(PubSubStream::Kafka(KafkaPubSubStream { subs })) + } +} + +impl KafkaPubSubStream { + pub async fn next(&mut self) -> Option { + let msg = self.subs.recv().await.ok()?; + let _ = self.subs.commit_message(&msg, CommitMode::Async); + Msg::Kafka(msg.payload().unwrap_or_default().to_vec()).into() + } +} diff --git a/crates/store/src/backend/mod.rs b/crates/store/src/backend/mod.rs index 62a63340..0e93051b 100644 --- a/crates/store/src/backend/mod.rs +++ b/crates/store/src/backend/mod.rs @@ -14,6 +14,8 @@ pub mod elastic; pub mod foundationdb; pub mod fs; pub mod http; +#[cfg(feature = "kafka")] +pub mod kafka; pub mod memory; #[cfg(feature = "mysql")] pub mod mysql; @@ -29,6 +31,8 @@ pub mod rocksdb; pub mod s3; #[cfg(feature = "sqlite")] pub mod sqlite; +#[cfg(feature = "zenoh")] +pub mod zenoh; pub const MAX_TOKEN_LENGTH: usize = (u8::MAX >> 1) as usize; pub const MAX_TOKEN_MASK: usize = MAX_TOKEN_LENGTH - 1; diff --git a/crates/store/src/backend/nats/mod.rs b/crates/store/src/backend/nats/mod.rs index 359280b3..1565d038 100644 --- a/crates/store/src/backend/nats/mod.rs +++ b/crates/store/src/backend/nats/mod.rs @@ -12,11 +12,11 @@ use utils::config::{Config, utils::AsKey}; pub mod pubsub; #[derive(Debug)] -pub struct NatsStore { +pub struct NatsPubSub { client: Client, } -impl NatsStore { +impl NatsPubSub { pub async fn open(config: &mut Config, prefix: impl AsKey) -> Option { let prefix = prefix.as_key(); let urls = config @@ -102,7 +102,7 @@ impl NatsStore { format!("Failed to connect to Nats: {}", err), ); }) - .map(|client| NatsStore { client }) + .map(|client| NatsPubSub { client }) .ok() } } diff --git a/crates/store/src/backend/nats/pubsub.rs b/crates/store/src/backend/nats/pubsub.rs index 81329a9e..728fd8c7 100644 --- a/crates/store/src/backend/nats/pubsub.rs +++ b/crates/store/src/backend/nats/pubsub.rs @@ -4,32 +4,31 @@ * SPDX-License-Identifier: AGPL-3.0-only OR LicenseRef-SEL */ -use std::fmt::Display; - -use futures::StreamExt; - +use super::NatsPubSub; use crate::dispatch::pubsub::{Msg, PubSubStream}; - -use super::NatsStore; +use futures::StreamExt; +use trc::{ClusterEvent, Error, EventType}; pub struct NatsPubSubStream { subs: async_nats::Subscriber, } -impl NatsStore { +impl NatsPubSub { pub async fn publish(&self, topic: &'static str, message: Vec) -> trc::Result<()> { self.client .publish(topic, message.into()) .await - .map_err(into_error) + .map_err(|err| Error::new(EventType::Cluster(ClusterEvent::PublisherError)).reason(err)) } pub async fn subscribe(&self, topic: &'static str) -> trc::Result { self.client .subscribe(topic) .await - .map_err(into_error) .map(|subs| PubSubStream::Nats(NatsPubSubStream { subs })) + .map_err(|err| { + Error::new(EventType::Cluster(ClusterEvent::SubscriberError)).reason(err) + }) } } @@ -38,8 +37,3 @@ impl NatsPubSubStream { self.subs.next().await.map(Msg::Nats) } } - -#[inline(always)] -fn into_error(err: impl Display) -> trc::Error { - trc::StoreEvent::NatsError.reason(err) -} diff --git a/crates/store/src/backend/zenoh/mod.rs b/crates/store/src/backend/zenoh/mod.rs new file mode 100644 index 00000000..b37ca69a --- /dev/null +++ b/crates/store/src/backend/zenoh/mod.rs @@ -0,0 +1,38 @@ +/* + * SPDX-FileCopyrightText: 2020 Stalwart Labs Ltd + * + * SPDX-License-Identifier: AGPL-3.0-only OR LicenseRef-SEL + */ + +use utils::config::{Config, utils::AsKey}; +pub mod pubsub; + +#[derive(Debug)] +pub struct ZenohPubSub { + session: zenoh::Session, +} + +impl ZenohPubSub { + pub async fn open(config: &mut Config, prefix: impl AsKey) -> Option { + let prefix = prefix.as_key(); + let zenoh_config = + zenoh::Config::from_json5(config.value_require_non_empty((&prefix, "config"))?) + .map_err(|err| { + config.new_build_error( + (&prefix, "config"), + format!("Invalid zenoh config: {}", err), + ); + }) + .ok()?; + zenoh::open(zenoh_config) + .await + .map_err(|err| { + config.new_build_error( + (&prefix, "config"), + format!("Failed to create zenoh session: {}", err), + ); + }) + .map(|session| ZenohPubSub { session }) + .ok() + } +} diff --git a/crates/store/src/backend/zenoh/pubsub.rs b/crates/store/src/backend/zenoh/pubsub.rs new file mode 100644 index 00000000..f133786e --- /dev/null +++ b/crates/store/src/backend/zenoh/pubsub.rs @@ -0,0 +1,47 @@ +/* + * SPDX-FileCopyrightText: 2020 Stalwart Labs Ltd + * + * SPDX-License-Identifier: AGPL-3.0-only OR LicenseRef-SEL + */ + +use super::ZenohPubSub; +use crate::dispatch::pubsub::{Msg, PubSubStream}; +use trc::{ClusterEvent, Error, EventType}; + +pub struct ZenohPubSubStream { + subs: zenoh::pubsub::Subscriber>, +} + +impl ZenohPubSub { + pub async fn publish(&self, topic: &'static str, message: Vec) -> trc::Result<()> { + self.session + .declare_publisher(topic) + .await + .map_err(|err| { + Error::new(EventType::Cluster(ClusterEvent::PublisherError)).reason(err) + })? + .put(message) + .await + .map_err(|err| Error::new(EventType::Cluster(ClusterEvent::PublisherError)).reason(err)) + } + + pub async fn subscribe(&self, topic: &'static str) -> trc::Result { + self.session + .declare_subscriber(topic) + .await + .map(|subs| PubSubStream::Zenoh(ZenohPubSubStream { subs })) + .map_err(|err| { + Error::new(EventType::Cluster(ClusterEvent::SubscriberError)).reason(err) + }) + } +} + +impl ZenohPubSubStream { + pub async fn next(&mut self) -> Option { + self.subs + .recv_async() + .await + .map(|sample| Msg::Zenoh(sample.payload().to_bytes().into_owned())) + .ok() + } +} diff --git a/crates/store/src/config.rs b/crates/store/src/config.rs index 4b7e3fdf..c6967529 100644 --- a/crates/store/src/config.rs +++ b/crates/store/src/config.rs @@ -215,7 +215,7 @@ impl Stores { } #[cfg(feature = "nats")] "nats" => { - if let Some(db) = crate::backend::nats::NatsStore::open(config, prefix) + if let Some(db) = crate::backend::nats::NatsPubSub::open(config, prefix) .await .map(std::sync::Arc::new) { @@ -223,6 +223,26 @@ impl Stores { .insert(store_id, crate::PubSubStore::Nats(db)); } } + #[cfg(feature = "zenoh")] + "zenoh" => { + if let Some(db) = crate::backend::zenoh::ZenohPubSub::open(config, prefix) + .await + .map(std::sync::Arc::new) + { + self.pubsub_stores + .insert(store_id, crate::PubSubStore::Zenoh(db)); + } + } + #[cfg(feature = "kafka")] + "kafka" => { + if let Some(db) = crate::backend::kafka::KafkaPubSub::open(config, prefix) + .await + .map(std::sync::Arc::new) + { + self.pubsub_stores + .insert(store_id, crate::PubSubStore::Kafka(db)); + } + } #[cfg(feature = "enterprise")] "sql-read-replica" => { #[cfg(any(feature = "postgres", feature = "mysql"))] diff --git a/crates/store/src/dispatch/pubsub.rs b/crates/store/src/dispatch/pubsub.rs index 11fc1e64..0e748cd2 100644 --- a/crates/store/src/dispatch/pubsub.rs +++ b/crates/store/src/dispatch/pubsub.rs @@ -13,6 +13,10 @@ pub enum PubSubStream { RedisCluster(crate::backend::redis::pubsub::RedisClusterPubSubStream), #[cfg(feature = "nats")] Nats(crate::backend::nats::pubsub::NatsPubSubStream), + #[cfg(feature = "zenoh")] + Zenoh(crate::backend::zenoh::pubsub::ZenohPubSubStream), + #[cfg(feature = "kafka")] + Kafka(crate::backend::kafka::pubsub::KafkaPubSubStream), #[cfg(not(any(feature = "redis", feature = "nats")))] Unimplemented, } @@ -22,6 +26,10 @@ pub enum Msg { Redis(redis::Msg), #[cfg(feature = "nats")] Nats(async_nats::Message), + #[cfg(feature = "zenoh")] + Zenoh(Vec), + #[cfg(feature = "kafka")] + Kafka(Vec), #[cfg(not(any(feature = "redis", feature = "nats")))] Unimplemented, } @@ -34,6 +42,10 @@ impl PubSubStore { PubSubStore::Redis(store) => store.publish(topic, message).await, #[cfg(feature = "nats")] PubSubStore::Nats(store) => store.publish(topic, message).await, + #[cfg(feature = "zenoh")] + PubSubStore::Zenoh(store) => store.publish(topic, message).await, + #[cfg(feature = "kafka")] + PubSubStore::Kafka(store) => store.publish(topic, message).await, PubSubStore::None => Err(trc::StoreEvent::NotSupported.into_err()), } } @@ -44,6 +56,10 @@ impl PubSubStore { PubSubStore::Redis(store) => store.subscribe(topic).await, #[cfg(feature = "nats")] PubSubStore::Nats(store) => store.subscribe(topic).await, + #[cfg(feature = "zenoh")] + PubSubStore::Zenoh(store) => store.subscribe(topic).await, + #[cfg(feature = "kafka")] + PubSubStore::Kafka(store) => store.subscribe(topic).await, PubSubStore::None => Err(trc::StoreEvent::NotSupported.into_err()), } } @@ -62,6 +78,10 @@ impl PubSubStream { PubSubStream::RedisCluster(stream) => stream.next().await, #[cfg(feature = "nats")] PubSubStream::Nats(stream) => stream.next().await, + #[cfg(feature = "zenoh")] + PubSubStream::Zenoh(stream) => stream.next().await, + #[cfg(feature = "kafka")] + PubSubStream::Kafka(stream) => stream.next().await, #[cfg(not(any(feature = "redis", feature = "nats")))] PubSubStream::Unimplemented => None, } @@ -75,6 +95,10 @@ impl Msg { Msg::Redis(msg) => msg.get_payload_bytes(), #[cfg(feature = "nats")] Msg::Nats(msg) => msg.payload.as_ref(), + #[cfg(feature = "zenoh")] + Msg::Zenoh(msg) => msg.as_slice(), + #[cfg(feature = "kafka")] + Msg::Kafka(msg) => msg.as_slice(), #[cfg(not(any(feature = "redis", feature = "nats")))] Msg::Unimplemented => &[], } @@ -86,6 +110,10 @@ impl Msg { Msg::Redis(msg) => msg.get_channel_name(), #[cfg(feature = "nats")] Msg::Nats(msg) => msg.subject.as_str(), + #[cfg(feature = "zenoh")] + Msg::Zenoh(_) => "", + #[cfg(feature = "kafka")] + Msg::Kafka(_) => "", #[cfg(not(any(feature = "redis", feature = "nats")))] Msg::Unimplemented => "", } diff --git a/crates/store/src/lib.rs b/crates/store/src/lib.rs index 81a6cfc4..f0440a57 100644 --- a/crates/store/src/lib.rs +++ b/crates/store/src/lib.rs @@ -226,7 +226,11 @@ pub enum PubSubStore { #[cfg(feature = "redis")] Redis(Arc), #[cfg(feature = "nats")] - Nats(Arc), + Nats(Arc), + #[cfg(feature = "zenoh")] + Zenoh(Arc), + #[cfg(feature = "kafka")] + Kafka(Arc), #[default] None, } diff --git a/crates/trc/src/event/description.rs b/crates/trc/src/event/description.rs index b34230b5..275ba124 100644 --- a/crates/trc/src/event/description.rs +++ b/crates/trc/src/event/description.rs @@ -1555,7 +1555,6 @@ impl StoreEvent { StoreEvent::DataIterate => "Data store iteration operation", StoreEvent::HttpStoreFetch => "HTTP store updated", StoreEvent::HttpStoreError => "Error updating HTTP store", - StoreEvent::NatsError => "NATS error", StoreEvent::CacheMiss => "Cache miss", StoreEvent::CacheHit => "Cache hit", StoreEvent::CacheStale => "Cache is stale", @@ -1597,7 +1596,6 @@ impl StoreEvent { StoreEvent::DataIterate => "A data store iteration operation was executed", StoreEvent::HttpStoreFetch => "The HTTP store was updated", StoreEvent::HttpStoreError => "An error occurred while updating the HTTP store", - StoreEvent::NatsError => "A NATS error occurred", StoreEvent::CacheMiss => "No cache entry found for the account", StoreEvent::CacheHit => "Cache entry found for the account, no update needed", StoreEvent::CacheStale => "Cache is too old, rebuilding", diff --git a/crates/trc/src/event/level.rs b/crates/trc/src/event/level.rs index 23b1e210..671851ef 100644 --- a/crates/trc/src/event/level.rs +++ b/crates/trc/src/event/level.rs @@ -35,7 +35,6 @@ impl EventType { | StoreEvent::LdapError | StoreEvent::ElasticsearchError | StoreEvent::RedisError - | StoreEvent::NatsError | StoreEvent::S3Error | StoreEvent::AzureError | StoreEvent::FilesystemError diff --git a/crates/trc/src/lib.rs b/crates/trc/src/lib.rs index 13e69a8f..8abaa68f 100644 --- a/crates/trc/src/lib.rs +++ b/crates/trc/src/lib.rs @@ -827,7 +827,6 @@ pub enum StoreEvent { ElasticsearchError, RedisError, S3Error, - NatsError, AzureError, FilesystemError, PoolError, diff --git a/crates/trc/src/serializers/binary.rs b/crates/trc/src/serializers/binary.rs index c61260ab..f109c10a 100644 --- a/crates/trc/src/serializers/binary.rs +++ b/crates/trc/src/serializers/binary.rs @@ -883,11 +883,10 @@ impl EventType { EventType::WebDav(WebDavEvent::Head) => 574, EventType::WebDav(WebDavEvent::Mkcalendar) => 575, EventType::Calendar(CalendarEvent::RuleExpansionError) => 576, - EventType::Store(StoreEvent::NatsError) => 577, EventType::Store(StoreEvent::CacheMiss) => 50, EventType::Store(StoreEvent::CacheHit) => 51, EventType::Store(StoreEvent::CacheStale) => 52, - EventType::Store(StoreEvent::CacheUpdate) => 578, + EventType::Store(StoreEvent::CacheUpdate) => 577, } } @@ -1507,11 +1506,10 @@ impl EventType { 574 => Some(EventType::WebDav(WebDavEvent::Head)), 575 => Some(EventType::WebDav(WebDavEvent::Mkcalendar)), 576 => Some(EventType::Calendar(CalendarEvent::RuleExpansionError)), - 577 => Some(EventType::Store(StoreEvent::NatsError)), 50 => Some(EventType::Store(StoreEvent::CacheMiss)), 51 => Some(EventType::Store(StoreEvent::CacheHit)), 52 => Some(EventType::Store(StoreEvent::CacheStale)), - 578 => Some(EventType::Store(StoreEvent::CacheUpdate)), + 577 => Some(EventType::Store(StoreEvent::CacheUpdate)), _ => None, } }