From 6443b183194e0329c40fc8fe9a9cac65aa66cb09 Mon Sep 17 00:00:00 2001 From: prabhaks Date: Tue, 25 Aug 2026 22:18:01 -0700 Subject: [PATCH 01/14] Add tombstone/deleting safety net for stream deletion (#1763) Lays groundwork for background stream deletion: a durable tombstone marker outside the deleted prefix, an in-memory `deleting` flag on resident streams, and guards in the reload/query/info-endpoint code paths that reject a stream once either is set. Purely additive, no behavior change to the current delete handlers, since nothing yet sets a tombstone or the flag. Prepares for the actual async-delete rewrite in a follow-up PR. --- src/handlers/http/ingest.rs | 4 ++ src/handlers/http/logstream.rs | 15 +++++ src/handlers/http/modal/utils/ingest_utils.rs | 4 ++ src/handlers/http/query.rs | 16 +++++ src/metadata.rs | 6 ++ .../metastores/object_store_metastore.rs | 3 +- src/migration/mod.rs | 1 + src/parseable/mod.rs | 8 ++- src/parseable/streams.rs | 27 ++++++++ src/storage/localfs.rs | 2 + src/storage/mod.rs | 4 ++ src/storage/object_storage.rs | 66 ++++++++++++++++++- 12 files changed, 153 insertions(+), 3 deletions(-) diff --git a/src/handlers/http/ingest.rs b/src/handlers/http/ingest.rs index a98e8ab3a..c61b81bae 100644 --- a/src/handlers/http/ingest.rs +++ b/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(), } } diff --git a/src/handlers/http/logstream.rs b/src/handlers/http/logstream.rs index f1c6e8f5b..ab6fd1a8b 100644 --- a/src/handlers/http/logstream.rs +++ b/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(); diff --git a/src/handlers/http/modal/utils/ingest_utils.rs b/src/handlers/http/modal/utils/ingest_utils.rs index d3fcba7ac..07c85395b 100644 --- a/src/handlers/http/modal/utils/ingest_utils.rs +++ b/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() diff --git a/src/handlers/http/query.rs b/src/handlers/http/query.rs index 34c8a7327..dd97bd08b 100644 --- a/src/handlers/http/query.rs +++ b/src/handlers/http/query.rs @@ -559,6 +559,22 @@ pub async fn create_streams_for_distributed( streams: Vec, tenant_id: &Option, ) -> 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(()); } diff --git a/src/metadata.rs b/src/metadata.rs index 983456447..3a4ec2e1c 100644 --- a/src/metadata.rs +++ b/src/metadata.rs @@ -99,6 +99,11 @@ pub struct LogStreamMetadata { pub dataset_tags: Vec, pub dataset_labels: Vec, 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` 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, } } } diff --git a/src/metastore/metastores/object_store_metastore.rs b/src/metastore/metastores/object_store_metastore.rs index a5cda811e..b6b446b97 100644 --- a/src/metastore/metastores/object_store_metastore.rs +++ b/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::>(); for stream in streams { diff --git a/src/migration/mod.rs b/src/migration/mod.rs index 989cdbbac..6821e0ab1 100644 --- a/src/migration/mod.rs +++ b/src/migration/mod.rs @@ -513,6 +513,7 @@ pub async fn setup_logstream_metadata( dataset_tags, dataset_labels, infer_timestamp, + deleting: false, }; Ok(metadata) diff --git a/src/parseable/mod.rs b/src/parseable/mod.rs index 70ba8208f..87d26f30e 100644 --- a/src/parseable/mod.rs +++ b/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 { // 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); diff --git a/src/parseable/streams.rs b/src/parseable/streams.rs index 2262f599c..69b6a0212 100644 --- a/src/parseable/streams.rs +++ b/src/parseable/streams.rs @@ -1352,6 +1352,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 +1755,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_!@#$%^&*()"; diff --git a/src/storage/localfs.rs b/src/storage/localfs.rs index bafd80404..791fe9cf5 100644 --- a/src/storage/localfs.rs +++ b/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; diff --git a/src/storage/mod.rs b/src/storage/mod.rs index 5c6a2e36c..4ead03fee 100644 --- a/src/storage/mod.rs +++ b/src/storage/mod.rs @@ -310,6 +310,10 @@ 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"; // max concurrent request allowed for datafusion object store, overridable per // backend with P_MAX_OBJECT_STORE_REQUESTS. diff --git a/src/storage/object_storage.rs b/src/storage/object_storage.rs index 38b2809be..7a37cb533 100644 --- a/src/storage/object_storage.rs +++ b/src/storage/object_storage.rs @@ -68,7 +68,8 @@ use ulid::Ulid; use super::{ ALERTS_ROOT_DIRECTORY, MANIFEST_FILE, ObjectStorageError, ObjectStoreFormat, PARSEABLE_METADATA_FILE_NAME, PARSEABLE_ROOT_DIRECTORY, SCHEMA_FILE_NAME, - STREAM_METADATA_FILE_NAME, STREAM_ROOT_DIRECTORY, retention::Retention, + STREAM_METADATA_FILE_NAME, STREAM_ROOT_DIRECTORY, TOMBSTONE_ROOT_DIRECTORY, + retention::Retention, }; /// Context for upload operations containing stream information @@ -1442,6 +1443,41 @@ pub fn stream_json_path(stream_name: &str, tenant_id: &Option) -> Relati } } +/// Path to a stream's deletion marker. Deliberately lives under +/// `TOMBSTONE_ROOT_DIRECTORY`, outside the `{tenant}/{stream_name}` prefix +/// that a bulk stream delete walks, so a mid-deletion crash can never lose +/// the marker before the deletion it records has actually finished. +#[inline(always)] +pub fn tombstone_path(stream_name: &str, tenant_id: &Option) -> RelativePathBuf { + let tenant = tenant_id.as_deref().unwrap_or(""); + RelativePathBuf::from_iter([TOMBSTONE_ROOT_DIRECTORY, tenant, stream_name]) +} + +/// Whether a stream has a deletion marker present, i.e. a deletion was +/// started (possibly by a node that has since crashed or restarted) and has +/// not yet completed. +pub async fn is_tombstoned( + storage: &(impl ObjectStorage + ?Sized), + stream_name: &str, + tenant_id: &Option, +) -> Result { + match storage + .head(&tombstone_path(stream_name, tenant_id), tenant_id) + .await + { + Ok(_) => Ok(true), + // NoSuchKey is the object-store backends' not-found; LocalFS instead + // surfaces a plain io::Error, so both must be treated as "absent" + // here (see ObjectStoreMetastore::is_missing_optional_dir for the + // same not-found reconciliation across backends). + Err(ObjectStorageError::NoSuchKey(_)) => Ok(false), + Err(ObjectStorageError::IoError(e)) if e.kind() == std::io::ErrorKind::NotFound => { + Ok(false) + } + Err(e) => Err(e), + } +} + /// if filter_id is an empty str it should not append it to the rel path #[inline(always)] pub fn filter_path( @@ -1567,6 +1603,34 @@ pub fn manifest_segment_matches(manifest_path_str: &str, file_name: &str) -> boo manifest_path_str.rsplit('/').next() == Some(file_name) } +#[cfg(test)] +mod tombstone_tests { + use super::{ObjectStorage, is_tombstoned, to_bytes, tombstone_path}; + use crate::storage::LocalFS; + use temp_dir::TempDir; + + #[tokio::test] + async fn no_marker_means_not_tombstoned() { + let dir = TempDir::new().unwrap(); + let storage = LocalFS::new(dir.path().to_path_buf()); + + assert!(!is_tombstoned(&storage, "test_stream", &None).await.unwrap()); + } + + #[tokio::test] + async fn marker_present_means_tombstoned() { + let dir = TempDir::new().unwrap(); + let storage = LocalFS::new(dir.path().to_path_buf()); + + storage + .put_object(&tombstone_path("test_stream", &None), to_bytes(&()), &None) + .await + .unwrap(); + + assert!(is_tombstoned(&storage, "test_stream", &None).await.unwrap()); + } +} + #[cfg(test)] mod manifest_ownership_tests { use super::manifest_segment_matches; From 690410a9099617caa9ff6e58ad2fd6df7d884bf4 Mon Sep 17 00:00:00 2001 From: prabhaks Date: Tue, 25 Aug 2026 22:34:58 -0700 Subject: [PATCH 02/14] Nest tombstone markers under a per-stream directory so they're discoverable list_dirs_relative only surfaces child directories on every backend (S3/GCS/Azure via list-with-delimiter's common_prefixes, LocalFS via read_dir + is_dir), never leaf objects. A tombstone stored as a bare key named after the stream was therefore invisible to any future scan that needs to discover tombstoned streams rather than check one known name at a time. Move the marker one level deeper, under a directory named after the stream, and add list_tombstoned_streams for that scan. --- src/storage/mod.rs | 7 ++++ src/storage/object_storage.rs | 61 ++++++++++++++++++++++++++++++++--- 2 files changed, 64 insertions(+), 4 deletions(-) diff --git a/src/storage/mod.rs b/src/storage/mod.rs index 4ead03fee..6271f14cf 100644 --- a/src/storage/mod.rs +++ b/src/storage/mod.rs @@ -314,6 +314,13 @@ pub const MANIFEST_FILE: &str = "manifest.json"; // 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. diff --git a/src/storage/object_storage.rs b/src/storage/object_storage.rs index 7a37cb533..d1fee87b3 100644 --- a/src/storage/object_storage.rs +++ b/src/storage/object_storage.rs @@ -68,8 +68,8 @@ use ulid::Ulid; use super::{ ALERTS_ROOT_DIRECTORY, MANIFEST_FILE, ObjectStorageError, ObjectStoreFormat, PARSEABLE_METADATA_FILE_NAME, PARSEABLE_ROOT_DIRECTORY, SCHEMA_FILE_NAME, - STREAM_METADATA_FILE_NAME, STREAM_ROOT_DIRECTORY, TOMBSTONE_ROOT_DIRECTORY, - retention::Retention, + STREAM_METADATA_FILE_NAME, STREAM_ROOT_DIRECTORY, TOMBSTONE_MARKER_FILE_NAME, + TOMBSTONE_ROOT_DIRECTORY, retention::Retention, }; /// Context for upload operations containing stream information @@ -1447,10 +1447,21 @@ pub fn stream_json_path(stream_name: &str, tenant_id: &Option) -> Relati /// `TOMBSTONE_ROOT_DIRECTORY`, outside the `{tenant}/{stream_name}` prefix /// that a bulk stream delete walks, so a mid-deletion crash can never lose /// the marker before the deletion it records has actually finished. +/// +/// The marker is nested one level under `{stream_name}/`, not stored as a +/// bare key named after the stream: `list_dirs_relative` (used to discover +/// tombstoned streams on restart) only surfaces child directories on every +/// backend, so `{stream_name}` must itself resolve to a directory for that +/// scan to find it. #[inline(always)] pub fn tombstone_path(stream_name: &str, tenant_id: &Option) -> RelativePathBuf { let tenant = tenant_id.as_deref().unwrap_or(""); - RelativePathBuf::from_iter([TOMBSTONE_ROOT_DIRECTORY, tenant, stream_name]) + RelativePathBuf::from_iter([ + TOMBSTONE_ROOT_DIRECTORY, + tenant, + stream_name, + TOMBSTONE_MARKER_FILE_NAME, + ]) } /// Whether a stream has a deletion marker present, i.e. a deletion was @@ -1478,6 +1489,21 @@ pub async fn is_tombstoned( } } +/// Stream names with a deletion marker for the given tenant, discovered by +/// listing `TOMBSTONE_ROOT_DIRECTORY` rather than checking one name at a +/// time. Used to resume deletions left unfinished by a crashed or restarted +/// node, since a tombstoned stream whose `.stream.json` is already gone +/// would otherwise never surface via `list_streams`. See `is_tombstoned` for +/// the equivalent single-name check. +pub async fn list_tombstoned_streams( + storage: &(impl ObjectStorage + ?Sized), + tenant_id: &Option, +) -> Result, ObjectStorageError> { + let tenant = tenant_id.as_deref().unwrap_or(""); + let root = RelativePathBuf::from_iter([TOMBSTONE_ROOT_DIRECTORY, tenant]); + storage.list_dirs_relative(&root, tenant_id).await +} + /// if filter_id is an empty str it should not append it to the rel path #[inline(always)] pub fn filter_path( @@ -1605,7 +1631,7 @@ pub fn manifest_segment_matches(manifest_path_str: &str, file_name: &str) -> boo #[cfg(test)] mod tombstone_tests { - use super::{ObjectStorage, is_tombstoned, to_bytes, tombstone_path}; + use super::{ObjectStorage, is_tombstoned, list_tombstoned_streams, to_bytes, tombstone_path}; use crate::storage::LocalFS; use temp_dir::TempDir; @@ -1629,6 +1655,33 @@ mod tombstone_tests { assert!(is_tombstoned(&storage, "test_stream", &None).await.unwrap()); } + + #[tokio::test] + async fn no_markers_means_empty_discovery_list() { + let dir = TempDir::new().unwrap(); + let storage = LocalFS::new(dir.path().to_path_buf()); + + assert!( + list_tombstoned_streams(&storage, &None) + .await + .unwrap() + .is_empty() + ); + } + + #[tokio::test] + async fn marker_present_means_discoverable_by_listing() { + let dir = TempDir::new().unwrap(); + let storage = LocalFS::new(dir.path().to_path_buf()); + + storage + .put_object(&tombstone_path("test_stream", &None), to_bytes(&()), &None) + .await + .unwrap(); + + let discovered = list_tombstoned_streams(&storage, &None).await.unwrap(); + assert_eq!(discovered, vec!["test_stream".to_string()]); + } } #[cfg(test)] From 4eafc27dfddb4b9162d65635cd1e84713ec8e385 Mon Sep 17 00:00:00 2001 From: prabhaks Date: Wed, 26 Aug 2026 07:11:17 -0700 Subject: [PATCH 03/14] Make stream deletion asynchronous, resumable across restarts (#1763) DELETE /logstream/{stream} now writes a tombstone, notifies ingestors, and returns 202 Accepted immediately instead of blocking on the full recursive object-store delete. The actual deletion runs in a deduplicated background task, resumes automatically if the node crashes or restarts mid-delete (via the tombstone left by PR #1768's safety net), and self-heals nodes that missed the live notification. Only the node that receives the original DELETE request ever runs the physical delete; ingestors flag the stream as deleting and wait for the tombstone to clear, so a large deletion doesn't get redundantly re-run by every node in the cluster. list_streams() on the local filesystem backend is also fixed to treat a stream mid-deletion as absent rather than erroring out the whole listing. --- src/handlers/http/logstream.rs | 59 ++++-- .../http/modal/ingest/ingestor_logstream.rs | 18 +- .../http/modal/query/querier_logstream.rs | 93 ++++++---- src/migration/mod.rs | 30 +++- src/parseable/mod.rs | 15 ++ src/storage/localfs.rs | 95 +++++++++- src/storage/object_storage.rs | 169 +++++++++++++++++- 7 files changed, 417 insertions(+), 62 deletions(-) diff --git a/src/handlers/http/logstream.rs b/src/handlers/http/logstream.rs index ab6fd1a8b..531259c56 100644 --- a/src/handlers/http/logstream.rs +++ b/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,51 @@ 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)?; + 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}" + ); + } + + stream.mark_deleting(); + // 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 +122,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 { diff --git a/src/handlers/http/modal/ingest/ingestor_logstream.rs b/src/handlers/http/modal/ingest/ingestor_logstream.rs index 9f7414baa..02b281813 100644 --- a/src/handlers/http/modal/ingest/ingestor_logstream.rs +++ b/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( diff --git a/src/handlers/http/modal/query/querier_logstream.rs b/src/handlers/http/modal/query/querier_logstream.rs index 2bb170104..69bd969f2 100644 --- a/src/handlers/http/modal/query/querier_logstream.rs +++ b/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,77 @@ 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)?; + 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 = - cluster::get_node_info(NodeType::Ingestor, &tenant_id) - .await - .map_err(|err| { - error!("Fatal: failed to get ingestor info: {:?}", err); - err - })?; + stream.mark_deleting(); + // 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( diff --git a/src/migration/mod.rs b/src/migration/mod.rs index 6821e0ab1..354c21782 100644 --- a/src/migration/mod.rs +++ b/src/migration/mod.rs @@ -35,7 +35,10 @@ use crate::{ metrics::fetch_stats_from_storage, option::Mode, parseable::{DEFAULT_TENANT, PARSEABLE, Parseable}, - storage::{ObjectStorage, ObjectStoreFormat, PARSEABLE_METADATA_FILE_NAME, StorageMetadata}, + storage::{ + ObjectStorage, ObjectStoreFormat, PARSEABLE_METADATA_FILE_NAME, StorageMetadata, + object_storage::{is_tombstoned, list_tombstoned_streams, spawn_stream_deletion}, + }, }; fn get_version(metadata: &serde_json::Value) -> Option<&str> { @@ -213,8 +216,12 @@ pub async fn run_migration(config: &Parseable) -> anyhow::Result<()> { let mut futures = Vec::new(); for tenant_id in tenants { - // Get all stream names - let stream_names = PARSEABLE.metastore.list_streams(&tenant_id).await?; + // Get all stream names, plus any stream whose `.stream.json` is + // already gone because it's mid-deletion -- `list_streams` alone + // would miss it, and `migration_stream` needs the chance to resume + // that deletion below if the process crashed before finishing it. + let mut stream_names = PARSEABLE.metastore.list_streams(&tenant_id).await?; + stream_names.extend(list_tombstoned_streams(storage.as_ref(), &tenant_id).await?); // Create futures for each stream migration let f = stream_names.into_iter().map(|stream_name| { @@ -267,6 +274,23 @@ async fn migration_stream( storage: &dyn ObjectStorage, tenant_id: &Option, ) -> anyhow::Result> { + if is_tombstoned(storage, stream, tenant_id).await? { + // Left mid-deletion by a node that crashed or restarted before the + // background job finished. Resume it here rather than treating the + // stream as a normal (possibly schema-less) migration candidate -- + // `create_schema_from_metastore` below can itself error out on a + // partially-swept schema file, which would otherwise turn "resume + // deletion" into "abort node startup". Only a query/standalone node + // ever owns this job (mirrors the same guard in + // `object_storage::sync_all_streams`'s self-heal check) -- an + // ingestor just skips the stream here and waits for the tombstone to + // clear. + if PARSEABLE.options.mode != Mode::Ingest { + spawn_stream_deletion(stream.to_string(), tenant_id.clone()); + } + return Ok(None); + } + let mut arrow_schema: Schema = Schema::empty(); let schema = storage diff --git a/src/parseable/mod.rs b/src/parseable/mod.rs index 87d26f30e..e55a5d863 100644 --- a/src/parseable/mod.rs +++ b/src/parseable/mod.rs @@ -789,6 +789,21 @@ impl Parseable { let stream_in_memory_dont_update = self.streams.contains(stream_name, tenant_id) && !update_stream_flag; + // A stream still resident with is_deleting()=true is functionally + // gone (reads/writes are already rejected elsewhere), but its entry + // isn't removed from memory until the background deletion job + // finishes -- surface that distinctly rather than telling the + // caller it "already exists", which reads as if nothing were wrong. + if stream_in_memory_dont_update + && let Ok(stream) = self.get_stream(stream_name, tenant_id) + && stream.is_deleting() + { + return Err(StreamError::Custom { + msg: format!("Logstream {stream_name} is being deleted, please retry shortly"), + status: StatusCode::CONFLICT, + }); + } + // check if stream in storage only if not in memory // for Parseable OSS, create_update_stream is called only from query node // for Parseable Enterprise, create_update_stream is called from prism node diff --git a/src/storage/localfs.rs b/src/storage/localfs.rs index 791fe9cf5..ef148b098 100644 --- a/src/storage/localfs.rs +++ b/src/storage/localfs.rs @@ -44,6 +44,7 @@ use crate::{ option::validation, parseable::{DEFAULT_TENANT, LogStream}, storage::SETTINGS_ROOT_DIRECTORY, + storage::object_storage::tombstone_path, }; use super::{ @@ -555,7 +556,7 @@ impl ObjectStorage for LocalFS { let entries: Vec = directories.try_collect().await?; let entries = entries .into_iter() - .map(|entry| dir_with_stream(entry, ignore_dir)); + .map(|entry| dir_with_stream(entry, ignore_dir, &self.root)); let logstream_dirs: Vec> = FuturesUnordered::from_iter(entries).try_collect().await?; @@ -862,6 +863,7 @@ async fn dir_with_old_stream( async fn dir_with_stream( entry: DirEntry, ignore_dirs: &[&str], + root: &Path, ) -> Result, ObjectStorageError> { let dir_name = entry .path() @@ -885,6 +887,14 @@ async fn dir_with_stream( if stream_json_path.exists() { Ok(Some(dir_name)) + } else if tombstone_path(&dir_name, &None).to_path(root).exists() { + // Mid-async-deletion: `.stream.json` is deleted eagerly by the + // DELETE handler well before the background job finishes + // physically clearing the rest of the prefix, so a directory + // without it is expected here, not corrupt -- don't fail the + // whole listing over a stream that's in the middle of being + // deleted. + Ok(None) } else { let err: Box = format!("found {}", entry.path().display()).into(); @@ -915,3 +925,86 @@ impl From for ObjectStorageError { ObjectStorageError::UnhandledError(Box::new(e)) } } + +#[cfg(test)] +mod list_streams_tombstone_tests { + use temp_dir::TempDir; + + use super::{LocalFS, ObjectStorage}; + use crate::storage::object_storage::{to_bytes, tombstone_path}; + use crate::storage::{STREAM_METADATA_FILE_NAME, STREAM_ROOT_DIRECTORY}; + use relative_path::RelativePathBuf; + + // Deliberately not using `object_storage::stream_json_path` here: it + // reads the global PARSEABLE.options.mode, which isn't initialized under + // `cargo test` and crashes the whole test binary. This replicates its + // non-Ingest-mode path (tenant/stream/.stream/.stream.json) directly. + fn stream_json_path_for_test(stream_name: &str) -> RelativePathBuf { + RelativePathBuf::from_iter([ + "", + stream_name, + STREAM_ROOT_DIRECTORY, + STREAM_METADATA_FILE_NAME, + ]) + } + + #[tokio::test] + async fn normal_stream_is_still_listed() { + let dir = TempDir::new().unwrap(); + let storage = LocalFS::new(dir.path().to_path_buf()); + + storage + .put_object(&stream_json_path_for_test("mystream"), to_bytes(&()), &None) + .await + .unwrap(); + + let listed = storage.list_streams().await.unwrap(); + assert!(listed.contains("mystream")); + } + + #[tokio::test] + async fn stream_mid_deletion_is_skipped_not_a_listing_error() { + let dir = TempDir::new().unwrap(); + let storage = LocalFS::new(dir.path().to_path_buf()); + + // Set up as the DELETE handler leaves it: stream.json already gone, + // tombstone marker present, rest of the directory (and its other + // files) still there because the background delete hasn't finished. + storage + .put_object( + &stream_json_path_for_test("deleting-stream"), + to_bytes(&()), + &None, + ) + .await + .unwrap(); + storage + .delete_object(&stream_json_path_for_test("deleting-stream"), &None) + .await + .unwrap(); + storage + .put_object( + &tombstone_path("deleting-stream", &None), + to_bytes(&()), + &None, + ) + .await + .unwrap(); + + let listed = storage.list_streams().await.unwrap(); + assert!(!listed.contains("deleting-stream")); + } + + #[tokio::test] + async fn genuinely_corrupt_directory_still_errors() { + let dir = TempDir::new().unwrap(); + let storage = LocalFS::new(dir.path().to_path_buf()); + + // A real directory with neither a stream.json nor a tombstone is + // still treated as unexpected/corrupt, not silently skipped -- + // the fix narrows the exception to the tombstoned case specifically. + std::fs::create_dir_all(dir.path().join("not-a-stream")).unwrap(); + + assert!(storage.list_streams().await.is_err()); + } +} diff --git a/src/storage/object_storage.rs b/src/storage/object_storage.rs index d1fee87b3..70384c4d4 100644 --- a/src/storage/object_storage.rs +++ b/src/storage/object_storage.rs @@ -30,6 +30,7 @@ use crate::metrics::{EVENTS_STORAGE_SIZE_DATE, LIFETIME_EVENTS_STORAGE_SIZE, STO use crate::option::Mode; use crate::parseable::DEFAULT_TENANT; use crate::parseable::{LogStream, PARSEABLE, Stream}; +use crate::stats; use crate::stats::FullStats; use crate::storage::SETTINGS_ROOT_DIRECTORY; use crate::storage::TARGETS_ROOT_DIRECTORY; @@ -41,13 +42,14 @@ use arrow_schema::Schema; use async_trait::async_trait; use bytes::Bytes; use chrono::{DateTime, Utc}; +use dashmap::DashMap; use dashmap::mapref::entry::Entry; use datafusion::{datasource::listing::ListingTableUrl, execution::runtime_env::RuntimeEnvBuilder}; use itertools::Itertools; use object_store::ListResult; use object_store::ObjectMeta; use object_store::buffered::BufReader; -use once_cell::sync::OnceCell; +use once_cell::sync::{Lazy, OnceCell}; use rayon::prelude::*; use relative_path::RelativePath; use relative_path::RelativePathBuf; @@ -1335,6 +1337,70 @@ fn stream_relative_path( } } +/// Dedupes concurrent background deletion jobs for the same (tenant, stream) +/// so a repeat DELETE request, or a restart-triggered resume racing a job +/// already spawned before the restart, can't run `delete_stream` twice +/// concurrently against the same prefix. Mirrors `ACTIVE_OBJECT_STORE_SYNC_FILES` +/// in `sync.rs`. Deliberately has no time-based expiry (unlike that sibling +/// map): a large stream's real delete can legitimately run for many minutes, +/// and an expiry short enough to be useful would risk starting a duplicate +/// job while the original is still healthily in progress. See +/// `StreamDeletionGuard` for how an entry is still guaranteed to be cleared. +pub static ACTIVE_STREAM_DELETIONS: Lazy, String), Instant>> = + Lazy::new(DashMap::new); + +/// RAII guard that removes a stream's entry from `ACTIVE_STREAM_DELETIONS` on +/// drop, including on an unexpected panic inside the deletion task -- so a +/// single bad run can't wedge all future retries for that stream by leaving +/// its dedup entry stuck forever. +struct StreamDeletionGuard(Option<(Option, String)>); + +impl Drop for StreamDeletionGuard { + fn drop(&mut self) { + if let Some(key) = self.0.take() { + ACTIVE_STREAM_DELETIONS.remove(&key); + } + } +} + +/// Deletes a stream's data in the background and clears its tombstone once +/// done, so the synchronous DELETE handler can respond before a TB-scale +/// prefix delete completes. Safe to call more than once for the same +/// stream: a job already in flight is skipped rather than duplicated. +pub fn spawn_stream_deletion(stream_name: String, tenant_id: Option) { + let key = (tenant_id, stream_name); + match ACTIVE_STREAM_DELETIONS.entry(key.clone()) { + Entry::Occupied(_) => return, + Entry::Vacant(entry) => { + entry.insert(Instant::now()); + } + } + tokio::spawn(async move { + let _guard = StreamDeletionGuard(Some(key.clone())); + let (tenant_id, stream_name) = &key; + let storage = PARSEABLE.storage.get_object_store(); + match storage.delete_stream(stream_name, tenant_id).await { + Ok(()) => { + if let Err(e) = storage + .delete_object(&tombstone_path(stream_name, tenant_id), tenant_id) + .await + { + warn!( + "background deletion of {stream_name} finished but failed to clear its tombstone: {e}" + ); + } + PARSEABLE.streams.delete(stream_name, tenant_id); + if let Err(e) = stats::delete_stats(stream_name, "json", tenant_id) { + warn!("failed to clear stats for deleted stream {stream_name}: {e:?}"); + } + } + Err(e) => error!( + "background deletion failed for {stream_name}: {e}. tombstone left in place, retried on next restart or repeat DELETE" + ), + } + }); +} + pub fn sync_all_streams(joinset: &mut JoinSet>) { let object_store = PARSEABLE.storage().get_object_store(); let tenants = if let Some(tenants) = PARSEABLE.list_tenants() { @@ -1345,11 +1411,62 @@ pub fn sync_all_streams(joinset: &mut JoinSet>) { let handle = FLUSH_AND_CONVERT_RUNTIME.handle(); for tenant_id in tenants { for stream_name in PARSEABLE.streams.list(&tenant_id) { - if let Ok(stream) = PARSEABLE.get_stream(&stream_name, &tenant_id) - && stream.parquet_files().is_empty() - && stream.schema_files().is_empty() - { - continue; + if let Ok(stream) = PARSEABLE.get_stream(&stream_name, &tenant_id) { + if stream.is_deleting() { + // Fast path (the DELETE handler's own `for_each_live_node` + // push) already reaches most nodes immediately. This is + // the fallback for one that missed it, e.g. a node that + // was down or partitioned at the time: bounded to at most + // one sync interval instead of running forever. + let object_store = object_store.clone(); + let tenant_id = tenant_id.clone(); + let stream_name = stream_name.clone(); + // Only a node type that can actually receive a client's + // original DELETE request (query/standalone) ever resumes + // the physical delete here. An ingestor only ever gets + // is_deleting()=true via the delete handler's own + // fan-out push, which never starts a job on the ingestor + // itself -- letting it also spawn one here would mean + // every ingestor independently runs a redundant, + // uncoordinated bulk delete against the same prefix for + // the entire (possibly long) duration of every deletion, + // not just the rare case of a genuinely missed + // notification. + let is_deletion_owner = PARSEABLE.options.mode != Mode::Ingest; + joinset.spawn_on( + async move { + match is_tombstoned(object_store.as_ref(), &stream_name, &tenant_id) + .await + { + Ok(true) => { + if is_deletion_owner { + spawn_stream_deletion(stream_name, tenant_id); + } + } + Ok(false) => { + // Deletion already finished elsewhere and + // the tombstone is gone, but this node's + // resident entry was never dropped -- e.g. + // an ingestor, which doesn't run the + // background deletion job itself. Reap it + // so a stream recreated under the same + // name doesn't inherit a stuck + // deleting=true state. + PARSEABLE.streams.delete(&stream_name, &tenant_id); + } + Err(e) => error!( + "failed to check tombstone status for {stream_name}: {e}" + ), + } + Ok(()) + }, + handle, + ); + continue; + } + if stream.parquet_files().is_empty() && stream.schema_files().is_empty() { + continue; + } } let object_store = object_store.clone(); let id = tenant_id.clone(); @@ -1684,6 +1801,46 @@ mod tombstone_tests { } } +#[cfg(test)] +mod stream_deletion_dedup_tests { + use super::ACTIVE_STREAM_DELETIONS; + + // `spawn_stream_deletion` itself isn't unit-testable here: it reads + // PARSEABLE.storage/PARSEABLE.streams, and the global PARSEABLE static + // isn't initialized under `cargo test`. This instead exercises the + // contains_key-then-insert guard directly against the same map the real + // function uses, since that guard is the actual dedup mechanism. + #[test] + fn duplicate_key_is_recognized_as_already_running() { + let key = (Some("tenant-a".to_string()), "stream-a".to_string()); + ACTIVE_STREAM_DELETIONS.remove(&key); + + assert!(!ACTIVE_STREAM_DELETIONS.contains_key(&key)); + ACTIVE_STREAM_DELETIONS.insert(key.clone(), std::time::Instant::now()); + assert!(ACTIVE_STREAM_DELETIONS.contains_key(&key)); + // A second caller sees the job as already in flight and would skip + // re-inserting -- this is the exact check spawn_stream_deletion + // makes before spawning its background task. + assert!(ACTIVE_STREAM_DELETIONS.contains_key(&key)); + + ACTIVE_STREAM_DELETIONS.remove(&key); + assert!(!ACTIVE_STREAM_DELETIONS.contains_key(&key)); + } + + #[test] + fn same_stream_name_different_tenant_is_a_distinct_key() { + let key_a = (Some("tenant-a".to_string()), "shared-name".to_string()); + let key_b = (Some("tenant-b".to_string()), "shared-name".to_string()); + ACTIVE_STREAM_DELETIONS.remove(&key_a); + ACTIVE_STREAM_DELETIONS.remove(&key_b); + + ACTIVE_STREAM_DELETIONS.insert(key_a.clone(), std::time::Instant::now()); + assert!(!ACTIVE_STREAM_DELETIONS.contains_key(&key_b)); + + ACTIVE_STREAM_DELETIONS.remove(&key_a); + } +} + #[cfg(test)] mod manifest_ownership_tests { use super::manifest_segment_matches; From 094389a85f330a0ca0aba58c96869aeee9c847f9 Mon Sep 17 00:00:00 2001 From: prabhaks Date: Wed, 26 Aug 2026 09:01:07 -0700 Subject: [PATCH 04/14] Address CodeRabbit review: preserve deleting flag, tighten tombstone discovery set_metadata replaced the whole LogStreamMetadata wholesale, so a reload racing a delete (e.g. a schema update landing after mark_deleting()) could silently clear the deleting flag back to false despite it being documented as monotonic. Now ORs it in instead of overwriting. list_tombstoned_streams trusted list_dirs_relative's raw directory listing as proof of a marker's existence, but a directory can exist under the tombstone root without the marker itself (e.g. an interrupted write). Each candidate is now re-verified with is_tombstoned before being reported. list_old_streams (unused elsewhere in this codebase, but kept consistent with list_streams) didn't exclude TOMBSTONE_ROOT_DIRECTORY, so dir_with_old_stream would treat it as a corrupt stream directory the same way list_streams did before the earlier fix. --- src/parseable/streams.rs | 8 +++++-- src/storage/localfs.rs | 1 + src/storage/object_storage.rs | 43 ++++++++++++++++++++++++++++++++++- 3 files changed, 49 insertions(+), 3 deletions(-) diff --git a/src/parseable/streams.rs b/src/parseable/streams.rs index 69b6a0212..a8592eccd 100644 --- a/src/parseable/streams.rs +++ b/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 { diff --git a/src/storage/localfs.rs b/src/storage/localfs.rs index 791fe9cf5..d66714023 100644 --- a/src/storage/localfs.rs +++ b/src/storage/localfs.rs @@ -572,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; diff --git a/src/storage/object_storage.rs b/src/storage/object_storage.rs index d1fee87b3..2229bd503 100644 --- a/src/storage/object_storage.rs +++ b/src/storage/object_storage.rs @@ -1495,13 +1495,27 @@ pub async fn is_tombstoned( /// node, since a tombstoned stream whose `.stream.json` is already gone /// would otherwise never surface via `list_streams`. See `is_tombstoned` for /// the equivalent single-name check. +/// +/// `list_dirs_relative` only proves a directory exists under the tombstone +/// root, not that it actually holds a `.tombstone` marker (e.g. a partial or +/// interrupted write could leave an empty one behind) -- each candidate is +/// re-checked with `is_tombstoned` before being returned, so a directory +/// without a real marker is silently skipped rather than misreported. pub async fn list_tombstoned_streams( storage: &(impl ObjectStorage + ?Sized), tenant_id: &Option, ) -> Result, ObjectStorageError> { let tenant = tenant_id.as_deref().unwrap_or(""); let root = RelativePathBuf::from_iter([TOMBSTONE_ROOT_DIRECTORY, tenant]); - storage.list_dirs_relative(&root, tenant_id).await + let candidates = storage.list_dirs_relative(&root, tenant_id).await?; + + let mut confirmed = Vec::with_capacity(candidates.len()); + for stream_name in candidates { + if is_tombstoned(storage, &stream_name, tenant_id).await? { + confirmed.push(stream_name); + } + } + Ok(confirmed) } /// if filter_id is an empty str it should not append it to the rel path @@ -1682,6 +1696,33 @@ mod tombstone_tests { let discovered = list_tombstoned_streams(&storage, &None).await.unwrap(); assert_eq!(discovered, vec!["test_stream".to_string()]); } + + #[tokio::test] + async fn directory_without_the_actual_marker_is_not_reported_as_tombstoned() { + let dir = TempDir::new().unwrap(); + let storage = LocalFS::new(dir.path().to_path_buf()); + + // A directory can exist under the tombstone root without ever + // containing the marker itself (e.g. an interrupted write) -- + // list_dirs_relative alone can't tell the difference, so + // list_tombstoned_streams must re-verify via is_tombstoned. + let sibling_path = tombstone_path("test_stream", &None) + .parent() + .unwrap() + .join("not-the-marker"); + storage + .put_object(&sibling_path, to_bytes(&()), &None) + .await + .unwrap(); + + assert!(!is_tombstoned(&storage, "test_stream", &None).await.unwrap()); + assert!( + list_tombstoned_streams(&storage, &None) + .await + .unwrap() + .is_empty() + ); + } } #[cfg(test)] From a3037b541a2c87b72451d85aba9e2a0cffdf19c3 Mon Sep 17 00:00:00 2001 From: prabhaks Date: Wed, 26 Aug 2026 09:05:22 -0700 Subject: [PATCH 05/14] Flip the deleting flag before the tombstone write, not after check_or_load_stream's resident-stream fast path doesn't itself check is_tombstoned (flagged in CodeRabbit's review of #1768), so a concurrent request on the same node could slip through in the window between the tombstone becoming durable and mark_deleting() actually running. Moving mark_deleting() before the tombstone write, with no await point in between, closes that window entirely for the initiating node. Cross-node propagation is still bounded by the existing fan-out push and self-heal, not synchronous -- that's an accepted, already-documented limitation of this design, not something this reorder attempts to fix. --- src/handlers/http/logstream.rs | 7 ++++++- src/handlers/http/modal/query/querier_logstream.rs | 7 ++++++- 2 files changed, 12 insertions(+), 2 deletions(-) diff --git a/src/handlers/http/logstream.rs b/src/handlers/http/logstream.rs index 531259c56..a48aee89a 100644 --- a/src/handlers/http/logstream.rs +++ b/src/handlers/http/logstream.rs @@ -72,6 +72,12 @@ pub async fn delete( // 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(); // Durable marker first: if the process crashes anywhere after this @@ -98,7 +104,6 @@ pub async fn delete( ); } - stream.mark_deleting(); // 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 diff --git a/src/handlers/http/modal/query/querier_logstream.rs b/src/handlers/http/modal/query/querier_logstream.rs index 69bd969f2..de6ae1f4b 100644 --- a/src/handlers/http/modal/query/querier_logstream.rs +++ b/src/handlers/http/modal/query/querier_logstream.rs @@ -81,6 +81,12 @@ pub async fn delete( // 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(); // Durable marker first: if the process crashes anywhere after this @@ -107,7 +113,6 @@ pub async fn delete( ); } - stream.mark_deleting(); // 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 From eba77acab8beac35a8af1f5464473675669a95f0 Mon Sep 17 00:00:00 2001 From: prabhaks Date: Wed, 16 Sep 2026 11:53:32 -0700 Subject: [PATCH 06/14] Address CodeRabbit review feedback on #1770, switch MinIO test image to Quay - clear_deleting() to roll back mark_deleting() when the tombstone write itself fails, so a transient storage error doesn't permanently strand a stream in "deleting" state - reorder delete handlers (standalone + query node) so spawn_stream_deletion runs last, after local/hot-tier cleanup, and make ingestor fan-out and hot-tier cleanup best-effort instead of bailing the request - reject create/update of a tombstoned-but-not-yet-purged stream even when it isn't resident in memory yet (create_update_stream) - treat a repeated LocalFS delete_stream as success when the prefix is already gone, matching S3/Azure/GCS's empty-prefix behavior - restrict the resident-entry reap in sync_all_streams's tombstone check to non-owner nodes, closing a race with the owner's own in-flight tombstone write - fix stale comment on the dedup test module to reflect the real entry()-based atomic guard Also switch the MinIO image in all four docker-compose test files from minio/minio to quay.io/minio/minio (same pinned release tag) -- minio/minio has been pulled from Docker Hub, which was failing CI at the image-pull step before tests even ran. --- ...r-compose-distributed-test-with-kafka.yaml | 2 +- docker-compose-distributed-test.yaml | 2 +- docker-compose-test-with-kafka.yaml | 2 +- docker-compose-test.yaml | 2 +- src/handlers/http/logstream.rs | 32 ++++++++++----- .../http/modal/query/querier_logstream.rs | 41 +++++++++++++------ src/parseable/mod.rs | 24 +++++++++++ src/parseable/streams.rs | 8 ++++ src/storage/localfs.rs | 9 ++++ src/storage/object_storage.rs | 14 ++++++- 10 files changed, 107 insertions(+), 29 deletions(-) diff --git a/docker-compose-distributed-test-with-kafka.yaml b/docker-compose-distributed-test-with-kafka.yaml index 373513225..d2d11e978 100644 --- a/docker-compose-distributed-test-with-kafka.yaml +++ b/docker-compose-distributed-test-with-kafka.yaml @@ -30,7 +30,7 @@ services: # minio minio: - image: minio/minio:RELEASE.2025-02-03T21-03-04Z + image: quay.io/minio/minio:RELEASE.2025-02-03T21-03-04Z entrypoint: - sh - -euc diff --git a/docker-compose-distributed-test.yaml b/docker-compose-distributed-test.yaml index 50de497a8..f93d15d4f 100644 --- a/docker-compose-distributed-test.yaml +++ b/docker-compose-distributed-test.yaml @@ -4,7 +4,7 @@ networks: services: # minio minio: - image: minio/minio:RELEASE.2025-02-03T21-03-04Z + image: quay.io/minio/minio:RELEASE.2025-02-03T21-03-04Z entrypoint: - sh - -euc diff --git a/docker-compose-test-with-kafka.yaml b/docker-compose-test-with-kafka.yaml index 24dd6872a..6e5ca1c92 100644 --- a/docker-compose-test-with-kafka.yaml +++ b/docker-compose-test-with-kafka.yaml @@ -3,7 +3,7 @@ networks: services: minio: - image: minio/minio:RELEASE.2025-02-03T21-03-04Z + image: quay.io/minio/minio:RELEASE.2025-02-03T21-03-04Z entrypoint: - sh - -euc diff --git a/docker-compose-test.yaml b/docker-compose-test.yaml index a9a42b8c9..0a7dafc2c 100644 --- a/docker-compose-test.yaml +++ b/docker-compose-test.yaml @@ -3,7 +3,7 @@ networks: services: minio: - image: minio/minio:RELEASE.2025-02-03T21-03-04Z + image: quay.io/minio/minio:RELEASE.2025-02-03T21-03-04Z entrypoint: - sh - -euc diff --git a/src/handlers/http/logstream.rs b/src/handlers/http/logstream.rs index a48aee89a..d53dc76bc 100644 --- a/src/handlers/http/logstream.rs +++ b/src/handlers/http/logstream.rs @@ -83,13 +83,19 @@ pub async fn delete( // 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 + if let Err(e) = objectstore .put_object( &tombstone_path(&stream_name, &tenant_id), to_bytes(&()), &tenant_id, ) - .await?; + .await + { + // Nothing durable happened -- undo the in-memory flag so the stream + // isn't left permanently blocked by a transient write failure. + stream.clear_deleting(); + return Err(e.into()); + } // Best-effort: makes the stream vanish from listings almost // immediately. Not fatal if it fails -- is_deleting()/is_tombstoned() @@ -104,12 +110,6 @@ pub async fn delete( ); } - // 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 if let Err(err) = fs::remove_dir_all(&stream.data_path) { warn!( @@ -119,14 +119,24 @@ pub async fn delete( ) } + // Best-effort: the tombstone is already durable and the stream is + // already flagged, so a hot-tier cleanup failure must not turn an + // already-accepted deletion into an error response. if let Some(hot_tier_manager) = GLOBAL_HOTTIER.get() && hot_tier_manager.check_stream_hot_tier_exists(&stream_name, &tenant_id) - { - hot_tier_manager + && let Err(e) = hot_tier_manager .delete_hot_tier(&stream_name, &tenant_id) - .await?; + .await + { + warn!("failed to delete hot tier for stream {stream_name}: {e}"); } + // Scheduled only once every other cleanup step above has run, so the + // background job (which clears the tombstone on completion) can't race + // ahead of them and let a stream recreated under this name get swept up + // by leftover local/hot-tier cleanup that's still in flight. + spawn_stream_deletion(stream_name.clone(), tenant_id.clone()); + Ok(( format!("log stream {stream_name} deletion started"), StatusCode::ACCEPTED, diff --git a/src/handlers/http/modal/query/querier_logstream.rs b/src/handlers/http/modal/query/querier_logstream.rs index de6ae1f4b..3ebc4ec65 100644 --- a/src/handlers/http/modal/query/querier_logstream.rs +++ b/src/handlers/http/modal/query/querier_logstream.rs @@ -92,13 +92,19 @@ pub async fn delete( // 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 + if let Err(e) = objectstore .put_object( &tombstone_path(&stream_name, &tenant_id), to_bytes(&()), &tenant_id, ) - .await?; + .await + { + // Nothing durable happened -- undo the in-memory flag so the stream + // isn't left permanently blocked by a transient write failure. + stream.clear_deleting(); + return Err(e.into()); + } // Best-effort: makes the stream vanish from listings almost // immediately, without touching every listing endpoint individually. @@ -113,14 +119,13 @@ pub async fn delete( ); } - // 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()); - + // Best-effort: the tombstone is already durable and the stream is + // already flagged, so a fan-out failure must not turn an + // already-accepted deletion into an error response -- an ingestor that + // misses this push self-heals via sync_all_streams within one sync + // interval regardless. let fanout_stream_name = stream_name.clone(); - cluster::for_each_live_node(&tenant_id, move |node| { + if let Err(e) = cluster::for_each_live_node(&tenant_id, move |node| { let url = format!( "{}{}/logstream/{}/sync", node.domain_name, @@ -129,7 +134,10 @@ pub async fn delete( ); async move { cluster::send_stream_delete_request(&url, node).await } }) - .await?; + .await + { + warn!("failed to notify all ingestors of deletion of {stream_name}: {e}"); + } if let Err(err) = fs::remove_dir_all(&stream.data_path) { warn!( @@ -141,12 +149,19 @@ pub async fn delete( if let Some(hot_tier_manager) = GLOBAL_HOTTIER.get() && hot_tier_manager.check_stream_hot_tier_exists(&stream_name, &tenant_id) - { - hot_tier_manager + && let Err(e) = hot_tier_manager .delete_hot_tier(&stream_name, &tenant_id) - .await?; + .await + { + warn!("failed to delete hot tier for stream {stream_name}: {e}"); } + // Scheduled only once every other cleanup step above has run, so the + // background job (which clears the tombstone on completion) can't race + // ahead of them and let a stream recreated under this name get swept up + // by leftover fan-out/local/hot-tier cleanup that's still in flight. + spawn_stream_deletion(stream_name.clone(), tenant_id.clone()); + Ok(( format!("log stream {stream_name} deletion started"), StatusCode::ACCEPTED, diff --git a/src/parseable/mod.rs b/src/parseable/mod.rs index bbc506d26..3dd151a5a 100644 --- a/src/parseable/mod.rs +++ b/src/parseable/mod.rs @@ -810,6 +810,30 @@ impl Parseable { }); } + // A tombstoned-but-not-yet-purged stream must be rejected the same + // way even when it isn't resident in memory on this node yet (e.g. a + // query node before restart-recovery has resumed it, or a node that + // never loaded it in the first place). create_stream_and_schema_from_storage + // below already checks this internally and returns Ok(false) for a + // tombstoned stream, but that's indistinguishable from "doesn't + // exist" to its caller -- without this explicit check, the client + // could recreate the name while the background deletion job is + // still sweeping its prefix, and that job would delete the freshly + // created data too. + if !stream_in_memory_dont_update + && is_tombstoned( + self.storage.get_object_store().as_ref(), + stream_name, + tenant_id, + ) + .await? + { + return Err(StreamError::Custom { + msg: format!("Logstream {stream_name} is being deleted, please retry shortly"), + status: StatusCode::CONFLICT, + }); + } + // check if stream in storage only if not in memory // for Parseable OSS, create_update_stream is called only from query node // for Parseable Enterprise, create_update_stream is called from prism node diff --git a/src/parseable/streams.rs b/src/parseable/streams.rs index 0e3801f5a..a0948b746 100644 --- a/src/parseable/streams.rs +++ b/src/parseable/streams.rs @@ -1632,6 +1632,14 @@ impl Stream { self.metadata.write().expect(LOCK_EXPECT).deleting = true; } + /// Undoes a `mark_deleting()` call whose tombstone write then failed. + /// Safe only because nothing durable was ever created -- unlike + /// `mark_deleting()`, this is not part of the monotonic contract, and + /// must never be called once the tombstone actually exists in storage. + pub fn clear_deleting(&self) { + self.metadata.write().expect(LOCK_EXPECT).deleting = false; + } + pub fn is_deleting(&self) -> bool { self.metadata.read().expect(LOCK_EXPECT).deleting } diff --git a/src/storage/localfs.rs b/src/storage/localfs.rs index 54f5e2300..78823cd2d 100644 --- a/src/storage/localfs.rs +++ b/src/storage/localfs.rs @@ -496,6 +496,15 @@ impl ObjectStorage for LocalFS { let tenant_str = tenant_id.as_deref().unwrap_or(DEFAULT_TENANT); let result = fs::remove_dir_all(path).await; + // A retried deletion (e.g. resumed after a crash between the + // directory being removed and the tombstone being cleared) finds + // nothing left to remove -- treat that the same as success, matching + // S3/Azure/GCS, whose prefix delete is already a no-op success when + // the prefix is already empty. + let result = match result { + Err(e) if e.kind() == std::io::ErrorKind::NotFound => Ok(()), + other => other, + }; if result.is_ok() { increment_object_store_calls_by_date( "DELETE", diff --git a/src/storage/object_storage.rs b/src/storage/object_storage.rs index b198a59b5..55a9a7f00 100644 --- a/src/storage/object_storage.rs +++ b/src/storage/object_storage.rs @@ -1441,7 +1441,7 @@ pub fn sync_all_streams(joinset: &mut JoinSet>) { spawn_stream_deletion(stream_name, tenant_id); } } - Ok(false) => { + Ok(false) if !is_deletion_owner => { // Deletion already finished elsewhere and // the tombstone is gone, but this node's // resident entry was never dropped -- e.g. @@ -1452,6 +1452,18 @@ pub fn sync_all_streams(joinset: &mut JoinSet>) { // deleting=true state. PARSEABLE.streams.delete(&stream_name, &tenant_id); } + Ok(false) => { + // On the owning node, an absent tombstone + // can also just mean this delete hasn't + // written it yet (mark_deleting() runs + // first, with the tombstone write still + // in flight). Reaping here would race + // that window and let a concurrent reload + // resurrect the stream mid-deletion -- + // the owner's own spawn_stream_deletion + // job removes this entry once the actual + // delete (and tombstone clear) completes. + } Err(e) => error!( "failed to check tombstone status for {stream_name}: {e}" ), From b0b0f659566e44c65377027dc543f5b059e91072 Mon Sep 17 00:00:00 2001 From: prabhaks Date: Wed, 16 Sep 2026 12:35:31 -0700 Subject: [PATCH 07/14] Fix stale is_deleting flag blocking recreation, serialize delete vs create/update Root cause of the Quest distributed CI failure: a node that doesn't run the background deletion job itself (an ingestor) never clears its resident stream's is_deleting flag on its own -- it only self-heals on the next sync_all_streams tick. Recreating a stream shortly after deleting it (a pattern several Quest tests use) landed in the window before that tick, so create_update_stream kept rejecting the recreate with 409 even though the tombstone was already gone. It now re-checks the durable tombstone before trusting the in-memory flag, and clears the flag itself when the tombstone turns out to already be cleared. Also addresses two new CodeRabbit findings on the delete/create race: mark_deleting() plus the tombstone write could interleave with a concurrent create_update_stream on the same node, since only the query node's put_stream held CREATE_STREAM_LOCK and DELETE never did. Standalone and ingestor put_stream/delete had no lock at all. All three now hold the same lock across the is_deleting()-check-through- tombstone-write window. --- src/handlers/http/logstream.rs | 79 ++++++++++------- .../http/modal/ingest/ingestor_logstream.rs | 19 ++++- .../http/modal/query/querier_logstream.rs | 85 +++++++++++-------- src/parseable/mod.rs | 25 +++++- src/parseable/streams.rs | 9 +- 5 files changed, 144 insertions(+), 73 deletions(-) diff --git a/src/handlers/http/logstream.rs b/src/handlers/http/logstream.rs index d53dc76bc..41d39292f 100644 --- a/src/handlers/http/logstream.rs +++ b/src/handlers/http/logstream.rs @@ -50,8 +50,16 @@ use itertools::Itertools; use serde_json::{Value, json}; use std::fs; use std::sync::Arc; +use tokio::sync::Mutex; use tracing::{Instrument, warn}; +// Shared between put_stream and delete: without it, a concurrent create/ +// update could read is_deleting()=false right before delete() flips it and +// writes the tombstone, then go on to write fresh stream data that the +// background deletion job -- already committed to running by that point -- +// would sweep up from underneath it. +pub static CREATE_STREAM_LOCK: Mutex<()> = Mutex::const_new(()); + pub async fn delete( req: HttpRequest, logstream: Path, @@ -59,43 +67,55 @@ pub async fn delete( let stream_name = logstream.into_inner(); // Error out if stream doesn't exist in memory, or in the case of query node, in storage as well let tenant_id = get_tenant_id_from_request(&req); - if !PARSEABLE - .check_or_load_stream(&stream_name, &tenant_id) - .await - { - 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)?; + let stream = { + // Scoped to just this check-through-tombstone-write window: + // everything after is already guarded by is_deleting()/ + // is_tombstoned() checks on the read/write paths. + let _guard = CREATE_STREAM_LOCK.lock().await; + + if !PARSEABLE + .check_or_load_stream(&stream_name, &tenant_id) + .await + { + return Err(StreamNotFound(stream_name).into()); + } - // 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 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(); + + // 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. + if let Err(e) = objectstore + .put_object( + &tombstone_path(&stream_name, &tenant_id), + to_bytes(&()), + &tenant_id, + ) + .await + { + // Nothing durable happened -- undo the in-memory flag so the stream + // isn't left permanently blocked by a transient write failure. + stream.clear_deleting(); + return Err(e.into()); + } - let objectstore = PARSEABLE.storage.get_object_store(); + stream + }; - // 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. - if let Err(e) = objectstore - .put_object( - &tombstone_path(&stream_name, &tenant_id), - to_bytes(&()), - &tenant_id, - ) - .await - { - // Nothing durable happened -- undo the in-memory flag so the stream - // isn't left permanently blocked by a transient write failure. - stream.clear_deleting(); - return Err(e.into()); - } + let objectstore = PARSEABLE.storage.get_object_store(); // Best-effort: makes the stream vanish from listings almost // immediately. Not fatal if it fails -- is_deleting()/is_tombstoned() @@ -259,6 +279,7 @@ pub async fn put_stream( let stream_name = logstream.into_inner(); let tenant_id = get_tenant_id_from_request(&req); + let _guard = CREATE_STREAM_LOCK.lock().await; PARSEABLE .create_update_stream(req.headers(), &body, &stream_name, &tenant_id) .await?; diff --git a/src/handlers/http/modal/ingest/ingestor_logstream.rs b/src/handlers/http/modal/ingest/ingestor_logstream.rs index 02b281813..9d885b819 100644 --- a/src/handlers/http/modal/ingest/ingestor_logstream.rs +++ b/src/handlers/http/modal/ingest/ingestor_logstream.rs @@ -24,6 +24,7 @@ use actix_web::{ web::{Json, Path}, }; use bytes::Bytes; +use tokio::sync::Mutex; use tracing::warn; use crate::option::Mode; @@ -34,6 +35,12 @@ use crate::{ utils::get_tenant_id_from_request, }; +// Shared between put_stream and delete: without it, a concurrent create/ +// update could read is_deleting()=false right before delete() flips it, +// then go on to write fresh stream data into a directory delete() is about +// to (or just did) remove. +static CREATE_STREAM_LOCK: Mutex<()> = Mutex::const_new(()); + pub async fn retention_cleanup( req: HttpRequest, stream_name: Path, @@ -75,9 +82,14 @@ pub async fn delete( ) -> Result { let stream_name = stream_name.into_inner(); 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(); + + let stream_dir = { + let _guard = CREATE_STREAM_LOCK.lock().await; + // Delete from staging + let stream_dir = PARSEABLE.get_stream(&stream_name, &tenant_id)?; + stream_dir.mark_deleting(); + stream_dir + }; // delete staging only for ingest server or standalone server // else skip @@ -110,6 +122,7 @@ pub async fn put_stream( ) -> Result { let stream_name = stream_name.into_inner(); let tenant_id = get_tenant_id_from_request(&req); + let _guard = CREATE_STREAM_LOCK.lock().await; PARSEABLE .create_update_stream(req.headers(), &body, &stream_name, &tenant_id) .await?; diff --git a/src/handlers/http/modal/query/querier_logstream.rs b/src/handlers/http/modal/query/querier_logstream.rs index 3ebc4ec65..f936137ad 100644 --- a/src/handlers/http/modal/query/querier_logstream.rs +++ b/src/handlers/http/modal/query/querier_logstream.rs @@ -63,48 +63,65 @@ pub async fn delete( ) -> Result { let stream_name = stream_name.into_inner(); let tenant_id = get_tenant_id_from_request(&req); - // if the stream not found in memory map, - //check if it exists in the storage - //create stream and schema from storage - if !PARSEABLE.streams.contains(&stream_name, &tenant_id) - && !PARSEABLE - .create_stream_and_schema_from_storage(&stream_name, &tenant_id) - .await - .unwrap_or(false) - { - 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)?; + let stream = { + // Shared with put_stream: without it, a concurrent create/update on + // this node could read is_deleting()=false right before this + // handler flips it and writes the tombstone, then go on to write + // fresh stream data that the background deletion job -- already + // committed to running by that point -- would sweep up from + // underneath it. Scoped to just this check-through-tombstone-write + // window: everything after is already guarded by is_deleting()/ + // is_tombstoned() checks on the read/write paths. + let _guard = CREATE_STREAM_LOCK.lock().await; + + // if the stream not found in memory map, + //check if it exists in the storage + //create stream and schema from storage + if !PARSEABLE.streams.contains(&stream_name, &tenant_id) + && !PARSEABLE + .create_stream_and_schema_from_storage(&stream_name, &tenant_id) + .await + .unwrap_or(false) + { + return Err(StreamNotFound(stream_name.clone()).into()); + } - // 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 stream = PARSEABLE.get_stream(&stream_name, &tenant_id)?; - let objectstore = PARSEABLE.storage.get_object_store(); + // 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(); - // 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. - if let Err(e) = objectstore - .put_object( - &tombstone_path(&stream_name, &tenant_id), - to_bytes(&()), - &tenant_id, - ) - .await - { - // Nothing durable happened -- undo the in-memory flag so the stream - // isn't left permanently blocked by a transient write failure. - stream.clear_deleting(); - return Err(e.into()); - } + let objectstore = PARSEABLE.storage.get_object_store(); + + // 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. + if let Err(e) = objectstore + .put_object( + &tombstone_path(&stream_name, &tenant_id), + to_bytes(&()), + &tenant_id, + ) + .await + { + // Nothing durable happened -- undo the in-memory flag so the stream + // isn't left permanently blocked by a transient write failure. + stream.clear_deleting(); + return Err(e.into()); + } + + stream + }; + + let objectstore = PARSEABLE.storage.get_object_store(); // Best-effort: makes the stream vanish from listings almost // immediately, without touching every listing endpoint individually. diff --git a/src/parseable/mod.rs b/src/parseable/mod.rs index 3dd151a5a..6b4b7dbab 100644 --- a/src/parseable/mod.rs +++ b/src/parseable/mod.rs @@ -804,10 +804,27 @@ impl Parseable { && let Ok(stream) = self.get_stream(stream_name, tenant_id) && stream.is_deleting() { - return Err(StreamError::Custom { - msg: format!("Logstream {stream_name} is being deleted, please retry shortly"), - status: StatusCode::CONFLICT, - }); + // The flag can be stale on a node that never runs the + // background deletion job itself (e.g. an ingestor: see its + // `delete()` handler) -- it only self-heals once + // `sync_all_streams` next notices the tombstone is gone, which + // can lag well behind a client recreating the stream right + // away. Re-check the durable tombstone before trusting the + // in-memory flag, and clear it here instead of blocking + // creation for up to a full sync interval. + if is_tombstoned( + self.storage.get_object_store().as_ref(), + stream_name, + tenant_id, + ) + .await? + { + return Err(StreamError::Custom { + msg: format!("Logstream {stream_name} is being deleted, please retry shortly"), + status: StatusCode::CONFLICT, + }); + } + stream.clear_deleting(); } // A tombstoned-but-not-yet-purged stream must be rejected the same diff --git a/src/parseable/streams.rs b/src/parseable/streams.rs index a0948b746..d475555b7 100644 --- a/src/parseable/streams.rs +++ b/src/parseable/streams.rs @@ -1632,10 +1632,13 @@ impl Stream { self.metadata.write().expect(LOCK_EXPECT).deleting = true; } - /// Undoes a `mark_deleting()` call whose tombstone write then failed. - /// Safe only because nothing durable was ever created -- unlike + /// Undoes a `mark_deleting()` call, in one of two cases: the tombstone + /// write that would have made it durable just failed, or the caller has + /// independently confirmed (via `is_tombstoned()`) that the tombstone is + /// already gone and this flag is just a stale leftover on a node that + /// doesn't run the background deletion job itself. Unlike /// `mark_deleting()`, this is not part of the monotonic contract, and - /// must never be called once the tombstone actually exists in storage. + /// must never be called while the tombstone still exists in storage. pub fn clear_deleting(&self) { self.metadata.write().expect(LOCK_EXPECT).deleting = false; } From 04fc9e761801ca435f241a094576b6a4026302b0 Mon Sep 17 00:00:00 2001 From: prabhaks Date: Wed, 16 Sep 2026 14:22:30 -0700 Subject: [PATCH 08/14] fix: stale is_deleting self-heal didn't clear the "already exists" check The stale-flag self-heal added in the previous commit cleared is_deleting()/removed the flag but left stream_in_memory_dont_update untouched, so create_update_stream still fell through to the "Logstream already exists" 400 right after self-healing -- exactly the Quest delete-then-recreate pattern that broke both the standalone and distributed CI runs (TestSmokeIngestEventsToStream: expected 200, got 400). Now the stale entry is dropped from the in-memory map (not just its flag cleared) so get_or_create doesn't hand back the same stale Arc, and stream_in_memory_dont_update is corrected so the "already exists" check no longer trips. Also extracted the deletion/tombstone checks in create_update_stream into reject_if_stream_deleting to bring its cyclomatic complexity back down (DeepSource flagged RS-R1000 after the previous commit pushed it to "very-high" risk). --- src/parseable/mod.rs | 127 ++++++++++++++++++++++++++----------------- 1 file changed, 77 insertions(+), 50 deletions(-) diff --git a/src/parseable/mod.rs b/src/parseable/mod.rs index 6b4b7dbab..10d2509d3 100644 --- a/src/parseable/mod.rs +++ b/src/parseable/mod.rs @@ -742,59 +742,17 @@ impl Parseable { Ok(()) } - pub async fn create_update_stream( + /// Rejects a create/update call blocked by an in-progress or stale + /// deletion, self-healing a stale `is_deleting()` flag once its + /// tombstone is confirmed gone. Returns the (possibly corrected) value + /// of `stream_in_memory_dont_update` for the caller's subsequent + /// "already exists" check. + async fn reject_if_stream_deleting( &self, - headers: &HeaderMap, - body: &Bytes, stream_name: &str, tenant_id: &Option, - ) -> Result { - let PutStreamHeaders { - time_partition_limit, - custom_partition, - static_schema_flag, - update_stream_flag, - stream_type, - log_source, - mut telemetry_type, - telemetry_type_set, - dataset_tags, - dataset_labels, - infer_timestamp, - infer_timestamp_set, - } = headers.into(); - - // x-p-infer-timestamp can only be set during stream creation, not on update. - // It is only valid for otel-metrics datasets. - if infer_timestamp_set { - if update_stream_flag { - return Err(StreamError::Custom { - msg: format!( - "Header {} can only be set at stream creation, not on update", - crate::handlers::INFER_TIMESTAMP_KEY - ), - status: StatusCode::BAD_REQUEST, - }); - } - if log_source != LogSource::OtelMetrics { - return Err(StreamError::Custom { - msg: format!( - "Header {} is only supported for otel-metrics datasets", - crate::handlers::INFER_TIMESTAMP_KEY - ), - status: StatusCode::BAD_REQUEST, - }); - } - } - - // For OTel datasets the telemetry_type is implied by the log_source and - // they must be consistent. Auto-derive it when not provided; reject any - // mismatch when both x-p-log-source and x-p-telemetry-type are set. - telemetry_type = resolve_telemetry_type(&log_source, telemetry_type, telemetry_type_set)?; - - let stream_in_memory_dont_update = - self.streams.contains(stream_name, tenant_id) && !update_stream_flag; - + stream_in_memory_dont_update: bool, + ) -> Result { // A stream still resident with is_deleting()=true is functionally // gone (reads/writes are already rejected elsewhere), but its entry // isn't removed from memory until the background deletion job @@ -825,6 +783,16 @@ impl Parseable { }); } stream.clear_deleting(); + + // The stale entry itself (data path, schema, old metadata) must + // go too, not just the flag -- otherwise the "already exists" + // check just below still trips on it, and get_or_create's + // resident-entry short-circuit would hand back this same stale + // Arc instead of actually recreating it. Dropping it + // here makes this indistinguishable from a stream that was + // never resident in the first place. + self.streams.delete(stream_name, tenant_id); + return Ok(false); } // A tombstoned-but-not-yet-purged stream must be rejected the same @@ -851,6 +819,65 @@ impl Parseable { }); } + Ok(stream_in_memory_dont_update) + } + + pub async fn create_update_stream( + &self, + headers: &HeaderMap, + body: &Bytes, + stream_name: &str, + tenant_id: &Option, + ) -> Result { + let PutStreamHeaders { + time_partition_limit, + custom_partition, + static_schema_flag, + update_stream_flag, + stream_type, + log_source, + mut telemetry_type, + telemetry_type_set, + dataset_tags, + dataset_labels, + infer_timestamp, + infer_timestamp_set, + } = headers.into(); + + // x-p-infer-timestamp can only be set during stream creation, not on update. + // It is only valid for otel-metrics datasets. + if infer_timestamp_set { + if update_stream_flag { + return Err(StreamError::Custom { + msg: format!( + "Header {} can only be set at stream creation, not on update", + crate::handlers::INFER_TIMESTAMP_KEY + ), + status: StatusCode::BAD_REQUEST, + }); + } + if log_source != LogSource::OtelMetrics { + return Err(StreamError::Custom { + msg: format!( + "Header {} is only supported for otel-metrics datasets", + crate::handlers::INFER_TIMESTAMP_KEY + ), + status: StatusCode::BAD_REQUEST, + }); + } + } + + // For OTel datasets the telemetry_type is implied by the log_source and + // they must be consistent. Auto-derive it when not provided; reject any + // mismatch when both x-p-log-source and x-p-telemetry-type are set. + telemetry_type = resolve_telemetry_type(&log_source, telemetry_type, telemetry_type_set)?; + + let stream_in_memory_dont_update = + self.streams.contains(stream_name, tenant_id) && !update_stream_flag; + let stream_in_memory_dont_update = self + .reject_if_stream_deleting(stream_name, tenant_id, stream_in_memory_dont_update) + .await?; + // check if stream in storage only if not in memory // for Parseable OSS, create_update_stream is called only from query node // for Parseable Enterprise, create_update_stream is called from prism node From 4a719ce78b64e55dfb69a5e979b46e4275db1cca Mon Sep 17 00:00:00 2001 From: prabhaks Date: Thu, 17 Sep 2026 09:40:54 -0700 Subject: [PATCH 09/14] 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. --- src/handlers/http/logstream.rs | 42 +++++++++++++------ .../http/modal/query/querier_logstream.rs | 7 ++++ src/storage/object_storage.rs | 38 +++++++++++++++++ 3 files changed, 75 insertions(+), 12 deletions(-) diff --git a/src/handlers/http/logstream.rs b/src/handlers/http/logstream.rs index 41d39292f..32266f029 100644 --- a/src/handlers/http/logstream.rs +++ b/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, diff --git a/src/handlers/http/modal/query/querier_logstream.rs b/src/handlers/http/modal/query/querier_logstream.rs index f936137ad..244b106de 100644 --- a/src/handlers/http/modal/query/querier_logstream.rs +++ b/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::>::from_query(req.query_string()) .map_err(|_| StreamError::InvalidQueryParameter(STATS_DATE_QUERY_PARAM.to_string()))?; diff --git a/src/storage/object_storage.rs b/src/storage/object_storage.rs index 55a9a7f00..400bbaf05 100644 --- a/src/storage/object_storage.rs +++ b/src/storage/object_storage.rs @@ -1501,6 +1501,44 @@ pub fn sync_all_streams(joinset: &mut JoinSet>) { 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, + ); + } } } From 2b5e0819311ad4947b8588a65c6a9b814aa14489 Mon Sep 17 00:00:00 2001 From: prabhaks Date: Thu, 17 Sep 2026 11:47:15 -0700 Subject: [PATCH 10/14] Close remaining resurrection/race gaps found in review, fix stale comments - create_stream_if_not_exists (and the OTEL ingest path specifically, which never went through validate_stream_for_ingestion at all) could implicitly recreate a stream while its background deletion was still running, since neither checked the tombstone before falling through to create_stream(). - spawn_stream_deletion's finalization removed a stream from memory by name only, so a client recreating the same name right as the background job finished could have its brand-new stream evicted instead of the stale one. Streams::delete_if_still_deleting only removes an entry that's still actually flagged deleting. - LocalFS list_streams could fail the entire listing if a stream's deletion (including its tombstone) finished in the narrow window between the directory snapshot and the per-entry check, misreading "already fully gone" as "corrupt". - sync_all_streams' tombstone reconciliation only ever looked at streams already flagged deleting locally, so a node that missed the delete handler's fan-out push (e.g. a transient liveness-check failure) never self-healed at all, contrary to what its own comments claimed. It now also scans the tombstone directory for resident streams that were never flagged, on every node type. - A retried DELETE against a query replica that hasn't resumed a tombstoned stream yet returned a plain 404, indistinguishable from the stream never having existed; it now reports the deletion as already in progress. - Ingestor's delete endpoint returned 200 while the other two returned 202 for the same "deletion started" response. Added LocalFS-level tests for the snapshot race, a mixed tombstoned/corrupt listing, and delete_stream's idempotent-retry behavior. --- src/handlers/http/ingest.rs | 10 ++ .../http/modal/ingest/ingestor_logstream.rs | 2 +- .../http/modal/query/querier_logstream.rs | 34 ++++++- src/parseable/mod.rs | 16 +++ src/parseable/streams.rs | 22 ++++- src/storage/localfs.rs | 97 +++++++++++++++++++ src/storage/object_storage.rs | 82 +++++++++++----- 7 files changed, 231 insertions(+), 32 deletions(-) diff --git a/src/handlers/http/ingest.rs b/src/handlers/http/ingest.rs index 2f79ba105..18f496dd0 100644 --- a/src/handlers/http/ingest.rs +++ b/src/handlers/http/ingest.rs @@ -256,6 +256,16 @@ pub async fn setup_otel_stream( let mut time_partition = None; // Validate stream compatibility if let Ok(stream) = PARSEABLE.get_stream(&stream_name, &tenant_id) { + // Unlike plain JSON ingest() and post_event(), this path doesn't go + // through validate_stream_for_ingestion, so it needs its own check: + // a stream flagged deleting must reject writes rather than staging + // data that will never be uploaded (sync_all_streams skips a + // deleting stream's staged files) nor cleaned up (the delete + // handler's own staging cleanup already ran before this write + // recreated the directory). + if stream.is_deleting() { + return Err(PostError::StreamBeingDeleted(stream_name.clone())); + } match log_source { LogSource::OtelLogs => { // For logs, reject if stream is metrics or traces diff --git a/src/handlers/http/modal/ingest/ingestor_logstream.rs b/src/handlers/http/modal/ingest/ingestor_logstream.rs index 9d885b819..b11f5c50c 100644 --- a/src/handlers/http/modal/ingest/ingestor_logstream.rs +++ b/src/handlers/http/modal/ingest/ingestor_logstream.rs @@ -111,7 +111,7 @@ pub async fn delete( // ingestion for this stream with a clear "being deleted" error. Ok(( format!("log stream {stream_name} deletion started"), - StatusCode::OK, + StatusCode::ACCEPTED, )) } diff --git a/src/handlers/http/modal/query/querier_logstream.rs b/src/handlers/http/modal/query/querier_logstream.rs index 244b106de..df6d3313a 100644 --- a/src/handlers/http/modal/query/querier_logstream.rs +++ b/src/handlers/http/modal/query/querier_logstream.rs @@ -51,7 +51,9 @@ use crate::{ stats, storage::{ ObjectStoreFormat, StreamType, - object_storage::{spawn_stream_deletion, stream_json_path, to_bytes, tombstone_path}, + object_storage::{ + is_tombstoned, spawn_stream_deletion, stream_json_path, to_bytes, tombstone_path, + }, }, utils::get_tenant_id_from_request, }; @@ -88,6 +90,25 @@ pub async fn delete( .await .unwrap_or(false) { + // create_stream_and_schema_from_storage returns false both for + // "genuinely doesn't exist" and "exists but is tombstoned" -- + // check the tombstone directly so a retried DELETE against a + // replica that hasn't resumed this stream yet reports the + // deletion as already accepted instead of a plain 404 + // indistinguishable from the stream never having existed. + if is_tombstoned( + PARSEABLE.storage.get_object_store().as_ref(), + &stream_name, + &tenant_id, + ) + .await + .unwrap_or(false) + { + return Ok(( + format!("log stream {stream_name} deletion already in progress"), + StatusCode::ACCEPTED, + )); + } return Err(StreamNotFound(stream_name.clone()).into()); } @@ -138,9 +159,14 @@ pub async fn delete( // Best-effort: the tombstone is already durable and the stream is // already flagged, so a fan-out failure must not turn an - // already-accepted deletion into an error response -- an ingestor that - // misses this push self-heals via sync_all_streams within one sync - // interval regardless. + // already-accepted deletion into an error response. An ingestor that + // misses this push doesn't self-heal on its own next sync tick (its + // local is_deleting() stays false, which is exactly what the resident- + // stream self-heal check keys off) -- recovery instead comes from + // sync_all_streams' separate tombstone-directory scan, which lists + // tombstones directly rather than depending on the flag already being + // set locally, at the cost of taking up to one sync interval rather + // than being immediate. let fanout_stream_name = stream_name.clone(); if let Err(e) = cluster::for_each_live_node(&tenant_id, move |node| { let url = format!( diff --git a/src/parseable/mod.rs b/src/parseable/mod.rs index 10d2509d3..58b5778fd 100644 --- a/src/parseable/mod.rs +++ b/src/parseable/mod.rs @@ -668,6 +668,22 @@ impl Parseable { return Ok(true); } + // create_stream_and_schema_from_storage (or its Mode::All skip above) + // can't tell this caller "false because a deletion is in progress" + // apart from "false because the name is genuinely free" -- both + // return/short-circuit to Ok(false)/skipped. Check the durable + // tombstone directly so an implicit create-on-ingest can't + // resurrect a stream whose background deletion is still running. + if is_tombstoned( + self.storage.get_object_store().as_ref(), + stream_name, + tenant_id, + ) + .await? + { + return Err(PostError::StreamBeingDeleted(stream_name.to_string())); + } + self.create_stream( stream_name.to_string(), "", diff --git a/src/parseable/streams.rs b/src/parseable/streams.rs index d475555b7..84a900ca5 100644 --- a/src/parseable/streams.rs +++ b/src/parseable/streams.rs @@ -2020,14 +2020,32 @@ impl Streams { plans } - /// TODO: validate possibility of stream continuing to exist despite being deleted pub fn delete(&self, stream_name: &str, tenant_id: &Option) { let tenant_id = tenant_id.as_deref().unwrap_or(DEFAULT_TENANT); let mut guard = self.write().expect(LOCK_EXPECT); if let Some(tenant_streams) = guard.get_mut(tenant_id) { tenant_streams.remove(stream_name); } - // self.write().expect(LOCK_EXPECT).remove(stream_name); + } + + /// Removes a stream's entry only if it's still flagged `deleting`. + /// A background deletion job finalizes by name, not by holding a + /// reference to the specific `Stream` instance it started deleting -- + /// so without this check, a client that recreates the same name while + /// the job's finalization is still in flight (e.g. right after a + /// concurrent self-heal cleared a stale flag and inserted a fresh + /// entry) would have that brand-new entry silently evicted instead of + /// the stale one the job actually meant to clean up. + pub fn delete_if_still_deleting(&self, stream_name: &str, tenant_id: &Option) { + let tenant_id = tenant_id.as_deref().unwrap_or(DEFAULT_TENANT); + let mut guard = self.write().expect(LOCK_EXPECT); + if let Some(tenant_streams) = guard.get_mut(tenant_id) + && tenant_streams + .get(stream_name) + .is_some_and(|stream| stream.is_deleting()) + { + tenant_streams.remove(stream_name); + } } pub fn contains(&self, stream_name: &str, tenant_id: &Option) -> bool { diff --git a/src/storage/localfs.rs b/src/storage/localfs.rs index 78823cd2d..5bae64b19 100644 --- a/src/storage/localfs.rs +++ b/src/storage/localfs.rs @@ -905,6 +905,13 @@ async fn dir_with_stream( // whole listing over a stream that's in the middle of being // deleted. Ok(None) + } else if !path.exists() { + // The whole directory -- including its tombstone -- can vanish + // between this listing's `read_dir` snapshot and the checks + // above, e.g. a small/fast stream's background deletion runs to + // completion in that exact window. Nothing here is actually + // corrupt; there's simply nothing left to report. + Ok(None) } else { let err: Box = format!("found {}", entry.path().display()).into(); @@ -1017,4 +1024,94 @@ mod list_streams_tombstone_tests { assert!(storage.list_streams().await.is_err()); } + + #[tokio::test] + async fn corrupt_directory_still_errors_alongside_a_tombstoned_one() { + let dir = TempDir::new().unwrap(); + let storage = LocalFS::new(dir.path().to_path_buf()); + + // A per-directory check keyed on that directory's own tombstone + // path must still catch a genuinely corrupt directory even when a + // legitimately tombstoned stream exists elsewhere in the same + // listing -- an implementation that treated "any tombstone exists + // anywhere" as license to skip every missing-stream.json directory + // would incorrectly let this one through too. + storage + .put_object( + &stream_json_path_for_test("deleting-stream"), + to_bytes(&()), + &None, + ) + .await + .unwrap(); + storage + .delete_object(&stream_json_path_for_test("deleting-stream"), &None) + .await + .unwrap(); + storage + .put_object( + &tombstone_path("deleting-stream", &None), + to_bytes(&()), + &None, + ) + .await + .unwrap(); + std::fs::create_dir_all(dir.path().join("not-a-stream")).unwrap(); + + assert!(storage.list_streams().await.is_err()); + } + + #[tokio::test] + async fn dir_that_finished_deleting_between_snapshot_and_check_is_skipped() { + use futures::stream::StreamExt; + use tokio_stream::wrappers::ReadDirStream; + + let dir = TempDir::new().unwrap(); + std::fs::create_dir_all(dir.path().join("deleting-stream")).unwrap(); + + // Snapshot the directory listing (as list_streams does) while + // "deleting-stream" still exists, capturing its DirEntry. + let read_dir = tokio::fs::read_dir(dir.path()).await.unwrap(); + let mut entries = ReadDirStream::new(read_dir); + let entry = entries.next().await.unwrap().unwrap(); + assert_eq!(entry.file_name(), "deleting-stream"); + + // Simulate the whole deletion (directory + tombstone) completing + // after the snapshot was taken but before dir_with_stream's own + // stream.json/tombstone checks run against it. + std::fs::remove_dir_all(dir.path().join("deleting-stream")).unwrap(); + + let result = super::dir_with_stream(entry, &[], dir.path()).await; + assert_eq!(result.unwrap(), None); + } +} + +#[cfg(test)] +mod delete_stream_idempotency_tests { + use temp_dir::TempDir; + + use super::{LocalFS, ObjectStorage}; + + #[tokio::test] + async fn deleting_an_existing_stream_directory_succeeds() { + let dir = TempDir::new().unwrap(); + let storage = LocalFS::new(dir.path().to_path_buf()); + std::fs::create_dir_all(dir.path().join("mystream")).unwrap(); + + assert!(storage.delete_stream("mystream", &None).await.is_ok()); + assert!(!dir.path().join("mystream").exists()); + } + + #[tokio::test] + async fn retrying_delete_on_an_already_removed_prefix_is_a_no_op_success() { + let dir = TempDir::new().unwrap(); + let storage = LocalFS::new(dir.path().to_path_buf()); + + // Never created "mystream" at all -- this is what a resumed + // deletion sees if the process crashed after `remove_dir_all` + // already finished but before the tombstone was cleared. Matches + // S3/Azure/GCS, whose prefix delete is already a no-op success + // against an empty/nonexistent prefix. + assert!(storage.delete_stream("mystream", &None).await.is_ok()); + } } diff --git a/src/storage/object_storage.rs b/src/storage/object_storage.rs index 400bbaf05..3c41af228 100644 --- a/src/storage/object_storage.rs +++ b/src/storage/object_storage.rs @@ -1387,7 +1387,9 @@ pub fn spawn_stream_deletion(stream_name: String, tenant_id: Option) { "background deletion of {stream_name} finished but failed to clear its tombstone: {e}" ); } - PARSEABLE.streams.delete(stream_name, tenant_id); + PARSEABLE + .streams + .delete_if_still_deleting(stream_name, tenant_id); if let Err(e) = stats::delete_stats(stream_name, "json", tenant_id) { warn!("failed to clear stats for deleted stream {stream_name}: {e:?}"); } @@ -1411,11 +1413,17 @@ pub fn sync_all_streams(joinset: &mut JoinSet>) { for stream_name in PARSEABLE.streams.list(&tenant_id) { if let Ok(stream) = PARSEABLE.get_stream(&stream_name, &tenant_id) { if stream.is_deleting() { - // Fast path (the DELETE handler's own `for_each_live_node` - // push) already reaches most nodes immediately. This is - // the fallback for one that missed it, e.g. a node that - // was down or partitioned at the time: bounded to at most - // one sync interval instead of running forever. + // This handles a stream this node already knows is + // deleting -- either it received the DELETE handler's + // `for_each_live_node` push directly, or it's the owning + // node itself resuming its own in-flight job after a + // restart. A node that missed the fan-out push entirely + // (e.g. it was down or transiently unreachable at that + // exact moment) never gets is_deleting()=true set in the + // first place, so it never reaches this branch at all -- + // that case is instead caught by the tombstone-directory + // scan below, which doesn't depend on the flag already + // being set locally. let object_store = object_store.clone(); let tenant_id = tenant_id.clone(); let stream_name = stream_name.clone(); @@ -1450,7 +1458,9 @@ pub fn sync_all_streams(joinset: &mut JoinSet>) { // so a stream recreated under the same // name doesn't inherit a stuck // deleting=true state. - PARSEABLE.streams.delete(&stream_name, &tenant_id); + PARSEABLE + .streams + .delete_if_still_deleting(&stream_name, &tenant_id); } Ok(false) => { // On the owning node, an absent tombstone @@ -1502,32 +1512,54 @@ pub fn sync_all_streams(joinset: &mut JoinSet>) { ); } - // 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 { + // Reconcile tombstones that the per-stream loop above can't reach on + // its own, since it only acts on a stream whose local is_deleting() + // is already true: + // - A resident stream that's still flagged false because this node + // never ran mark_deleting() on it at all -- e.g. an ingestor that + // missed the DELETE handler's `for_each_live_node` fan-out push + // (a transient liveness-check failure at the wrong instant is + // enough), so it kept accepting ingestion into an already-deleted + // stream indefinitely. Flagging it here is just an in-memory + // update, so this runs on every node type, not only the owner. + // - A stream already gone from this node's memory entirely because + // its owning deletion job already finished, but whose final + // tombstone-clear call failed (e.g. a transient network error) -- + // only the owner should retry the physical delete for this case. + // A single tombstone-directory LIST finds every candidate at once, + // rather than a HEAD per resident stream every interval. + { let object_store = object_store.clone(); let tenant_id = tenant_id.clone(); + let is_deletion_owner = PARSEABLE.options.mode != Mode::Ingest; 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; + match PARSEABLE.get_stream(&stream_name, &tenant_id) { + Ok(stream) if !stream.is_deleting() => { + stream.mark_deleting(); + } + // Already flagged: the per-stream loop + // above already owns reconciling this + // one, nothing more to do here. + Ok(_) => {} + Err(_) if is_deletion_owner => { + // 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()); + } + // Not resident on a non-owner (e.g. an + // ingestor that never loaded this stream + // in the first place) -- nothing to + // reconcile. + Err(_) => {} } - // 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!( From 8893e2542116d2f09d9edbc808fcfa114d4840f1 Mon Sep 17 00:00:00 2001 From: prabhaks Date: Thu, 17 Sep 2026 12:55:10 -0700 Subject: [PATCH 11/14] Address CodeRabbit findings: stats race, error swallowing, stale snapshot - spawn_stream_deletion cleared the tombstone before removing the map entry and clearing stats. While the tombstone is still present, a recreate is rejected with 409, so nothing could race the map cleanup -- but clearing it first opened a window where a legitimate recreate's stats could be zeroed by the trailing delete_stats call. Reordered so tombstone clear is last. - The new tombstone-check on a retried DELETE swallowed real storage errors as "not tombstoned", turning them into a misleading 404 instead of surfacing the actual failure. - sync_all_streams' tombstone-directory scan acted on a snapshot: if the original deletion finished and the name was legitimately recreated between the snapshot and this stream being processed, the fresh stream could be marked deleting with no tombstone left to ever clear it again. Added a live re-check of that one name immediately before flagging it. --- .../http/modal/query/querier_logstream.rs | 7 +-- src/storage/object_storage.rs | 43 ++++++++++++++++--- 2 files changed, 40 insertions(+), 10 deletions(-) diff --git a/src/handlers/http/modal/query/querier_logstream.rs b/src/handlers/http/modal/query/querier_logstream.rs index df6d3313a..5a8628176 100644 --- a/src/handlers/http/modal/query/querier_logstream.rs +++ b/src/handlers/http/modal/query/querier_logstream.rs @@ -95,14 +95,15 @@ pub async fn delete( // check the tombstone directly so a retried DELETE against a // replica that hasn't resumed this stream yet reports the // deletion as already accepted instead of a plain 404 - // indistinguishable from the stream never having existed. + // indistinguishable from the stream never having existed. Real + // storage errors propagate instead of being folded into "not + // tombstoned", which would otherwise misreport them as 404. if is_tombstoned( PARSEABLE.storage.get_object_store().as_ref(), &stream_name, &tenant_id, ) - .await - .unwrap_or(false) + .await? { return Ok(( format!("log stream {stream_name} deletion already in progress"), diff --git a/src/storage/object_storage.rs b/src/storage/object_storage.rs index 3c41af228..4d83b81b6 100644 --- a/src/storage/object_storage.rs +++ b/src/storage/object_storage.rs @@ -1379,6 +1379,23 @@ pub fn spawn_stream_deletion(stream_name: String, tenant_id: Option) { let storage = PARSEABLE.storage.get_object_store(); match storage.delete_stream(stream_name, tenant_id).await { Ok(()) => { + // Clear the map entry and stats *before* the tombstone, not + // after: while the tombstone is still present, any + // concurrent create/update for this name is rejected with + // 409 by reject_if_stream_deleting's own tombstone check, so + // nothing can have recreated this name yet. Clearing the + // tombstone first would open a window where a legitimate + // recreate succeeds before this cleanup runs -- harmless for + // the map entry itself (delete_if_still_deleting only + // touches an entry still flagged deleting), but + // stats::delete_stats has no such per-entry guard and would + // zero out the freshly recreated stream's stats instead. + PARSEABLE + .streams + .delete_if_still_deleting(stream_name, tenant_id); + if let Err(e) = stats::delete_stats(stream_name, "json", tenant_id) { + warn!("failed to clear stats for deleted stream {stream_name}: {e:?}"); + } if let Err(e) = storage .delete_object(&tombstone_path(stream_name, tenant_id), tenant_id) .await @@ -1387,12 +1404,6 @@ pub fn spawn_stream_deletion(stream_name: String, tenant_id: Option) { "background deletion of {stream_name} finished but failed to clear its tombstone: {e}" ); } - PARSEABLE - .streams - .delete_if_still_deleting(stream_name, tenant_id); - if let Err(e) = stats::delete_stats(stream_name, "json", tenant_id) { - warn!("failed to clear stats for deleted stream {stream_name}: {e:?}"); - } } Err(e) => error!( "background deletion failed for {stream_name}: {e}. tombstone left in place, retried on next restart or repeat DELETE" @@ -1539,7 +1550,25 @@ pub fn sync_all_streams(joinset: &mut JoinSet>) { for stream_name in stream_names { match PARSEABLE.get_stream(&stream_name, &tenant_id) { Ok(stream) if !stream.is_deleting() => { - stream.mark_deleting(); + // list_tombstoned_streams above is a + // snapshot -- by the time we get + // here, the original deletion this + // name was tombstoned for could have + // already finished and the name been + // legitimately recreated (its tombstone + // gone, so the recreate wasn't even + // blocked). Re-check this one name + // live, right before acting, so a + // fresh stream can't be flagged + // deleting based on stale snapshot + // data with no tombstone left to ever + // clear it again. + if is_tombstoned(object_store.as_ref(), &stream_name, &tenant_id) + .await + .unwrap_or(false) + { + stream.mark_deleting(); + } } // Already flagged: the per-stream loop // above already owns reconciling this From c007bdab49317e7e857e2e01457e5c039be6f2ef Mon Sep 17 00:00:00 2001 From: prabhaks Date: Thu, 17 Sep 2026 14:05:58 -0700 Subject: [PATCH 12/14] fix: don't swallow errors from live tombstone recheck in sync_all_streams is_tombstoned().unwrap_or(false) treated a real error the same as "not tombstoned", silently leaving an actually-tombstoned stream unflagged and open to ingestion for a cycle. Match on the result instead so a failed check just retries next interval without masking the error. --- src/storage/object_storage.rs | 17 ++++++++++++++--- 1 file changed, 14 insertions(+), 3 deletions(-) diff --git a/src/storage/object_storage.rs b/src/storage/object_storage.rs index 4d83b81b6..5b09f218c 100644 --- a/src/storage/object_storage.rs +++ b/src/storage/object_storage.rs @@ -1563,11 +1563,22 @@ pub fn sync_all_streams(joinset: &mut JoinSet>) { // deleting based on stale snapshot // data with no tombstone left to ever // clear it again. - if is_tombstoned(object_store.as_ref(), &stream_name, &tenant_id) + match is_tombstoned(object_store.as_ref(), &stream_name, &tenant_id) .await - .unwrap_or(false) { - stream.mark_deleting(); + Ok(true) => stream.mark_deleting(), + Ok(false) => {} + // A real error here (as opposed to + // "no tombstone") is not evidence + // the tombstone is gone -- folding + // it into Ok(false) would leave a + // still-tombstoned stream + // unflagged and accepting writes. + // Leave it be and retry on the + // next sync interval instead. + Err(e) => warn!( + "failed to re-check tombstone for {stream_name} during reconciliation, retrying next sync interval: {e}" + ), } } // Already flagged: the per-stream loop From 186947c38529bbf997b8ab637dd3e46dda4b62a8 Mon Sep 17 00:00:00 2001 From: prabhaks Date: Thu, 17 Sep 2026 20:49:45 -0700 Subject: [PATCH 13/14] refactor: split sync_all_streams reconciliation into helper functions DeepSource flagged sync_all_streams at cyclomatic complexity 27 (very-high risk) after the tombstone reconciliation logic was added. Extract the two reconciliation passes into their own functions with no behavior change, to bring the top-level function's complexity back down. --- src/storage/object_storage.rs | 320 ++++++++++++++++++---------------- 1 file changed, 165 insertions(+), 155 deletions(-) diff --git a/src/storage/object_storage.rs b/src/storage/object_storage.rs index 5b09f218c..678b58d0c 100644 --- a/src/storage/object_storage.rs +++ b/src/storage/object_storage.rs @@ -1412,6 +1412,160 @@ pub fn spawn_stream_deletion(stream_name: String, tenant_id: Option) { }); } +// Extracted out of sync_all_streams to keep its own cyclomatic complexity +// down -- this owns just the "this node already knows the stream is +// deleting" branch. It received the DELETE handler's `for_each_live_node` +// push directly, or it's the owning node itself resuming its own in-flight +// job after a restart. A node that missed the fan-out push entirely (e.g. it +// was down or transiently unreachable at that exact moment) never gets +// is_deleting()=true set in the first place, so it never reaches this +// function at all -- that case is instead caught by +// spawn_tombstone_directory_reconciliation, which doesn't depend on the flag +// already being set locally. +fn spawn_deleting_stream_reconciliation( + joinset: &mut JoinSet>, + handle: &tokio::runtime::Handle, + object_store: Arc, + tenant_id: Option, + stream_name: String, +) { + // Only a node type that can actually receive a client's original DELETE + // request (query/standalone) ever resumes the physical delete here. An + // ingestor only ever gets is_deleting()=true via the delete handler's + // own fan-out push, which never starts a job on the ingestor itself -- + // letting it also spawn one here would mean every ingestor independently + // runs a redundant, uncoordinated bulk delete against the same prefix + // for the entire (possibly long) duration of every deletion, not just + // the rare case of a genuinely missed notification. + let is_deletion_owner = PARSEABLE.options.mode != Mode::Ingest; + joinset.spawn_on( + async move { + match is_tombstoned(object_store.as_ref(), &stream_name, &tenant_id).await { + Ok(true) => { + if is_deletion_owner { + spawn_stream_deletion(stream_name, tenant_id); + } + } + Ok(false) if !is_deletion_owner => { + // Deletion already finished elsewhere and the tombstone + // is gone, but this node's resident entry was never + // dropped -- e.g. an ingestor, which doesn't run the + // background deletion job itself. Reap it so a stream + // recreated under the same name doesn't inherit a stuck + // deleting=true state. + PARSEABLE + .streams + .delete_if_still_deleting(&stream_name, &tenant_id); + } + Ok(false) => { + // On the owning node, an absent tombstone can also just + // mean this delete hasn't written it yet (mark_deleting() + // runs first, with the tombstone write still in flight). + // Reaping here would race that window and let a + // concurrent reload resurrect the stream mid-deletion -- + // the owner's own spawn_stream_deletion job removes this + // entry once the actual delete (and tombstone clear) + // completes. + } + Err(e) => error!("failed to check tombstone status for {stream_name}: {e}"), + } + Ok(()) + }, + handle, + ); +} + +// Extracted out of sync_all_streams to keep its own cyclomatic complexity +// down -- this reconciles tombstones that the per-stream loop can't reach on +// its own, since it only acts on a stream whose local is_deleting() is +// already true: +// - A resident stream that's still flagged false because this node never +// ran mark_deleting() on it at all -- e.g. an ingestor that missed the +// DELETE handler's `for_each_live_node` fan-out push (a transient +// liveness-check failure at the wrong instant is enough), so it kept +// accepting ingestion into an already-deleted stream indefinitely. +// Flagging it here is just an in-memory update, so this runs on every +// node type, not only the owner. +// - A stream already gone from this node's memory entirely because its +// owning deletion job already finished, but whose final tombstone-clear +// call failed (e.g. a transient network error) -- only the owner should +// retry the physical delete for this case. +// A single tombstone-directory LIST finds every candidate at once, rather +// than a HEAD per resident stream every interval. +fn spawn_tombstone_directory_reconciliation( + joinset: &mut JoinSet>, + handle: &tokio::runtime::Handle, + object_store: Arc, + tenant_id: Option, +) { + let is_deletion_owner = PARSEABLE.options.mode != Mode::Ingest; + 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 { + reconcile_one_tombstoned_stream( + object_store.as_ref(), + &tenant_id, + stream_name, + is_deletion_owner, + ) + .await; + } + } + Err(e) => error!( + "failed to list tombstoned streams while syncing tenant {tenant_id:?}: {e}" + ), + } + Ok(()) + }, + handle, + ); +} + +async fn reconcile_one_tombstoned_stream( + object_store: &dyn ObjectStorage, + tenant_id: &Option, + stream_name: String, + is_deletion_owner: bool, +) { + match PARSEABLE.get_stream(&stream_name, tenant_id) { + Ok(stream) if !stream.is_deleting() => { + // list_tombstoned_streams is a snapshot -- by the time we get + // here, the original deletion this name was tombstoned for could + // have already finished and the name been legitimately recreated + // (its tombstone gone, so the recreate wasn't even blocked). + // Re-check this one name live, right before acting, so a fresh + // stream can't be flagged deleting based on stale snapshot data + // with no tombstone left to ever clear it again. + match is_tombstoned(object_store, &stream_name, tenant_id).await { + Ok(true) => stream.mark_deleting(), + Ok(false) => {} + // A real error here (as opposed to "no tombstone") is not + // evidence the tombstone is gone -- folding it into Ok(false) + // would leave a still-tombstoned stream unflagged and + // accepting writes. Leave it be and retry on the next sync + // interval instead. + Err(e) => warn!( + "failed to re-check tombstone for {stream_name} during reconciliation, retrying next sync interval: {e}" + ), + } + } + // Already flagged: the per-stream loop above already owns + // reconciling this one, nothing more to do here. + Ok(_) => {} + Err(_) if is_deletion_owner => { + // 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()); + } + // Not resident on a non-owner (e.g. an ingestor that never loaded + // this stream in the first place) -- nothing to reconcile. + Err(_) => {} + } +} + pub fn sync_all_streams(joinset: &mut JoinSet>) { let object_store = PARSEABLE.storage().get_object_store(); let tenants = if let Some(tenants) = PARSEABLE.list_tenants() { @@ -1424,74 +1578,12 @@ pub fn sync_all_streams(joinset: &mut JoinSet>) { for stream_name in PARSEABLE.streams.list(&tenant_id) { if let Ok(stream) = PARSEABLE.get_stream(&stream_name, &tenant_id) { if stream.is_deleting() { - // This handles a stream this node already knows is - // deleting -- either it received the DELETE handler's - // `for_each_live_node` push directly, or it's the owning - // node itself resuming its own in-flight job after a - // restart. A node that missed the fan-out push entirely - // (e.g. it was down or transiently unreachable at that - // exact moment) never gets is_deleting()=true set in the - // first place, so it never reaches this branch at all -- - // that case is instead caught by the tombstone-directory - // scan below, which doesn't depend on the flag already - // being set locally. - let object_store = object_store.clone(); - let tenant_id = tenant_id.clone(); - let stream_name = stream_name.clone(); - // Only a node type that can actually receive a client's - // original DELETE request (query/standalone) ever resumes - // the physical delete here. An ingestor only ever gets - // is_deleting()=true via the delete handler's own - // fan-out push, which never starts a job on the ingestor - // itself -- letting it also spawn one here would mean - // every ingestor independently runs a redundant, - // uncoordinated bulk delete against the same prefix for - // the entire (possibly long) duration of every deletion, - // not just the rare case of a genuinely missed - // notification. - let is_deletion_owner = PARSEABLE.options.mode != Mode::Ingest; - joinset.spawn_on( - async move { - match is_tombstoned(object_store.as_ref(), &stream_name, &tenant_id) - .await - { - Ok(true) => { - if is_deletion_owner { - spawn_stream_deletion(stream_name, tenant_id); - } - } - Ok(false) if !is_deletion_owner => { - // Deletion already finished elsewhere and - // the tombstone is gone, but this node's - // resident entry was never dropped -- e.g. - // an ingestor, which doesn't run the - // background deletion job itself. Reap it - // so a stream recreated under the same - // name doesn't inherit a stuck - // deleting=true state. - PARSEABLE - .streams - .delete_if_still_deleting(&stream_name, &tenant_id); - } - Ok(false) => { - // On the owning node, an absent tombstone - // can also just mean this delete hasn't - // written it yet (mark_deleting() runs - // first, with the tombstone write still - // in flight). Reaping here would race - // that window and let a concurrent reload - // resurrect the stream mid-deletion -- - // the owner's own spawn_stream_deletion - // job removes this entry once the actual - // delete (and tombstone clear) completes. - } - Err(e) => error!( - "failed to check tombstone status for {stream_name}: {e}" - ), - } - Ok(()) - }, + spawn_deleting_stream_reconciliation( + joinset, handle, + object_store.clone(), + tenant_id.clone(), + stream_name.clone(), ); continue; } @@ -1523,94 +1615,12 @@ pub fn sync_all_streams(joinset: &mut JoinSet>) { ); } - // Reconcile tombstones that the per-stream loop above can't reach on - // its own, since it only acts on a stream whose local is_deleting() - // is already true: - // - A resident stream that's still flagged false because this node - // never ran mark_deleting() on it at all -- e.g. an ingestor that - // missed the DELETE handler's `for_each_live_node` fan-out push - // (a transient liveness-check failure at the wrong instant is - // enough), so it kept accepting ingestion into an already-deleted - // stream indefinitely. Flagging it here is just an in-memory - // update, so this runs on every node type, not only the owner. - // - A stream already gone from this node's memory entirely because - // its owning deletion job already finished, but whose final - // tombstone-clear call failed (e.g. a transient network error) -- - // only the owner should retry the physical delete for this case. - // A single tombstone-directory LIST finds every candidate at once, - // rather than a HEAD per resident stream every interval. - { - let object_store = object_store.clone(); - let tenant_id = tenant_id.clone(); - let is_deletion_owner = PARSEABLE.options.mode != Mode::Ingest; - 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 { - match PARSEABLE.get_stream(&stream_name, &tenant_id) { - Ok(stream) if !stream.is_deleting() => { - // list_tombstoned_streams above is a - // snapshot -- by the time we get - // here, the original deletion this - // name was tombstoned for could have - // already finished and the name been - // legitimately recreated (its tombstone - // gone, so the recreate wasn't even - // blocked). Re-check this one name - // live, right before acting, so a - // fresh stream can't be flagged - // deleting based on stale snapshot - // data with no tombstone left to ever - // clear it again. - match is_tombstoned(object_store.as_ref(), &stream_name, &tenant_id) - .await - { - Ok(true) => stream.mark_deleting(), - Ok(false) => {} - // A real error here (as opposed to - // "no tombstone") is not evidence - // the tombstone is gone -- folding - // it into Ok(false) would leave a - // still-tombstoned stream - // unflagged and accepting writes. - // Leave it be and retry on the - // next sync interval instead. - Err(e) => warn!( - "failed to re-check tombstone for {stream_name} during reconciliation, retrying next sync interval: {e}" - ), - } - } - // Already flagged: the per-stream loop - // above already owns reconciling this - // one, nothing more to do here. - Ok(_) => {} - Err(_) if is_deletion_owner => { - // 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()); - } - // Not resident on a non-owner (e.g. an - // ingestor that never loaded this stream - // in the first place) -- nothing to - // reconcile. - Err(_) => {} - } - } - } - Err(e) => error!( - "failed to list tombstoned streams while syncing tenant {tenant_id:?}: {e}" - ), - } - Ok(()) - }, - handle, - ); - } + spawn_tombstone_directory_reconciliation( + joinset, + handle, + object_store.clone(), + tenant_id.clone(), + ); } } From 4403a62e8cb26dd1725da92aabd3e89335145565 Mon Sep 17 00:00:00 2001 From: prabhaks Date: Thu, 17 Sep 2026 21:00:52 -0700 Subject: [PATCH 14/14] fix: close two self-review gaps in standalone delete and update self-heal Standalone/all-in-one delete() had no tombstone-aware fallback for the narrow window where spawn_stream_deletion has already evicted the map entry but hasn't cleared the tombstone yet, unlike the query node's handler -- a retried DELETE landing there got a plain 404 instead of 202 "already in progress". Mirror the query node's check. reject_if_stream_deleting only self-healed a stale is_deleting flag on the create path; on the update path (e.g. a PUT reaching an ingestor whose resident copy was flagged by a delete fan-out push, with the tombstone since cleared elsewhere) the update would go through but the flag stayed set, needlessly rejecting ingestion for up to another sync interval. Clear it there too. Also add unit tests for delete_if_still_deleting and clear_deleting, which had no direct coverage despite delete_if_still_deleting being the core compare-and-remove guard the whole background-deletion cleanup order depends on. --- src/handlers/http/logstream.rs | 22 ++++++++++- src/parseable/mod.rs | 28 ++++++++++---- src/parseable/streams.rs | 69 ++++++++++++++++++++++++++++++++++ 3 files changed, 111 insertions(+), 8 deletions(-) diff --git a/src/handlers/http/logstream.rs b/src/handlers/http/logstream.rs index 32266f029..5afb5a244 100644 --- a/src/handlers/http/logstream.rs +++ b/src/handlers/http/logstream.rs @@ -30,7 +30,9 @@ use crate::stats::{Stats, event_labels_date, storage_size_labels_date}; use crate::storage::retention::Retention; use crate::storage::{ ObjectStoreFormat, StreamInfo, StreamType, - object_storage::{spawn_stream_deletion, stream_json_path, to_bytes, tombstone_path}, + object_storage::{ + is_tombstoned, spawn_stream_deletion, stream_json_path, to_bytes, tombstone_path, + }, }; use crate::tenants::TenantNotFound; use crate::utils::actix::extract_session_key_from_req; @@ -82,6 +84,24 @@ pub async fn delete( .check_or_load_stream(&stream_name, &tenant_id) .await { + // check_or_load_stream returning false is also what a retried + // DELETE sees once the background job has evicted this name + // from memory but hasn't cleared its tombstone yet (see + // spawn_stream_deletion's cleanup order) -- check the tombstone + // directly so that narrow window reports "already in progress" + // instead of a misleading "not found". + if is_tombstoned( + PARSEABLE.storage.get_object_store().as_ref(), + &stream_name, + &tenant_id, + ) + .await? + { + return Ok(( + format!("log stream {stream_name} deletion already in progress"), + StatusCode::ACCEPTED, + )); + } return Err(StreamNotFound(stream_name).into()); } diff --git a/src/parseable/mod.rs b/src/parseable/mod.rs index 58b5778fd..70b8337a8 100644 --- a/src/parseable/mod.rs +++ b/src/parseable/mod.rs @@ -821,18 +821,32 @@ impl Parseable { // could recreate the name while the background deletion job is // still sweeping its prefix, and that job would delete the freshly // created data too. - if !stream_in_memory_dont_update - && is_tombstoned( + if !stream_in_memory_dont_update { + if is_tombstoned( self.storage.get_object_store().as_ref(), stream_name, tenant_id, ) .await? - { - return Err(StreamError::Custom { - msg: format!("Logstream {stream_name} is being deleted, please retry shortly"), - status: StatusCode::CONFLICT, - }); + { + return Err(StreamError::Custom { + msg: format!("Logstream {stream_name} is being deleted, please retry shortly"), + status: StatusCode::CONFLICT, + }); + } + + // Self-heal a stale is_deleting flag on this path too (e.g. an + // ingestor's resident copy, flagged by the delete handler's + // fan-out push, whose tombstone has since cleared without + // sync_all_streams having noticed yet) -- otherwise this update + // succeeds but ingestion keeps being rejected here for up to + // another sync interval regardless. + if let Ok(stream) = self.get_stream(stream_name, tenant_id) + && stream.is_deleting() + { + stream.clear_deleting(); + self.streams.delete(stream_name, tenant_id); + } } Ok(stream_in_memory_dont_update) diff --git a/src/parseable/streams.rs b/src/parseable/streams.rs index 84a900ca5..408491fd3 100644 --- a/src/parseable/streams.rs +++ b/src/parseable/streams.rs @@ -2183,6 +2183,75 @@ mod tests { assert!(stream.is_deleting()); } + #[test] + fn test_clear_deleting_resets_is_deleting() { + let options = Arc::new(Options::default()); + let stream = Stream::new( + options, + "test_stream", + LogStreamMetadata::default(), + None, + &None, + ); + + stream.mark_deleting(); + assert!(stream.is_deleting()); + stream.clear_deleting(); + assert!(!stream.is_deleting()); + } + + fn insert_stream(streams: &Streams, stream_name: &str, deleting: bool) { + let options = Arc::new(Options::default()); + let stream = Stream::new( + options, + stream_name, + LogStreamMetadata::default(), + None, + &None, + ); + if deleting { + stream.mark_deleting(); + } + streams + .write() + .expect(LOCK_EXPECT) + .entry(DEFAULT_TENANT.to_string()) + .or_default() + .insert(stream_name.to_string(), stream); + } + + #[test] + fn delete_if_still_deleting_removes_a_stream_still_flagged_deleting() { + let streams = Streams::default(); + insert_stream(&streams, "doomed", true); + + streams.delete_if_still_deleting("doomed", &None); + + assert!(!streams.contains("doomed", &None)); + } + + #[test] + fn delete_if_still_deleting_leaves_a_recreated_stream_untouched() { + let streams = Streams::default(); + // Simulates a background deletion job finalizing by name after a + // concurrent self-heal already cleared the stale flag and inserted + // a fresh, non-deleting entry under the same name. + insert_stream(&streams, "recreated", false); + + streams.delete_if_still_deleting("recreated", &None); + + assert!(streams.contains("recreated", &None)); + } + + #[test] + fn delete_if_still_deleting_is_a_no_op_for_an_absent_stream() { + let streams = Streams::default(); + + streams.delete_if_still_deleting("never-existed", &None); + + assert!(!streams.contains("never-existed", &None)); + } + #[test] fn test_staging_with_special_characters() { let stream_name = "test_stream_!@#$%^&*()";