From 44c35bfb519143f1726f7eba9636a87dbef04800 Mon Sep 17 00:00:00 2001 From: Tom Date: Tue, 21 Jul 2026 18:28:16 -0700 Subject: [PATCH] fix: increase cache size to hopefully deduplicate events (#865) * fix: presence key was incorrect after other related update Signed-off-by: IAmTomahawkx * rust analyzer try not to be annoying challenge Signed-off-by: IAmTomahawkx * fix: increase seen events cache and reenable typing events when disabling overload events Signed-off-by: IAmTomahawkx * add canary publisher and ability to lower job count to not OOM Signed-off-by: IAmTomahawkx --------- Signed-off-by: IAmTomahawkx --- .mise/tasks/docker/publish-canary | 59 ++++++++++++++++++++++++++++++ Dockerfile | 3 ++ Dockerfile.useCurrentArch | 3 ++ crates/bonfire/src/events/state.rs | 26 +++++++++---- crates/bonfire/src/websocket.rs | 30 +++++++-------- crates/core/config/Revolt.toml | 3 ++ crates/core/config/src/lib.rs | 3 ++ scripts/build-image-layer.sh | 10 +++-- 8 files changed, 109 insertions(+), 28 deletions(-) create mode 100755 .mise/tasks/docker/publish-canary diff --git a/.mise/tasks/docker/publish-canary b/.mise/tasks/docker/publish-canary new file mode 100755 index 00000000..26cb9065 --- /dev/null +++ b/.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" diff --git a/Dockerfile b/Dockerfile index aca233ec..ec61859b 100644 --- a/Dockerfile +++ b/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 && \ diff --git a/Dockerfile.useCurrentArch b/Dockerfile.useCurrentArch index 9bcce8ac..37ab8248 100644 --- a/Dockerfile.useCurrentArch +++ b/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 \ diff --git a/crates/bonfire/src/events/state.rs b/crates/bonfire/src/events/state.rs index 09bef49e..bebc0c3a 100644 --- a/crates/bonfire/src/events/state.rs +++ b/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, } +impl Cache { + fn with_events_size(user_id: Option, 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 { - user_id: user.id.clone(), - ..Default::default() - }; + let mut cache: Cache = Cache::with_events_size(Some(user.id.clone()), cache_size); cache.users.insert(user.id.clone(), user); diff --git a/crates/bonfire/src/websocket.rs b/crates/bonfire/src/websocket.rs index 172cf74a..426f9f27 100644 --- a/crates/bonfire/src/websocket.rs +++ b/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 revolt_result::create_error; -use sentry::Level; +use tokio_util::compat::{Compat, TokioAsyncReadCompatExt}; use crate::config::{ProtocolConfiguration, WebsocketHandshakeCallback}; use crate::events::state::{State, SubscriptionStateChange}; use revolt_models::v0; type WsReader = SplitStream>>; -type WsWriter = SplitSink>, async_tungstenite::tungstenite::Message>; +type WsWriter = + SplitSink>, 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, ) { - 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; } diff --git a/crates/core/config/Revolt.toml b/crates/core/config/Revolt.toml index 0861fd3e..15bc483c 100644 --- a/crates/core/config/Revolt.toml +++ b/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 diff --git a/crates/core/config/src/lib.rs b/crates/core/config/src/lib.rs index 41b5e3dd..6eccbf59 100644 --- a/crates/core/config/src/lib.rs +++ b/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, } } } diff --git a/scripts/build-image-layer.sh b/scripts/build-image-layer.sh index 52170fdd..26f8d54d 100644 --- a/scripts/build-image-layer.sh +++ b/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 }