feat(spooler): Add initialization of stacks (#3967)

CHANGELOG.md

@@ -22,6 +22,7 @@
 - Forward profiles of non-sampled transactions (with no options filtering). ([#3963](https://github.com/getsentry/relay/pull/3963))
 - Make EnvelopeBuffer a Service. ([#3965](https://github.com/getsentry/relay/pull/3965))
 - No longer send COGS data to dedicated Kafka topic. ([#3953](https://github.com/getsentry/relay/pull/3953))
+- Allow creation of `SqliteEnvelopeBuffer` from config, and load existing stacks from db on startup. ([#3967](https://github.com/getsentry/relay/pull/3967))
 
 ## 24.8.0
 

relay-server/benches/benches.rs

@@ -259,7 +259,9 @@ fn benchmark_envelope_buffer(c: &mut Criterion) {
             |envelopes| {
                 runtime.block_on(async {
                     let mut buffer =
-                        PolymorphicEnvelopeBuffer::from_config(&config, memory_checker.clone());
+                        PolymorphicEnvelopeBuffer::from_config(&config, memory_checker.clone())
+                            .await
+                            .unwrap();
                     for envelope in envelopes.into_iter() {
                         buffer.push(envelope).await.unwrap();
                     }
@@ -289,7 +291,9 @@ fn benchmark_envelope_buffer(c: &mut Criterion) {
             |envelopes| {
                 runtime.block_on(async {
                     let mut buffer =
-                        PolymorphicEnvelopeBuffer::from_config(&config, memory_checker.clone());
+                        PolymorphicEnvelopeBuffer::from_config(&config, memory_checker.clone())
+                            .await
+                            .unwrap();
                     let n = envelopes.len();
                     for envelope in envelopes.into_iter() {
                         let public_key = envelope.meta().public_key();

relay-server/src/service.rs

@@ -241,7 +241,7 @@ impl ServiceState {
         .spawn_handler(processor_rx);
 
         let envelope_buffer = EnvelopeBufferService::new(
-            &config,
+            config.clone(),
             MemoryChecker::new(memory_stat.clone(), config.clone()),
             project_cache.clone(),
         )

relay-server/src/services/buffer/common.rs

@@ -0,0 +1,33 @@
+use relay_base_schema::project::ProjectKey;
+
+use crate::Envelope;
+
+/// Struct that represents two project keys.
+#[derive(Debug, Clone, Copy, Eq, Hash, Ord, PartialOrd, PartialEq)]
+pub struct ProjectKeyPair {
+    pub own_key: ProjectKey,
+    pub sampling_key: ProjectKey,
+}
+
+impl ProjectKeyPair {
+    pub fn new(own_key: ProjectKey, sampling_key: ProjectKey) -> Self {
+        Self {
+            own_key,
+            sampling_key,
+        }
+    }
+
+    pub fn from_envelope(envelope: &Envelope) -> Self {
+        let own_key = envelope.meta().public_key();
+        let sampling_key = envelope.sampling_key().unwrap_or(own_key);
+        Self::new(own_key, sampling_key)
+    }
+
+    pub fn iter(&self) -> impl Iterator<Item = ProjectKey> {
+        let Self {
+            own_key,
+            sampling_key,
+        } = self;
+        std::iter::once(*own_key).chain((own_key != sampling_key).then_some(*sampling_key))
+    }
+}