Make stream deletion asynchronous, resumable across restarts

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

@@ -28,7 +28,10 @@ use crate::rbac::Users;
 use crate::rbac::role::Action;
 use crate::stats::{Stats, event_labels_date, storage_size_labels_date};
 use crate::storage::retention::Retention;
-use crate::storage::{ObjectStoreFormat, StreamInfo, StreamType};
+use crate::storage::{
+    ObjectStoreFormat, StreamInfo, StreamType,
+    object_storage::{spawn_stream_deletion, stream_json_path, to_bytes, tombstone_path},
+};
 use crate::tenants::TenantNotFound;
 use crate::utils::actix::extract_session_key_from_req;
 use crate::utils::get_tenant_id_from_request;
@@ -63,17 +66,56 @@ pub async fn delete(
         return Err(StreamNotFound(stream_name).into());
     }
 
+    // Fetched once, up front: every step below this point is either
+    // infallible or best-effort, so nothing after this line can bail out
+    // with "stream not found" partway through an already-durably-started
+    // deletion.
+    let stream = PARSEABLE.get_stream(&stream_name, &tenant_id)?;
+
+    // Flip the in-memory guard before any `.await` point: check_or_load_stream's
+    // resident-stream fast path doesn't itself consult is_tombstoned, so a
+    // concurrent request on this node could otherwise slip through in the
+    // window between the tombstone becoming durable and this flag being set.
+    stream.mark_deleting();
+
     let objectstore = PARSEABLE.storage.get_object_store();
 
-    // Delete from storage
-    objectstore.delete_stream(&stream_name, &tenant_id).await?;
+    // Durable marker first: if the process crashes anywhere after this
+    // point, restart-recovery resumes the deletion instead of silently
+    // leaving the stream half-deleted with no record of it.
+    objectstore
+        .put_object(
+            &tombstone_path(&stream_name, &tenant_id),
+            to_bytes(&()),
+            &tenant_id,
+        )
+        .await?;
+
+    // Best-effort: makes the stream vanish from listings almost
+    // immediately. Not fatal if it fails -- is_deleting()/is_tombstoned()
+    // checks already block reads and writes regardless of whether this file
+    // is gone yet.
+    if let Err(e) = objectstore
+        .delete_object(&stream_json_path(&stream_name, &tenant_id), &tenant_id)
+        .await
+    {
+        warn!(
+            "failed to eagerly delete stream.json for {stream_name}, will be removed with the rest of the prefix: {e}"
+        );
+    }
+
+    // Scheduled immediately once the stream is durably tombstoned and
+    // flagged locally, before any of the remaining best-effort steps --
+    // none of them are allowed to leave the deletion itself unscheduled if
+    // they fail.
+    spawn_stream_deletion(stream_name.clone(), tenant_id.clone());
+
     // Delete from staging
-    let stream_dir = PARSEABLE.get_or_create_stream(&stream_name, &tenant_id);
-    if let Err(err) = fs::remove_dir_all(&stream_dir.data_path) {
+    if let Err(err) = fs::remove_dir_all(&stream.data_path) {
         warn!(
             "failed to delete local data for stream {} with error {err}. Clean {} manually",
             stream_name,
-            stream_dir.data_path.to_string_lossy()
+            stream.data_path.to_string_lossy()
         )
     }
 
@@ -85,12 +127,10 @@ pub async fn delete(
             .await?;
     }
 
-    // Delete from memory
-    PARSEABLE.streams.delete(&stream_name, &tenant_id);
-    stats::delete_stats(&stream_name, "json", &tenant_id)
-        .unwrap_or_else(|e| warn!("failed to delete stats for stream {}: {:?}", stream_name, e));
-
-    Ok((format!("log stream {stream_name} deleted"), StatusCode::OK))
+    Ok((
+        format!("log stream {stream_name} deletion started"),
+        StatusCode::ACCEPTED,
+    ))
 }
 
 pub async fn list(req: HttpRequest) -> Result<impl Responder, StreamError> {
@@ -186,6 +226,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 +356,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 +427,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/ingest/ingestor_logstream.rs

@@ -31,7 +31,6 @@ use crate::{
     catalog::remove_manifest_from_snapshot,
     handlers::http::logstream::error::StreamError,
     parseable::{PARSEABLE, StreamNotFound},
-    stats,
     utils::get_tenant_id_from_request,
 };
 
@@ -78,6 +77,7 @@ pub async fn delete(
     let tenant_id = get_tenant_id_from_request(&req);
     // Delete from staging
     let stream_dir = PARSEABLE.get_stream(&stream_name, &tenant_id)?;
+    stream_dir.mark_deleting();
 
     // delete staging only for ingest server or standalone server
     // else skip
@@ -91,12 +91,16 @@ pub async fn delete(
         )
     }
 
-    // Delete from memory
-    PARSEABLE.streams.delete(&stream_name, &tenant_id);
-    stats::delete_stats(&stream_name, "json", &tenant_id)
-        .unwrap_or_else(|e| warn!("failed to delete stats for stream {}: {:?}", stream_name, e));
-
-    Ok((format!("log stream {stream_name} deleted"), StatusCode::OK))
+    // Not removed from memory here: this node doesn't run the background
+    // deletion job, so it doesn't know when the underlying prefix is
+    // actually gone. The entry is reaped once `sync_all_streams` notices
+    // the tombstone has cleared (see its is_deleting()/is_tombstoned()
+    // self-heal check) -- until then, `is_deleting()` keeps rejecting
+    // ingestion for this stream with a clear "being deleted" error.
+    Ok((
+        format!("log stream {stream_name} deletion started"),
+        StatusCode::OK,
+    ))
 }
 
 pub async fn put_stream(

src/handlers/http/modal/query/querier_logstream.rs

@@ -45,12 +45,14 @@ use crate::{
                 utils::{IngestionStats, QueriedStats, StorageStats, merge_queried_stats},
             },
             logstream::error::StreamError,
-            modal::{NodeMetadata, NodeType},
         },
     },
     parseable::{PARSEABLE, StreamNotFound},
     stats,
