Add tombstone/deleting safety net for stream deletion (#1763)

src/handlers/http/ingest.rs

@@ -553,6 +553,8 @@ pub enum PostError {
     MissingQueryParameter,
     #[error(transparent)]
     MetastoreError(#[from] MetastoreError),
+    #[error("Stream {0} is being deleted, please retry after some time")]
+    StreamBeingDeleted(String),
 }
 
 impl actix_web::ResponseError for PostError {
@@ -586,6 +588,8 @@ impl actix_web::ResponseError for PostError {
 
             StreamNotFound(_) => StatusCode::NOT_FOUND,
 
+            StreamBeingDeleted(_) => StatusCode::CONFLICT,
+
             MetastoreError(e) => e.status_code(),
         }
     }

src/handlers/http/logstream.rs

@@ -186,6 +186,9 @@ pub async fn get_schema(
     }
 
     let stream = PARSEABLE.get_stream(&stream_name, &tenant_id)?;
+    if stream.is_deleting() {
+        return Err(StreamNotFound(stream_name.clone()).into());
+    }
     match update_schema_when_distributed(&vec![stream_name.clone()], &tenant_id).await {
         Ok(_) => {
             let schema = stream.get_schema();
@@ -313,6 +316,12 @@ pub async fn get_stats(
     {
         return Err(StreamNotFound(stream_name.clone()).into());
     }
+    if PARSEABLE
+        .get_stream(&stream_name, &tenant_id)
+        .is_ok_and(|stream| stream.is_deleting())
+    {
+        return Err(StreamNotFound(stream_name.clone()).into());
+    }
 
     let query_string = req.query_string();
     if !query_string.is_empty() {
@@ -378,6 +387,12 @@ pub async fn get_stream_info(
     {
         return Err(StreamNotFound(stream_name.clone()).into());
     }
+    if PARSEABLE
+        .get_stream(&stream_name, &tenant_id)
+        .is_ok_and(|stream| stream.is_deleting())
+    {
+        return Err(StreamNotFound(stream_name.clone()).into());
+    }
 
     let storage = PARSEABLE.storage().get_object_store();
 

src/handlers/http/modal/utils/ingest_utils.rs

@@ -509,6 +509,10 @@ pub fn validate_stream_for_ingestion(
 ) -> Result<(), PostError> {
     let stream = PARSEABLE.get_stream(stream_name, tenant_id)?;
 
+    if stream.is_deleting() {
+        return Err(PostError::StreamBeingDeleted(stream_name.to_string()));
+    }
+
     // Validate that the stream's log source is compatible
     stream
         .get_log_source()

src/handlers/http/query.rs

@@ -559,6 +559,22 @@ pub async fn create_streams_for_distributed(
     streams: Vec<String>,
     tenant_id: &Option<String>,
 ) -> Result<(), QueryError> {
+    // A stream that's already resident in memory but flagged `deleting`
+    // must reject the query outright. Checked unconditionally, ahead of
+    // the mode gate below, since this function backs every query-side
+    // call site (ad-hoc queries, alerts, saved query context, traces),
+    // not just the querier's own reload path.
+    for stream_name in &streams {
+        if PARSEABLE.streams.contains(stream_name, tenant_id)
+            && let Ok(stream) = PARSEABLE.get_stream(stream_name, tenant_id)
+            && stream.is_deleting()
+        {
+            return Err(QueryError::StreamNotFound(StreamNotFound(
+                stream_name.clone(),
+            )));
+        }
+    }
+
     if PARSEABLE.options.mode != Mode::Query && PARSEABLE.options.mode != Mode::Prism {
         return Ok(());
     }

src/metadata.rs

@@ -99,6 +99,11 @@ pub struct LogStreamMetadata {
     pub dataset_tags: Vec<DatasetTag>,
     pub dataset_labels: Vec<String>,
     pub infer_timestamp: bool,
+    /// Transient, in-memory only — never persisted to `ObjectStoreFormat`.
+    /// Set once a deletion has been initiated for this stream so that
+    /// readers/writers reached via an already-resident `Arc<Stream>` reject
+    /// it instead of racing the background deletion.
+    pub deleting: bool,
 }
 
 impl Default for LogStreamMetadata {
@@ -121,6 +126,7 @@ impl Default for LogStreamMetadata {
             dataset_tags: Vec::new(),
             dataset_labels: Vec::new(),
             infer_timestamp: true,
+            deleting: false,
         }
     }
 }

src/metastore/metastores/object_store_metastore.rs

@@ -55,7 +55,7 @@ use crate::{
     storage::{
         ALERTS_ROOT_DIRECTORY, ObjectStorage, ObjectStorageError, PARSEABLE_ROOT_DIRECTORY,
         SETTINGS_ROOT_DIRECTORY, STREAM_METADATA_FILE_NAME, STREAM_ROOT_DIRECTORY,
-        TARGETS_ROOT_DIRECTORY,
+        TARGETS_ROOT_DIRECTORY, TOMBSTONE_ROOT_DIRECTORY,
         object_storage::{
             alert_json_path, alert_state_json_path, filter_path, manifest_path, mttr_json_path,
             outbound_http_policy_json_path, parseable_json_path, schema_path, stream_json_path,
@@ -1433,6 +1433,7 @@ impl Metastore for ObjectStoreMetastore {
                         && name != USERS_ROOT_DIR
                         && name != SETTINGS_ROOT_DIRECTORY
                         && name != ALERTS_ROOT_DIRECTORY
+                        && name != TOMBSTONE_ROOT_DIRECTORY
                 })
                 .collect::<Vec<_>>();
             for stream in streams {

src/migration/mod.rs

@@ -513,6 +513,7 @@ pub async fn setup_logstream_metadata(
         dataset_tags,
         dataset_labels,
         infer_timestamp,
+        deleting: false,
     };
 
     Ok(metadata)

src/parseable/mod.rs

@@ -77,7 +77,8 @@ use crate::{
     static_schema::{StaticSchema, convert_static_schema_to_arrow_schema},
     storage::{
         ObjectStorage, ObjectStorageError, ObjectStorageProvider, ObjectStoreFormat, Owner,
-        Permisssion, StorageMetadata, StreamType, put_remote_metadata,
+        Permisssion, StorageMetadata, StreamType, object_storage::is_tombstoned,
+        put_remote_metadata,
     },
     tenants::{Service, TENANT_METADATA},
     validator,
@@ -472,6 +473,11 @@ impl Parseable {
     ) -> Result<bool, StreamError> {
         // Proceed to create log stream if it doesn't exist
         let storage = self.storage.get_object_store();
+        // A deletion in progress (or left unfinished by a crashed node) must
+        // never be resurrected by a concurrent lazy reload.
+        if is_tombstoned(storage.as_ref(), stream_name, tenant_id).await? {
+            return Ok(false);
+        }
         let streams = PARSEABLE.metastore.list_streams(tenant_id).await?;
         if !streams.contains(stream_name) {
             return Ok(false);

src/parseable/streams.rs

@@ -1217,8 +1217,12 @@ impl Stream {
     }
 
     /// Stores the provided stream metadata in memory mapping
-    pub async fn set_metadata(&self, updated_metadata: LogStreamMetadata) {
-        *self.metadata.write().expect(LOCK_EXPECT) = updated_metadata;
+    pub async fn set_metadata(&self, mut updated_metadata: LogStreamMetadata) {
+        let mut metadata = self.metadata.write().expect(LOCK_EXPECT);
+        // mark_deleting() is documented as monotonic -- a reload racing a
+        // delete must not silently clear it back to false.
+        updated_metadata.deleting |= metadata.deleting;
+        *metadata = updated_metadata;
     }
 
     pub fn get_first_event(&self) -> Option<String> {
@@ -1352,6 +1356,17 @@ impl Stream {
         self.metadata.read().expect(LOCK_EXPECT).hot_tier_enabled
     }
 
+    /// Marks this stream as being deleted. Once set, this flag is never
+    /// cleared for this in-memory entry — a deletion in progress runs to
+    /// completion (or is resumed on restart), it is never cancelled.
+    pub fn mark_deleting(&self) {
+        self.metadata.write().expect(LOCK_EXPECT).deleting = true;
+    }
+
+    pub fn is_deleting(&self) -> bool {
+        self.metadata.read().expect(LOCK_EXPECT).deleting
+    }
+
     pub fn get_stream_type(&self) -> StreamType {
         self.metadata.read().expect(LOCK_EXPECT).stream_type
     }
@@ -1744,6 +1759,22 @@ mod tests {
         );
     }
 
+    #[test]
+    fn test_mark_deleting_sets_is_deleting() {
+        let options = Arc::new(Options::default());
+        let stream = Stream::new(
+            options,
+            "test_stream",
+            LogStreamMetadata::default(),
+            None,
+            &None,
+        );
+
+        assert!(!stream.is_deleting());
+        stream.mark_deleting();
+        assert!(stream.is_deleting());
+    }
+
     #[test]
     fn test_staging_with_special_characters() {
         let stream_name = "test_stream_!@#$%^&*()";

src/storage/localfs.rs

@@ -49,6 +49,7 @@ use crate::{
 use super::{
     ALERTS_ROOT_DIRECTORY, ObjectStorage, ObjectStorageError, ObjectStorageProvider,
     PARSEABLE_ROOT_DIRECTORY, STREAM_METADATA_FILE_NAME, STREAM_ROOT_DIRECTORY,
+    TOMBSTONE_ROOT_DIRECTORY,
 };
 
 #[derive(Debug, Clone, clap::Args)]
@@ -533,6 +534,7 @@ impl ObjectStorage for LocalFS {
             USERS_ROOT_DIR,
             ALERTS_ROOT_DIRECTORY,
             SETTINGS_ROOT_DIRECTORY,
+            TOMBSTONE_ROOT_DIRECTORY,
         ];
 
         let result = fs::read_dir(&self.root).await;
@@ -570,6 +572,7 @@ impl ObjectStorage for LocalFS {
             PARSEABLE_ROOT_DIRECTORY,
             ALERTS_ROOT_DIRECTORY,
             SETTINGS_ROOT_DIRECTORY,
+            TOMBSTONE_ROOT_DIRECTORY,
         ];
 
         let result = fs::read_dir(&self.root).await;

src/storage/mod.rs

@@ -310,6 +310,17 @@ pub const ALERTS_ROOT_DIRECTORY: &str = ".alerts";
 pub const SETTINGS_ROOT_DIRECTORY: &str = ".settings";
 pub const TARGETS_ROOT_DIRECTORY: &str = ".targets";
 pub const MANIFEST_FILE: &str = "manifest.json";
+// top-level registry of streams currently being deleted; kept outside every
+// stream's own prefix so a bulk prefix-delete can never sweep up a marker
+// that's supposed to survive it (see is_tombstoned/tombstone_path)
+pub const TOMBSTONE_ROOT_DIRECTORY: &str = ".tombstones";
+// the marker itself lives one level below `{tenant}/{stream_name}/`, not as
+// a leaf key directly named after the stream: list_dirs_relative on every
+// backend (S3/GCS/Azure via list-with-delimiter's common_prefixes, LocalFS
+// via read_dir + is_dir) only surfaces child *directories*, never leaf
+// objects, so a tombstone recorded as a bare `{stream_name}` key would be
+// invisible to the restart-recovery scan that discovers tombstoned streams
+pub const TOMBSTONE_MARKER_FILE_NAME: &str = ".tombstone";
 
 // max concurrent request allowed for datafusion object store, overridable per
 // backend with P_MAX_OBJECT_STORE_REQUESTS.