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/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/logstream.rs b/src/handlers/http/logstream.rs index ab6fd1a8b..5afb5a244 100644 --- a/src/handlers/http/logstream.rs +++ b/src/handlers/http/logstream.rs @@ -28,7 +28,12 @@ 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::{ + 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; use crate::utils::get_tenant_id_from_request; @@ -47,8 +52,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, @@ -56,41 +69,118 @@ 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) + + // 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 = { + // 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 + { + // 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()); + } + + 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()); + } + + stream + }; + + 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() + // 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 { - return Err(StreamNotFound(stream_name).into()); + warn!( + "failed to eagerly delete stream.json for {stream_name}, will be removed with the rest of the prefix: {e}" + ); } - let objectstore = PARSEABLE.storage.get_object_store(); - - // Delete from storage - objectstore.delete_stream(&stream_name, &tenant_id).await?; // 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() ) } + // 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}"); } - // 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)); + // 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} deleted"), StatusCode::OK)) + Ok(( + format!("log stream {stream_name} deletion started"), + StatusCode::ACCEPTED, + )) } pub async fn list(req: HttpRequest) -> Result { @@ -209,6 +299,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?; @@ -232,10 +323,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)) } @@ -256,15 +349,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}"), @@ -456,6 +552,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(), @@ -529,6 +629,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)); }; @@ -564,11 +671,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/ingest/ingestor_logstream.rs b/src/handlers/http/modal/ingest/ingestor_logstream.rs index 9f7414baa..b11f5c50c 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; @@ -31,10 +32,15 @@ use crate::{ catalog::remove_manifest_from_snapshot, handlers::http::logstream::error::StreamError, parseable::{PARSEABLE, StreamNotFound}, - stats, 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, @@ -76,8 +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)?; + + 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 @@ -91,12 +103,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::ACCEPTED, + )) } pub async fn put_stream( @@ -106,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 2bb170104..5a8628176 100644 --- a/src/handlers/http/modal/query/querier_logstream.rs +++ b/src/handlers/http/modal/query/querier_logstream.rs @@ -45,12 +45,16 @@ 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::{ + is_tombstoned, spawn_stream_deletion, stream_json_path, to_bytes, tombstone_path, + }, + }, utils::get_tenant_id_from_request, }; const STATS_DATE_QUERY_PARAM: &str = "date"; @@ -61,64 +65,151 @@ 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) + + // 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 = { + // 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) + { + // 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. 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? + { + return Ok(( + format!("log stream {stream_name} deletion already in progress"), + StatusCode::ACCEPTED, + )); + } + return Err(StreamNotFound(stream_name.clone()).into()); + } + + 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 - .unwrap_or(false) + { + // 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. + // 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 { - return Err(StreamNotFound(stream_name.clone()).into()); + warn!( + "failed to eagerly delete stream.json for {stream_name}, will be removed with the rest of the prefix: {e}" + ); } - 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) { + // 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 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!( + "{}{}/logstream/{}/sync", + node.domain_name, + base_path_without_preceding_slash(), + fanout_stream_name + ); + async move { cluster::send_stream_delete_request(&url, node).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!( "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() ) } 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?; - } - - 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 - })?; - - for ingestor in ingestor_metadata { - let url = format!( - "{}{}/logstream/{}/sync", - ingestor.domain_name, - base_path_without_preceding_slash(), - stream_name - ); - - // delete the stream - cluster::send_stream_delete_request(&url, ingestor.clone()).await?; + { + warn!("failed to delete hot tier for stream {stream_name}: {e}"); } - // 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)); + // 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} deleted"), StatusCode::OK)) + Ok(( + format!("log stream {stream_name} deletion started"), + StatusCode::ACCEPTED, + )) } pub async fn put_stream( @@ -186,6 +277,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/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 9d2d05de4..70b8337a8 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(), "", @@ -742,6 +758,100 @@ impl Parseable { Ok(()) } + /// 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, + stream_name: &str, + tenant_id: &Option, + 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 + // 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() + { + // 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(); + + // 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 + // 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 { + 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, + }); + } + + // 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) + } + pub async fn create_update_stream( &self, headers: &HeaderMap, @@ -794,6 +904,9 @@ impl Parseable { 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 diff --git a/src/parseable/streams.rs b/src/parseable/streams.rs index 0e3801f5a..408491fd3 100644 --- a/src/parseable/streams.rs +++ b/src/parseable/streams.rs @@ -1632,6 +1632,17 @@ impl Stream { self.metadata.write().expect(LOCK_EXPECT).deleting = true; } + /// 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 while the tombstone still 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 } @@ -2009,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 { @@ -2154,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_!@#$%^&*()"; diff --git a/src/storage/localfs.rs b/src/storage/localfs.rs index d66714023..5bae64b19 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::{ @@ -495,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", @@ -555,7 +565,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?; @@ -863,6 +873,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() @@ -886,6 +897,21 @@ 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 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(); @@ -916,3 +942,176 @@ 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()); + } + + #[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 c7eb3f208..678b58d0c 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; @@ -1333,6 +1335,237 @@ 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(()) => { + // 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 + { + warn!( + "background deletion of {stream_name} finished but failed to clear its tombstone: {e}" + ); + } + } + Err(e) => error!( + "background deletion failed for {stream_name}: {e}. tombstone left in place, retried on next restart or repeat DELETE" + ), + } + }); +} + +// 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() { @@ -1343,11 +1576,20 @@ 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() { + spawn_deleting_stream_reconciliation( + joinset, + handle, + object_store.clone(), + tenant_id.clone(), + stream_name.clone(), + ); + continue; + } + if stream.parquet_files().is_empty() && stream.schema_files().is_empty() { + continue; + } } let object_store = object_store.clone(); let id = tenant_id.clone(); @@ -1372,6 +1614,13 @@ pub fn sync_all_streams(joinset: &mut JoinSet>) { handle, ); } + + spawn_tombstone_directory_reconciliation( + joinset, + handle, + object_store.clone(), + tenant_id.clone(), + ); } } @@ -1723,6 +1972,48 @@ 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 same + // map the real function uses. The real function's actual dedup guard is + // an atomic `entry()` check-and-insert; a plain contains_key-then-insert + // here is fine since these tests are single-threaded and only need to + // assert the map's observable state, not re-prove atomicity. + #[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;