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..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 { @@ -1352,6 +1356,17 @@ impl Stream { self.metadata.read().expect(LOCK_EXPECT).hot_tier_enabled } + /// Marks this stream as being deleted. Once set, this flag is never + /// cleared for this in-memory entry — a deletion in progress runs to + /// completion (or is resumed on restart), it is never cancelled. + pub fn mark_deleting(&self) { + self.metadata.write().expect(LOCK_EXPECT).deleting = true; + } + + pub fn is_deleting(&self) -> bool { + self.metadata.read().expect(LOCK_EXPECT).deleting + } + pub fn get_stream_type(&self) -> StreamType { self.metadata.read().expect(LOCK_EXPECT).stream_type } @@ -1744,6 +1759,22 @@ mod tests { ); } + #[test] + fn test_mark_deleting_sets_is_deleting() { + let options = Arc::new(Options::default()); + let stream = Stream::new( + options, + "test_stream", + LogStreamMetadata::default(), + None, + &None, + ); + + assert!(!stream.is_deleting()); + stream.mark_deleting(); + assert!(stream.is_deleting()); + } + #[test] fn test_staging_with_special_characters() { let stream_name = "test_stream_!@#$%^&*()"; diff --git a/src/storage/localfs.rs b/src/storage/localfs.rs index bafd80404..d66714023 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; @@ -570,6 +572,7 @@ impl ObjectStorage for LocalFS { PARSEABLE_ROOT_DIRECTORY, ALERTS_ROOT_DIRECTORY, SETTINGS_ROOT_DIRECTORY, + TOMBSTONE_ROOT_DIRECTORY, ]; let result = fs::read_dir(&self.root).await; diff --git a/src/storage/mod.rs b/src/storage/mod.rs index 5c6a2e36c..6271f14cf 100644 --- a/src/storage/mod.rs +++ b/src/storage/mod.rs @@ -310,6 +310,17 @@ pub const ALERTS_ROOT_DIRECTORY: &str = ".alerts"; pub const SETTINGS_ROOT_DIRECTORY: &str = ".settings"; pub const TARGETS_ROOT_DIRECTORY: &str = ".targets"; pub const MANIFEST_FILE: &str = "manifest.json"; +// top-level registry of streams currently being deleted; kept outside every +// stream's own prefix so a bulk prefix-delete can never sweep up a marker +// that's supposed to survive it (see is_tombstoned/tombstone_path) +pub const TOMBSTONE_ROOT_DIRECTORY: &str = ".tombstones"; +// the marker itself lives one level below `{tenant}/{stream_name}/`, not as +// a leaf key directly named after the stream: list_dirs_relative on every +// backend (S3/GCS/Azure via list-with-delimiter's common_prefixes, LocalFS +// via read_dir + is_dir) only surfaces child *directories*, never leaf +// objects, so a tombstone recorded as a bare `{stream_name}` key would be +// invisible to the restart-recovery scan that discovers tombstoned streams +pub const TOMBSTONE_MARKER_FILE_NAME: &str = ".tombstone"; // max concurrent request allowed for datafusion object store, overridable per // backend with P_MAX_OBJECT_STORE_REQUESTS. diff --git a/src/storage/object_storage.rs b/src/storage/object_storage.rs index 38b2809be..2229bd503 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_MARKER_FILE_NAME, + TOMBSTONE_ROOT_DIRECTORY, retention::Retention, }; /// Context for upload operations containing stream information @@ -1442,6 +1443,81 @@ 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. +/// +/// 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, + TOMBSTONE_MARKER_FILE_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), + } +} + +/// 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. +/// +/// `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]); + 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 #[inline(always)] pub fn filter_path( @@ -1567,6 +1643,88 @@ 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, list_tombstoned_streams, 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()); + } + + #[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()]); + } + + #[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)] mod manifest_ownership_tests { use super::manifest_segment_matches;