Add missing is_deleting() guards and reconcile stuck tombstones

Retention (get/put), hot tier (put/get/delete), and the distributed
get_stats handler could still act on a stream mid-deletion since they
never checked is_deleting() after resolving it. Add the same guard
used by the other stream endpoints to each of them.

Also extend sync_all_streams to periodically scan the tombstone
directory itself, not just resident streams: if a background deletion
finishes but the final tombstone-clear call fails, the stream is
already gone from memory and storage, so the existing per-stream
self-heal never sees it again. Retrying delete_stream against an
already-empty prefix is a cheap no-op, so this converges the tombstone
without waiting for a node restart.

Author: prabhaks

src/handlers/http/logstream.rs

@@ -303,10 +303,12 @@ pub async fn get_retention(
         return Err(StreamNotFound(stream_name.clone()).into());
     }
 
-    let retention = PARSEABLE
-        .get_stream(&stream_name, &tenant_id)?
-        .get_retention()
-        .unwrap_or_default();
+    let stream = PARSEABLE.get_stream(&stream_name, &tenant_id)?;
+    if stream.is_deleting() {
+        return Err(StreamNotFound(stream_name.clone()).into());
+    }
+
+    let retention = stream.get_retention().unwrap_or_default();
     Ok((web::Json(retention), StatusCode::OK))
 }
 
@@ -327,15 +329,18 @@ pub async fn put_retention(
         return Err(StreamNotFound(stream_name).into());
     }
 
+    let stream = PARSEABLE.get_stream(&stream_name, &tenant_id)?;
+    if stream.is_deleting() {
+        return Err(StreamNotFound(stream_name).into());
+    }
+
     PARSEABLE
         .storage
         .get_object_store()
         .put_retention(&stream_name, &retention, &tenant_id)
         .await?;
 
-    PARSEABLE
-        .get_stream(&stream_name, &tenant_id)?
-        .set_retention(retention);
+    stream.set_retention(retention);
 
     Ok((
         format!("set retention configuration for log stream {stream_name}"),
@@ -527,6 +532,10 @@ pub async fn put_stream_hot_tier(
 
     let stream = PARSEABLE.get_stream(&stream_name, &tenant_id)?;
 
+    if stream.is_deleting() {
+        return Err(StreamNotFound(stream_name).into());
+    }
+
     if stream.get_stream_type() == StreamType::Internal {
         return Err(StreamError::Custom {
             msg: "Hot tier can not be updated for internal stream".to_string(),
@@ -600,6 +609,13 @@ pub async fn get_stream_hot_tier(
         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 Some(hot_tier_manager) = GLOBAL_HOTTIER.get() else {
         return Err(StreamError::HotTierNotEnabled(stream_name));
     };
@@ -635,11 +651,13 @@ pub async fn delete_stream_hot_tier(
         return Err(StreamNotFound(stream_name).into());
     }
 
-    if PARSEABLE
-        .get_stream(&stream_name, &tenant_id)?
-        .get_stream_type()
-        == StreamType::Internal
-    {
+    let stream = PARSEABLE.get_stream(&stream_name, &tenant_id)?;
+
+    if stream.is_deleting() {
+        return Err(StreamNotFound(stream_name).into());
+    }
+
+    if stream.get_stream_type() == StreamType::Internal {
         return Err(StreamError::Custom {
             msg: "Hot tier can not be deleted for internal stream".to_string(),
             status: StatusCode::BAD_REQUEST,

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

@@ -250,6 +250,13 @@ 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_map = web::Query::<HashMap<String, String>>::from_query(req.query_string())
         .map_err(|_| StreamError::InvalidQueryParameter(STATS_DATE_QUERY_PARAM.to_string()))?;
 

src/storage/object_storage.rs

@@ -1501,6 +1501,44 @@ pub fn sync_all_streams(joinset: &mut JoinSet<Result<(), ObjectStorageError>>) {
                 handle,
             );
         }
+
+        // Reconcile tombstones left behind by a deletion whose bulk delete
+        // succeeded but whose final tombstone-clear call failed (e.g. a
+        // transient network error). That stream is already gone from both
+        // this node's memory and from storage by the time that happens, so
+        // the resident-stream loop above never sees it again -- only a scan
+        // of the tombstone directory itself can find it. Same
+        // is_deletion_owner restriction as above: only a node that can
+        // actually run the physical delete should retry it.
+        if PARSEABLE.options.mode != Mode::Ingest {
+            let object_store = object_store.clone();
+            let tenant_id = tenant_id.clone();
+            joinset.spawn_on(
+                async move {
+                    match list_tombstoned_streams(object_store.as_ref(), &tenant_id).await {
+                        Ok(stream_names) => {
+                            for stream_name in stream_names {
+                                // Still resident: already handled by the
+                                // per-stream loop above, don't double-spawn.
+                                if PARSEABLE.streams.contains(&stream_name, &tenant_id) {
+                                    continue;
+                                }
+                                // delete_stream against an already-empty
+                                // prefix is a cheap no-op, so this only ever
+                                // repeats the tombstone clear until it
+                                // succeeds -- safe to retry every interval.
+                                spawn_stream_deletion(stream_name, tenant_id.clone());
+                            }
+                        }
+                        Err(e) => error!(
+                            "failed to list tombstoned streams while syncing tenant {tenant_id:?}: {e}"
+                        ),
+                    }
+                    Ok(())
+                },
+                handle,
+            );
+        }
     }
 }