Fix

relay-server/benches/benches.rs

@@ -4,13 +4,15 @@ use relay_config::Config;
 use sqlx::sqlite::{SqliteConnectOptions, SqliteJournalMode, SqlitePoolOptions};
 use sqlx::{Pool, Sqlite};
 use std::path::PathBuf;
+use std::sync::Arc;
 use std::time::{Duration, Instant};
 use tempfile::TempDir;
 use tokio::runtime::Runtime;
 
 use relay_base_schema::project::ProjectKey;
 use relay_server::{
-    Envelope, EnvelopeStack, PolymorphicEnvelopeBuffer, SqliteEnvelopeStack, SqliteEnvelopeStore,
+    Envelope, EnvelopeStack, MemoryChecker, MemoryStat, PolymorphicEnvelopeBuffer,
+    SqliteEnvelopeStack, SqliteEnvelopeStore,
 };
 
 fn setup_db(path: &PathBuf) -> Pool<Sqlite> {
@@ -221,6 +223,17 @@ fn benchmark_envelope_buffer(c: &mut Criterion) {
     let num_projects = 100000;
     let envelopes_per_project = 10;
 
+    let config: Arc<Config> = Config::from_json_value(serde_json::json!({
+        "spool": {
+            "health": {
+                "max_memory_percent": 1.0
+            }
+        }
+    }))
+    .unwrap()
+    .into();
+    let memory_checker = MemoryChecker::new(MemoryStat::default(), config.clone());
+
     group.throughput(Throughput::Elements(
         num_projects * envelopes_per_project as u64,
     ));
@@ -245,7 +258,8 @@ fn benchmark_envelope_buffer(c: &mut Criterion) {
             },
             |envelopes| {
                 runtime.block_on(async {
-                    let mut buffer = PolymorphicEnvelopeBuffer::from_config(&Config::default());
+                    let mut buffer =
+                        PolymorphicEnvelopeBuffer::from_config(&config, memory_checker.clone());
                     for envelope in envelopes.into_iter() {
                         buffer.push(envelope).await.unwrap();
                     }
@@ -274,7 +288,8 @@ fn benchmark_envelope_buffer(c: &mut Criterion) {
             },
             |envelopes| {
                 runtime.block_on(async {
-                    let mut buffer = PolymorphicEnvelopeBuffer::from_config(&Config::default());
+                    let mut buffer =
+                        PolymorphicEnvelopeBuffer::from_config(&config, memory_checker.clone());
                     let n = envelopes.len();
                     for envelope in envelopes.into_iter() {
                         let public_key = envelope.meta().public_key();

relay-server/src/lib.rs

@@ -271,6 +271,7 @@ pub use self::services::buffer::{
     EnvelopeStack, PolymorphicEnvelopeBuffer, SqliteEnvelopeStack, SqliteEnvelopeStore,
 }; // pub for benchmarks
 pub use self::services::spooler::spool_utils;
+pub use self::utils::{MemoryChecker, MemoryStat}; // pub for benchmarks
 
 #[cfg(test)]
 mod testutils;

relay-server/src/services/buffer/envelope_buffer/mod.rs

@@ -270,7 +270,7 @@ where
     }
 
     pub fn has_capacity(&self) -> bool {
-        matches!(self.stack_provider.capacity(), Capacity::FREE)
+        matches!(self.stack_provider.capacity(), Capacity::Free)
     }
 
     fn pop_stack(&mut self, stack_key: StackKey) {

relay-server/src/services/buffer/envelope_store/mod.rs

@@ -28,6 +28,7 @@ pub trait EnvelopeStore {
         limit: i64,
     ) -> impl Future<Output = Result<Vec<Box<Envelope>>, Self::Error>>;
 
+    #[allow(dead_code)]
     /// Returns a set of project key pairs, representing all the unique combinations of
     /// `own_key` and `project_key` that are found in the store.
     fn project_key_pairs(

relay-server/src/services/buffer/stacks_manager/memory.rs

@@ -23,9 +23,9 @@ impl StacksManager for MemoryStacksManager {
 
     fn capacity(&self) -> Capacity {
         if self.memory_checker.check_memory().has_capacity() {
-            Capacity::FREE
+            Capacity::Free
         } else {
-            Capacity::FULL
+            Capacity::Full
         }
     }
 }

relay-server/src/services/buffer/stacks_manager/mod.rs

@@ -5,8 +5,8 @@ pub mod sqlite;
 
 /// Enum representing the current capacity of the [`StacksManager`] to accept new [`Envelope`]s.
 pub enum Capacity {
-    FREE,
-    FULL,
+    Free,
+    Full,
 }
 
 /// A provider of [`EnvelopeStack`] instances that is responsible for creating them.

relay-server/src/services/buffer/stacks_manager/sqlite.rs

@@ -3,6 +3,7 @@ use relay_config::Config;
 use crate::services::buffer::envelope_store::sqlite::{
     SqliteEnvelopeStore, SqliteEnvelopeStoreError,
 };
+use crate::services::buffer::envelope_store::EnvelopeStore;
 use crate::services::buffer::stacks_manager::{Capacity, StacksManager};
 use crate::{Envelope, SqliteEnvelopeStack};
 
@@ -45,8 +46,10 @@ impl StacksManager for SqliteStacksManager {
     }
 
     fn capacity(&self) -> Capacity {
-        // TODO: how to we make the check async or sync.
-        // self.envelope_store.usage()
-        Capacity::FREE
+        if self.envelope_store.usage() < self.max_disk_size {
+            Capacity::Free
+        } else {
+            Capacity::Full
+        }
     }
 }

relay-server/src/utils/memory.rs

@@ -177,7 +177,7 @@ impl MemoryCheck {
 /// decides how memory readings are interpreted.
 #[derive(Clone, Debug)]
 pub struct MemoryChecker {
-    pub memory_stat: MemoryStat,
+    memory_stat: MemoryStat,
     config: Arc<Config>,
 }