fix: dataloss due to contention at stream creation

src/parseable/mod.rs

@@ -171,7 +171,7 @@ impl Parseable {
         }
 
         // Gets write privileges only for creating the stream when it doesn't already exist.
-        self.streams.create(
+        self.streams.get_or_create(
             self.options.clone(),
             stream_name.to_owned(),
             LogStreamMetadata::default(),
@@ -342,7 +342,7 @@ impl Parseable {
             schema_version,
             log_source,
         );
-        self.streams.create(
+        self.streams.get_or_create(
             self.options.clone(),
             stream_name.to_string(),
             metadata,
@@ -652,7 +652,7 @@ impl Parseable {
                     SchemaVersion::V1, // New stream
                     log_source,
                 );
-                self.streams.create(
+                self.streams.get_or_create(
                     self.options.clone(),
                     stream_name.to_string(),
                     metadata,

src/parseable/streams.rs

@@ -737,17 +737,22 @@ pub struct Streams(RwLock<HashMap<String, StreamRef>>);
 // 4. When first event is sent to stream (update the schema)
 // 5. When set alert API is called (update the alert)
 impl Streams {
-    pub fn create(
+    /// Checks after getting an excluse lock whether already stream exists, else creates it.
+    /// NOTE: This is done to ensure we don't have contention among threads.
+    pub fn get_or_create(
         &self,
         options: Arc<Options>,
         stream_name: String,
         metadata: LogStreamMetadata,
         ingestor_id: Option<String>,
     ) -> StreamRef {
+        let mut guard = self.write().expect(LOCK_EXPECT);
+        if let Some(stream) = guard.get(&stream_name) {
+            return stream.clone();
+        }
+
         let stream = Stream::new(options, &stream_name, metadata, ingestor_id);
-        self.write()
-            .expect(LOCK_EXPECT)
-            .insert(stream_name, stream.clone());
+        guard.insert(stream_name, stream.clone());
 
         stream
     }