fix: increase cache size to hopefully deduplicate events (#865)

* fix: presence key was incorrect after other related update

Signed-off-by: IAmTomahawkx <iamtomahawkx@gmail.com>

* rust analyzer try not to be annoying challenge

Signed-off-by: IAmTomahawkx <iamtomahawkx@gmail.com>

* fix: increase seen events cache and reenable typing events when disabling overload events

Signed-off-by: IAmTomahawkx <iamtomahawkx@gmail.com>

* add canary publisher and ability to lower job count to not OOM

Signed-off-by: IAmTomahawkx <iamtomahawkx@gmail.com>

---------

Signed-off-by: IAmTomahawkx <iamtomahawkx@gmail.com>

.mise/tasks/docker/publish-canary

@@ -0,0 +1,59 @@
+#!/usr/bin/env bash
+#MISE description="Build the API image and push it to the canary registry. Use CARGO_BUILD_JOBS= to lower the job count if this OOMs."
+set -e
+
+IMAGE="registry.stoatinternal.com/stoat/backend-canary"
+TAG=latest
+PLATFORM=linux/amd64
+
+# echo "Building base image for ${PLATFORM}..."
+# docker buildx build --platform "${PLATFORM}" --load \
+#   -t ghcr.io/stoatchat/base:latest -f Dockerfile \
+#   --build-arg CARGO_BUILD_JOBS="${CARGO_BUILD_JOBS:-10}" .
+
+echo "Building API image for ${PLATFORM}..."
+docker buildx build --platform "${PLATFORM}" --load \
+  -t "${IMAGE}:${TAG}-delta" - < crates/delta/Dockerfile
+
+echo "Building Bonfire image for ${PLATFORM}..."
+docker buildx build --platform "${PLATFORM}" --load \
+  -t "${IMAGE}:${TAG}-bonfire" - < crates/bonfire/Dockerfile
+
+echo "Building Autumn image for ${PLATFORM}..."
+docker buildx build --platform "${PLATFORM}" --load \
+  -t "${IMAGE}:${TAG}-autumn" - < crates/services/autumn/Dockerfile
+
+echo "Building Gifbox image for ${PLATFORM}..."
+docker buildx build --platform "${PLATFORM}" --load \
+  -t "${IMAGE}:${TAG}-gifbox" - < crates/services/gifbox/Dockerfile
+
+echo "Building January image for ${PLATFORM}..."
+docker buildx build --platform "${PLATFORM}" --load \
+  -t "${IMAGE}:${TAG}-january" - < crates/services/january/Dockerfile
+
+echo "Building Crond image for ${PLATFORM}..."
+docker buildx build --platform "${PLATFORM}" --load \
+  -t "${IMAGE}:${TAG}-crond" - < crates/daemons/crond/Dockerfile
+
+echo "Building Pushd image for ${PLATFORM}..."
+docker buildx build --platform "${PLATFORM}" --load \
+  -t "${IMAGE}:${TAG}-pushd" - < crates/daemons/pushd/Dockerfile
+
+echo "Building Voice-ingress image for ${PLATFORM}..."
+docker buildx build --platform "${PLATFORM}" --load \
+  -t "${IMAGE}:${TAG}-voice-ingress" - < crates/daemons/voice-ingress/Dockerfile
+
+echo "Logging in to registry.stoatinternal.com..."
+docker login registry.stoatinternal.com \
+  --username "$(op read "op://Employee/canary-robot/username")" \
+  --password-stdin < <(op read "op://Employee/canary-robot/credential")
+
+echo "Pushing ${IMAGE}:${TAG}..."
+docker push "${IMAGE}:${TAG}-delta"
+docker push "${IMAGE}:${TAG}-bonfire"
+docker push "${IMAGE}:${TAG}-autumn"
+docker push "${IMAGE}:${TAG}-gifbox"
+docker push "${IMAGE}:${TAG}-january"
+docker push "${IMAGE}:${TAG}-crond"
+docker push "${IMAGE}:${TAG}-pushd"
+docker push "${IMAGE}:${TAG}-voice-ingress"

Dockerfile

@@ -5,6 +5,9 @@ WORKDIR /home/rust/src
 
 ARG TARGETARCH
 
+ARG CARGO_BUILD_JOBS=10
+ENV CARGO_BUILD_JOBS=${CARGO_BUILD_JOBS}
+
 # Install build requirements
 RUN dpkg --add-architecture "${TARGETARCH}"
 RUN apt-get update && \

Dockerfile.useCurrentArch

@@ -3,6 +3,9 @@ FROM rust:1.92.0-slim-bookworm
 USER 0:0
 WORKDIR /home/rust/src
 
+ARG CARGO_BUILD_JOBS=10
+ENV CARGO_BUILD_JOBS=${CARGO_BUILD_JOBS}
+
 # Install build requirements
 RUN apt-get update && \
     apt-get install -y \

crates/bonfire/src/events/state.rs

@@ -1,11 +1,14 @@
 use std::{
-    collections::{HashMap, HashSet}, num::NonZeroUsize, sync::Arc, time::Duration
+    collections::{HashMap, HashSet},
+    num::NonZeroUsize,
+    sync::Arc,
+    time::Duration,
 };
 
-use tokio::sync::{Mutex, RwLock};
 use lru::LruCache;
 use lru_time_cache::{LruCache as LruTimeCache, TimedEntry};
 use revolt_database::{Channel, Member, Server, User};
+use tokio::sync::{Mutex, RwLock};
 
 /// Enumeration representing some change in subscriptions
 pub enum SubscriptionStateChange {
@@ -44,6 +47,16 @@ pub struct Cache {
     pub seen_events: LruCache<String, ()>,
 }
 
+impl Cache {
+    fn with_events_size(user_id: Option<String>, seen_events_size: NonZeroUsize) -> Self {
+        Self {
+            user_id: user_id.unwrap_or(String::default()),
+            seen_events: LruCache::new(seen_events_size),
+            ..Default::default()
+        }
+    }
+}
+
 impl Default for Cache {
     fn default() -> Self {
         Cache {
@@ -55,7 +68,7 @@ impl Default for Cache {
             members: Default::default(),
             servers: Default::default(),
 
-            seen_events: LruCache::new(NonZeroUsize::new(20).unwrap()),
+            seen_events: LruCache::new(NonZeroUsize::new(2048).unwrap()),
         }
     }
 }
@@ -74,16 +87,13 @@ pub struct State {
 
 impl State {
     /// Create state from User
-    pub fn from(user: User, session_id: String) -> State {
+    pub fn from(user: User, session_id: String, cache_size: NonZeroUsize) -> State {
         let mut subscribed = HashSet::new();
         let private_topic = format!("{}!", user.id);
         subscribed.insert(private_topic.clone());
         subscribed.insert(user.id.clone());
 
-        let mut cache: Cache = Cache {
+        let mut cache: Cache = Cache::with_events_size(Some(user.id.clone()), cache_size);
-            user_id: user.id.clone(),
-            ..Default::default()
-        };
 
         cache.users.insert(user.id.clone(), user);
 

crates/bonfire/src/websocket.rs

@@ -1,4 +1,4 @@
-use std::{collections::HashSet, net::SocketAddr, sync::Arc};
+use std::{collections::HashSet, net::SocketAddr, num::NonZeroUsize, sync::Arc};
 
 use async_tungstenite::WebSocketStream;
 use fred::{
@@ -21,21 +21,22 @@ use revolt_database::{
 };
 use revolt_presence::{create_session, delete_session};
 
+use revolt_result::create_error;
+use sentry::Level;
 use tokio::{
     net::TcpStream,
     sync::{Mutex, RwLock},
     task::spawn,
 };
-use tokio_util::compat::{TokioAsyncReadCompatExt, Compat};
+use tokio_util::compat::{Compat, TokioAsyncReadCompatExt};
-use revolt_result::create_error;
-use sentry::Level;
 
 use crate::config::{ProtocolConfiguration, WebsocketHandshakeCallback};
 use crate::events::state::{State, SubscriptionStateChange};
 use revolt_models::v0;
 
 type WsReader = SplitStream<WebSocketStream<Compat<TcpStream>>>;
-type WsWriter = SplitSink<WebSocketStream<Compat<TcpStream>>, async_tungstenite::tungstenite::Message>;
+type WsWriter =
+    SplitSink<WebSocketStream<Compat<TcpStream>>, async_tungstenite::tungstenite::Message>;
 
 /// Start a new WebSocket client worker given access to the database,
 /// the relevant TCP stream and the remote address of the client.
@@ -106,8 +107,15 @@ pub async fn client(db: &'static Database, stream: TcpStream, addr: SocketAddr)
         .await
         .ok();
 
+    let backend_config = revolt_config::config().await;
+
     // Create local state.
-    let mut state = State::from(user, session_id);
+    let mut state = State::from(
+        user,
+        session_id,
+        NonZeroUsize::new(backend_config.features.advanced.seen_events_cache_size as usize)
+            .expect("config.features.advanced.seen_events_cache_size cannot be 0!"),
+    );
     let user_id = state.cache.user_id.clone();
 
     // Notify socket we have authenticated.
@@ -440,8 +448,6 @@ async fn worker(
     mut read: WsReader,
     write: &Mutex<WsWriter>,
 ) {
-    let revolt_config = revolt_config::config().await;
-
     loop {
         let t1 = read.try_next().fuse();
         let t2 = kill_signal_r.recv().fuse();
@@ -478,10 +484,6 @@ async fn worker(
 
                 match payload {
                     ClientMessage::BeginTyping { channel } => {
-                        if revolt_config.disable_events_dont_use {
-                            continue;
-                        }
-
                         if !subscribed.read().await.contains(&channel) {
                             continue;
                         }
@@ -494,10 +496,6 @@ async fn worker(
                         .await;
                     }
                     ClientMessage::EndTyping { channel } => {
-                        if revolt_config.disable_events_dont_use {
-                            continue;
-                        }
-
                         if !subscribed.read().await.contains(&channel) {
                             continue;
                         }

crates/core/config/Revolt.toml

@@ -333,6 +333,9 @@ emojis = 500_000
 # The max amount of messages the rabbitmq provider/db mention adder job will delay for before forcing handling of a channel.
 # default: 5
 process_message_delay_limit = 5
+# How many event ids to cache per bonfire connections.
+# Higher numbers result in more deduplication but more ram usage.
+seen_events_cache_size = 2048
 
 [features.legal_links]
 # URLs for legal documents

crates/core/config/src/lib.rs

@@ -429,12 +429,15 @@ pub struct LegalLinks {
 pub struct FeaturesAdvanced {
     #[serde(default)]
     pub process_message_delay_limit: u16,
+    #[serde(default)]
+    pub seen_events_cache_size: u32,
 }
 
 impl Default for FeaturesAdvanced {
     fn default() -> Self {
         Self {
             process_message_delay_limit: 5,
+            seen_events_cache_size: 20,
         }
     }
 }

scripts/build-image-layer.sh

@@ -1,4 +1,6 @@
 #!/bin/sh
+# If you're having trouble building this locally or on your CI, try lowering
+# the job count via CARGO_BUILD_JOBS. It defaults to 10.
 
 if [ -z "$TARGETARCH" ]; then
   :
@@ -62,9 +64,9 @@ deps() {
     tee crates/core/ratelimits/src/lib.rs
   
   if [ -z "$TARGETARCH" ]; then
-    cargo build -j 10 --locked --release
+    cargo build -j "${CARGO_BUILD_JOBS:-10}" --locked --release
   else
-    cargo build -j 10 --locked --release --target "${BUILD_TARGET}"
+    cargo build -j "${CARGO_BUILD_JOBS:-10}" --locked --release --target "${BUILD_TARGET}"
   fi
 }
 
@@ -86,9 +88,9 @@ apps() {
     crates/core/ratelimits/src/lib.rs
   
   if [ -z "$TARGETARCH" ]; then
-    cargo build -j 10 --locked --release
+    cargo build -j "${CARGO_BUILD_JOBS:-10}" --locked --release
   else
-    cargo build -j 10 --locked --release --target "${BUILD_TARGET}"
+    cargo build -j "${CARGO_BUILD_JOBS:-10}" --locked --release --target "${BUILD_TARGET}"
     mv target _target && mv _target/"${BUILD_TARGET}" target
   fi
 }