feat(spooler): Implement eviction algorithm for EnvelopeBuffer

relay-config/src/config.rs

@@ -860,10 +860,16 @@ fn spool_envelopes_stack_max_batches() -> usize {
     2
 }
 
+/// Default maximum time between receiving the envelope and processing it.
 fn spool_envelopes_max_envelope_delay_secs() -> u64 {
     24 * 60 * 60
 }
 
+/// Default maximum number of envelopes that can be evicted for each stack.
+fn spool_envelopes_max_evictable_envelopes() -> usize {
+    100
+}
+
 /// Persistent buffering configuration for incoming envelopes.
 #[derive(Debug, Serialize, Deserialize)]
 pub struct EnvelopeSpool {
@@ -903,6 +909,9 @@ pub struct EnvelopeSpool {
     /// they are dropped. Defaults to 24h.
     #[serde(default = "spool_envelopes_max_envelope_delay_secs")]
     max_envelope_delay_secs: u64,
+    /// Maximum number of envelopes that can be evicted for each stack.
+    #[serde(default = "spool_envelopes_max_evictable_envelopes")]
+    max_evictable_envelopes: usize,
     /// Version of the spooler.
     #[serde(default)]
     version: EnvelopeSpoolVersion,
@@ -938,6 +947,7 @@ impl Default for EnvelopeSpool {
             disk_batch_size: spool_envelopes_stack_disk_batch_size(),
             max_batches: spool_envelopes_stack_max_batches(),
             max_envelope_delay_secs: spool_envelopes_max_envelope_delay_secs(),
+            max_evictable_envelopes: spool_envelopes_max_evictable_envelopes(),
             version: EnvelopeSpoolVersion::default(),
         }
     }
@@ -2149,6 +2159,11 @@ impl Config {
         self.values.spool.envelopes.max_batches
     }
 
+    /// Maximum number of envelopes that can be evicted per stack.
+    pub fn spool_envelopes_stack_max_evictable_envelopes(&self) -> usize {
+        self.values.spool.envelopes.max_evictable_envelopes
+    }
+
     /// Returns `true` if version 2 of the spooling mechanism is used.
     pub fn spool_v2(&self) -> bool {
         matches!(

relay-server/benches/benches.rs

@@ -99,6 +99,7 @@ fn benchmark_sqlite_envelope_stack(c: &mut Criterion) {
                                 envelope_store.clone(),
                                 disk_batch_size,
                                 2,
+                                10,
                                 ProjectKey::parse("e12d836b15bb49d7bbf99e64295d995b").unwrap(),
                                 ProjectKey::parse("e12d836b15bb49d7bbf99e64295d995b").unwrap(),
                             );
@@ -135,6 +136,7 @@ fn benchmark_sqlite_envelope_stack(c: &mut Criterion) {
                                     envelope_store.clone(),
                                     disk_batch_size,
                                     2,
+                                    10,
                                     ProjectKey::parse("e12d836b15bb49d7bbf99e64295d995b").unwrap(),
                                     ProjectKey::parse("e12d836b15bb49d7bbf99e64295d995b").unwrap(),
                                 );
@@ -175,6 +177,7 @@ fn benchmark_sqlite_envelope_stack(c: &mut Criterion) {
                                 envelope_store.clone(),
                                 disk_batch_size,
                                 2,
+                                10,
                                 ProjectKey::parse("e12d836b15bb49d7bbf99e64295d995b").unwrap(),
                                 ProjectKey::parse("e12d836b15bb49d7bbf99e64295d995b").unwrap(),
                             );