fix: poison lock, use parking_lot

parking_lot in
- Stream metadata
- Stream writer
- Streams struct

Author: nikhilsinhaparseable

src/event/mod.rs

@@ -28,7 +28,6 @@ use tracing::{info_span, instrument};
 
 use self::error::EventError;
 use crate::{
-    LOCK_EXPECT,
     handlers::TelemetryType,
     metadata::update_stats,
     metrics::{increment_events_ingested_by_date, increment_events_ingested_size_by_date},
@@ -157,7 +156,7 @@ pub fn commit_schema(
     tenant_id: &Option<String>,
 ) -> Result<(), StagingError> {
     let _span = info_span!("commit_schema", stream_name).entered();
-    let mut stream_metadata = PARSEABLE.streams.write().expect("lock poisoned");
+    let mut stream_metadata = PARSEABLE.streams.write();
     let tenant_id = tenant_id.as_deref().unwrap_or(DEFAULT_TENANT);
     let map = &mut stream_metadata
         .get_mut(tenant_id)
@@ -166,7 +165,6 @@ pub fn commit_schema(
         .ok_or_else(|| StreamNotFound(stream_name.to_string()))?
         .metadata
         .write()
-        .expect(LOCK_EXPECT)
         .schema;
     let current_schema = Schema::new(map.values().cloned().collect::<Fields>());
     let schema = Schema::try_merge(vec![current_schema, schema.as_ref().clone()])?;

src/handlers/http/logstream.rs

@@ -35,7 +35,7 @@ use crate::utils::get_tenant_id_from_request;
 use crate::utils::json::flatten::{
     self, convert_to_array, generic_flattening, has_more_than_max_allowed_levels,
 };
-use crate::{LOCK_EXPECT, stats, validator};
+use crate::{stats, validator};
 
 use actix_web::http::StatusCode;
 use actix_web::web::{Json, Path};
@@ -397,15 +397,14 @@ pub async fn get_stream_info(
     };
 
     let tenant_id = tenant_id.as_deref().unwrap_or(DEFAULT_TENANT);
-    let hash_map = PARSEABLE.streams.read().unwrap();
+    let hash_map = PARSEABLE.streams.read();
     let stream_meta = hash_map
         .get(tenant_id)
         .ok_or_else(|| TenantNotFound(tenant_id.to_owned()))?
         .get(&stream_name)
         .ok_or_else(|| StreamNotFound(stream_name.clone()))?
         .metadata
-        .read()
-        .expect(LOCK_EXPECT);
+        .read();
 
     let stream_info =
         StreamInfo::from_metadata(&stream_meta, stream_first_event_at, stream_latest_event_at);

src/handlers/http/modal/mod.rs

@@ -636,7 +636,7 @@ pub async fn initialize_hot_tier_metadata_on_startup(
 ) -> anyhow::Result<()> {
     // Collect hot tier configurations from streams before doing async operations
     let hot_tier_configs: Vec<(String, Option<String>, StreamHotTier)> = {
-        let tenants_guard = PARSEABLE.streams.read().unwrap();
+        let tenants_guard = PARSEABLE.streams.read();
         tenants_guard
             .iter()
             .flat_map(|(tenant_id, streams)| {

src/parseable/mod.rs

@@ -280,7 +280,6 @@ impl Parseable {
         let tenant_id = tenant_id.as_deref().unwrap_or(DEFAULT_TENANT);
         self.streams
             .read()
-            .unwrap()
             .get(tenant_id)
             .ok_or_else(|| StreamNotFound(format!("{stream_name} with tenant {tenant_id}")))
             .map(|v| v.get(stream_name))?
@@ -1219,9 +1218,7 @@ impl Parseable {
         write_user_groups().remove(tenant_id);
 
         // delete streams
-        if let Ok(mut streams) = PARSEABLE.streams.write() {
-            streams.remove(tenant_id);
-        }
+        PARSEABLE.streams.write().remove(tenant_id);
 
         // delete from in-mem
         if let Ok(mut tenants) = self.tenants.write() {

src/parseable/staging/mod.rs

@@ -17,8 +17,6 @@
  *
  */
 
-use std::sync::PoisonError;
-
 use crate::{parseable::StreamNotFound, tenants::TenantNotFound};
 
 pub mod reader;
@@ -38,6 +36,4 @@ pub enum StagingError {
     StreamNotFound(#[from] StreamNotFound),
     #[error("{0}")]
     TenantNotFound(#[from] TenantNotFound),
-    #[error("{0}")]
-    PoisonError(#[from] PoisonError<String>),
 }