-    storage::{ObjectStoreFormat, StreamType},
+    storage::{
+        ObjectStoreFormat, StreamType,
+        object_storage::{spawn_stream_deletion, stream_json_path, to_bytes, tombstone_path},
+    },
     utils::get_tenant_id_from_request,
 };
 const STATS_DATE_QUERY_PARAM: &str = "date";
@@ -73,52 +75,82 @@ pub async fn delete(
         return Err(StreamNotFound(stream_name.clone()).into());
     }
 
+    // Fetched once, up front: every step below this point is either
+    // infallible or best-effort, so nothing after this line can bail out
+    // with "stream not found" partway through an already-durably-started
+    // deletion.
+    let stream = PARSEABLE.get_stream(&stream_name, &tenant_id)?;
+
+    // Flip the in-memory guard before any `.await` point: check_or_load_stream's
+    // resident-stream fast path doesn't itself consult is_tombstoned, so a
+    // concurrent request on this node could otherwise slip through in the
+    // window between the tombstone becoming durable and this flag being set.
+    stream.mark_deleting();
+
     let objectstore = PARSEABLE.storage.get_object_store();
-    // Delete from storage
-    objectstore.delete_stream(&stream_name, &tenant_id).await?;
-    let stream_dir = PARSEABLE.get_or_create_stream(&stream_name, &tenant_id);
-    if let Err(err) = fs::remove_dir_all(&stream_dir.data_path) {
-        warn!(
-            "failed to delete local data for stream {} with error {err}. Clean {} manually",
-            stream_name,
-            stream_dir.data_path.to_string_lossy()
+
+    // Durable marker first: if the process crashes anywhere after this
+    // point, restart-recovery resumes the deletion instead of silently
+    // leaving the stream half-deleted with no record of it.
+    objectstore
+        .put_object(
+            &tombstone_path(&stream_name, &tenant_id),
+            to_bytes(&()),
+            &tenant_id,
         )
-    }
+        .await?;
 
-    if let Some(hot_tier_manager) = GLOBAL_HOTTIER.get()
-        && hot_tier_manager.check_stream_hot_tier_exists(&stream_name, &tenant_id)
+    // Best-effort: makes the stream vanish from listings almost
+    // immediately, without touching every listing endpoint individually.
+    // Not fatal if it fails -- is_deleting()/is_tombstoned() checks already
+    // block reads and writes regardless of whether this file is gone yet.
+    if let Err(e) = objectstore
+        .delete_object(&stream_json_path(&stream_name, &tenant_id), &tenant_id)
+        .await
     {
-        hot_tier_manager
-            .delete_hot_tier(&stream_name, &tenant_id)
-            .await?;
+        warn!(
+            "failed to eagerly delete stream.json for {stream_name}, will be removed with the rest of the prefix: {e}"
+        );
     }
 
-    let ingestor_metadata: Vec<NodeMetadata> =
-        cluster::get_node_info(NodeType::Ingestor, &tenant_id)
-            .await
-            .map_err(|err| {
-                error!("Fatal: failed to get ingestor info: {:?}", err);
-                err
-            })?;
+    // Scheduled immediately once the stream is durably tombstoned and
+    // flagged locally, before any of the remaining best-effort steps --
+    // none of them are allowed to leave the deletion itself unscheduled if
+    // they fail.
+    spawn_stream_deletion(stream_name.clone(), tenant_id.clone());
 
-    for ingestor in ingestor_metadata {
+    let fanout_stream_name = stream_name.clone();
+    cluster::for_each_live_node(&tenant_id, move |node| {
         let url = format!(
             "{}{}/logstream/{}/sync",
-            ingestor.domain_name,
+            node.domain_name,
             base_path_without_preceding_slash(),
-            stream_name
+            fanout_stream_name
         );
+        async move { cluster::send_stream_delete_request(&url, node).await }
+    })
+    .await?;
 
-        // delete the stream
-        cluster::send_stream_delete_request(&url, ingestor.clone()).await?;
+    if let Err(err) = fs::remove_dir_all(&stream.data_path) {
+        warn!(
+            "failed to delete local data for stream {} with error {err}. Clean {} manually",
+            stream_name,
+            stream.data_path.to_string_lossy()
+        )
     }
 
-    // Delete from memory
-    PARSEABLE.streams.delete(&stream_name, &tenant_id);
-    stats::delete_stats(&stream_name, "json", &tenant_id)
-        .unwrap_or_else(|e| warn!("failed to delete stats for stream {}: {:?}", stream_name, e));
+    if let Some(hot_tier_manager) = GLOBAL_HOTTIER.get()
+        && hot_tier_manager.check_stream_hot_tier_exists(&stream_name, &tenant_id)
+    {
+        hot_tier_manager
+            .delete_hot_tier(&stream_name, &tenant_id)
+            .await?;
+    }
 
-    Ok((format!("log stream {stream_name} deleted"), StatusCode::OK))
+    Ok((
+        format!("log stream {stream_name} deletion started"),
+        StatusCode::ACCEPTED,
+    ))
 }
 
 pub async fn put_stream(

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 {