From 3754ff1e53edc56bd353fc728a9117e055905f90 Mon Sep 17 00:00:00 2001 From: VibeSWE <338779468+VibeSWE@users.noreply.github.com> Date: Tue, 10 Nov 2026 21:19:21 -0600 Subject: [PATCH 1/2] Route metadata lifecycle bypasses through Metastore --- docs/metastore-boundary-audit.md | 146 ++++++++++++++++ src/banner.rs | 2 +- src/handlers/http/health_check.rs | 8 +- src/handlers/http/logstream.rs | 8 +- .../http/modal/query/querier_logstream.rs | 7 +- src/metastore/metastore_traits.rs | 32 ++++ .../metastores/object_store_metastore.rs | 164 ++++++++++++++++++ src/metastore/mod.rs | 111 ++++++++++++ src/parseable/mod.rs | 39 +++-- src/storage/retention.rs | 9 +- 10 files changed, 493 insertions(+), 33 deletions(-) create mode 100644 docs/metastore-boundary-audit.md diff --git a/docs/metastore-boundary-audit.md b/docs/metastore-boundary-audit.md new file mode 100644 index 000000000..087ae05e2 --- /dev/null +++ b/docs/metastore-boundary-audit.md @@ -0,0 +1,146 @@ +# Mutable metadata boundary audit + +Baseline: origin/main `d6a155f2`. Scope: stage 0 seams only. No PostgreSQL +implementation, transactions, caching, serialization changes, or race fixes. + +A site below is a logical operation in a named symbol (several expressions in +one operation count once). Grouped backend implementations count once per row, +not once per provider. A includes reviewed operations already using the trait; +these are called out so an apparent ObjectStorage helper is not mistaken for a +remaining bypass. Test fixture writes are excluded from production counts. + +## Inventory + +Paths in the table are relative to `src/`. + +| File / symbol | Class | Objects or meaning; disposition | +| --- | --- | --- | +| `parseable/mod.rs::create_stream_and_schema_from_storage`, deletion check | A | `.tombstones///.tombstone` HEAD; now `Metastore::is_stream_deleting`. | +| `parseable/mod.rs::create_stream_and_schema_from_storage`, existence check | A | `.stream/*stream.json` discovery; now `Metastore::stream_exists`. Preserves LocalFS's canonical-only predicate and cloud's ingestor acceptance. | +| `parseable/mod.rs::load_tenants`, discovery | A | Root directory candidates standing for tenants; moved unchanged to `list_tenant_candidates`. Includes unrecognized directories for historical cleanup. | +| `parseable/mod.rs::load_tenants`, invalid directory cleanup | A | Whole tenant prefix, including any metadata; now existing `delete_tenant`. Same path and tenant metrics context (`None`). | +| `handlers/http/logstream.rs::delete` | A | Stream prefix, metadata and data together; now `Metastore::delete_stream`, which delegates to the identical provider operation. | +| `handlers/http/modal/query/querier_logstream.rs::delete` | A | Same deletion operation; same reroute. | +| `handlers/http/health_check.rs::readiness_internal` | A | Provider HEAD of `.parseable.json` (LocalFS has its own check); now `check_health`, preserving the provider implementation. | +| `banner.rs::storage_info` | A | Timed `.parseable.json` GET, including timing failed GETs; now `read_latency`, preserving the existing probe. | +| `storage/object_storage.rs::create_stream` | A | Schema then stream JSON; already calls `put_schema` then `put_stream_json`; unchanged. | +| `storage/object_storage.rs::update_time_partition_limit_in_stream` | A | Stream JSON read/overwrite through trait already; unchanged. | +| `storage/object_storage.rs::update_custom_partition_in_stream` | A | Same, custom partition; unchanged. | +| `storage/object_storage.rs::update_log_source_in_stream` | A | Same, log sources; unchanged. | +| `storage/object_storage.rs::update_dataset_tags_and_labels_in_stream` | A | Same, dataset tags/labels; unchanged. | +| `storage/object_storage.rs::update_first_event_in_stream` | A | Same, persisted first event timestamp; unchanged. | +| `storage/object_storage.rs::put_stats` | A | Same, whole stats replacement; unchanged. | +| `storage/object_storage.rs::put_retention` | A | Same, retention configuration; unchanged. | +| `storage/object_storage.rs::upsert_stream_metadata` | A | Mode-specific stream JSON or fallback base read then mode write; already trait. Retains even the original return of base bytes after writing modified config. | +| `storage/object_storage.rs::put_snapshot` | A | Upsert then snapshot replacement via trait; unchanged. | +| `storage/object_storage.rs::create_stream_from_querier` | A | Base stream JSON read, stats/snapshot reset, mode-specific write; already trait. | +| `storage/object_storage.rs::create_stream_from_ingestor` | A | Enumerate ingestor JSONs, merge sources/tags/labels, publish; already trait. | +| `storage/object_storage.rs::create_schema_from_metastore` | A | `fetch_schema` then `put_schema`; already trait. | +| `storage/object_storage.rs::get_log_source_from_storage` | A | Merge source descriptions from all stream JSONs; already trait. | +| `storage/object_storage.rs::commit_schema_to_storage` / `process_schema_files` | A | Schema read/merge/write then memory update and staging cleanup; already trait. No reordering. | +| `storage/store_metadata.rs::resolve_parseable_metadata` / `put_remote_metadata` | A | `.parseable.json`, RBAC, deployment and tenant configuration; already `get/put_parseable_metadata`. | +| `catalog/mod.rs::update_snapshot`, `handle_existing_partition`, `finalize_snapshot_update`, `remove_manifest_from_snapshot` | A | Stream snapshot, manifest reads/writes and stats; already trait (including `put_snapshot` helper). | +| `handlers/http/logstream.rs::{put_stream_hot_tier,delete_stream_hot_tier,get_stream_hot_tier_internal}` | A | Shared hot-tier enablement/config reads and whole stream JSON writes already use trait; local cache config is C below. | +| `storage/retention.rs::action::delete` metadata publication | A | Remove snapshot references and update deleted stats/first event through trait-backed helpers; unchanged. | +| `handlers/http/modal/mod.rs` node registration and HTTP modal node management | A | Node registry JSON; already `put/get/delete_node_metadata`; unchanged. | +| `migration/mod.rs::{run_migration,migration_stream,fetch_or_create_stream_metadata,migrate_stream_metadata,setup_logstream_metadata,put_remote_metadata}` | A | Discover streams, migrate JSON versions, publish schema/stream/deployment metadata; already trait, including helper calls. Retained this policy for all live migration reads/writes. Pure transforms in `metadata_migration.rs` and `stream_metadata_migration.rs` have no storage access. | +| `metastore/metastores/object_store_metastore.rs::{refresh_schema,get_all_schemas}` and `storage/object_storage.rs::schema_objects`, `storage/localfs.rs::schema_objects` | A | Schema enumeration/canonical union publication; raw storage only inside backend implementation/adapters. No application calls to `schema_objects`. | +| `storage/{s3,gcs,azure_blob,localfs}.rs::{stream_exists,list_streams,list_old_streams,get_ingestor_meta_file_paths,try_delete_node_meta,check}` and `object_storage.rs::{stream_exists,get_latency}` | A | Physical metadata predicates, node file enumeration/deletion and diagnostic probes. Provider primitives remain for backend use; active application calls now enter the trait. Legacy unused primitives noted below. | +| `metastore/metastores/object_store_metastore.rs` remaining metadata CRUD | A | Keystones, conversations, alerts, alert history, MTTR, LLM config, targets, outbound policy, dashboards (including conditional tombstones), chats, filters, schemas, manifests, deployment/tenant and node records. All raw object operations are backend implementation details already. | +| `storage/object_storage.rs::{upload_single_parquet_file,validate_uploaded_parquet_file}` | B | Multipart Parquet write, HEAD length validation, failed-upload object cleanup; data durability/size, not catalog authority. Left unchanged. | +| `storage/object_storage.rs::{get_first_and_latest_event_from_storage,extract_timestamp_for_date}` | B | Date/hour/minute directory scans implicitly stand for first/latest event timestamps with minute rounding. Future catalog bounds must preserve rounding. | +| `storage/object_storage.rs::{update_storage_metrics,upload_files_from_staging}` | B | Physical Parquet sizes/date names and publication pipeline. Metrics implicitly stand for ingestion/storage totals; metadata publication already enters trait. | +| `storage/retention.rs::action::delete` partition deletion | A | Mixed daily data/manifest prefix deletion now enters `delete_stream_partition`; the backend performs exactly the original prefix DELETE after snapshot publication, with the same concurrent per-date tasks. | +| `storage/retention.rs::action::delete` date listing | B | Directory names imply retention candidates; remains on storage. A future catalog/GC design must select catalog-native partitions instead. | +| `query/listing_table_builder.rs::ListingTableBuilder::populate_via_listing` | B | Partition prefixes imply legacy file membership; unchanged. | +| `query/stream_schema_provider.rs::{scan,legacy_listing_table,parquet_scan_plan}` | B | Legacy directory-derived membership and Parquet execution; modern metadata fetches already trait. File untouched, as requested. | +| `hottier.rs` download/planning and `hottier/planner.rs::build_work` | B | Remote Parquet fetches and directory-derived minute buckets imply cache work/time ranges; manifests already trait. | +| `parseable/mod.rs::validate_storage` root directory occupancy probe | B | Physical nonempty-store detection (even arbitrary nonmetadata directories affect startup). Metadata existence and stream listing already trait. Kept physical occupancy check; replacing it with catalog listing changes stale-storage detection. | +| `storage/{s3,gcs,azure_blob,localfs}.rs::{list_dates,list_hours,list_minutes,list_dirs_relative,delete_prefix}` | B | Data layout primitives, tenant scoping, metrics, retries and provider errors. Kept as physical adapters; metadata consumers confined to metastore. | +| `storage/field_stats.rs::calculate_field_stats` | B | Derived field statistics ingested as stream data, not a mutable JSON authority; file-derived timestamp remains. | +| `storage/store_metadata.rs::{get_staging_metadata,put_staging_metadata}` | C | Staging `.parseable.json` describes local deployment association; filesystem only, retained. | +| `migration/mod.rs::{get_staging_metadata,put_staging_metadata}` | C | Local migration staging copy; retained. | +| `parseable/streams.rs` schema staging, stream staging metadata and WAL | C | Local ingestion recovery state and schema files; retained. Shared schema commit already trait. | +| `hottier.rs::{read_hot_tier_config,put_hot_tier,delete_hot_tier}` and `hottier/local_state.rs` | C | Local `.hot_tier.json`, cache manifests/runtime state and local filesystem operations; retained. Shared hot-tier enablement/config in stream JSON already trait. | +| `storage/object_storage.rs::list_tombstoned_streams` and unused legacy provider discovery/node helpers | D | No production callers of tombstone discovery, `list_old_streams`, `get_ingestor_meta_file_paths`, or `try_delete_node_meta` outside backend/adapters. Retain dormant compatibility APIs and fixture tests. A future deletion recovery/legacy-layout import feature must expose semantic metastore methods before using them. | + +Counts: **33 A, 10 B, 4 C, 1 D** logical operation groups. Nine application +call sites changed; the other A groups already persisted through the trait or +implement its physical adapter. + +## Contract additions + +Existing methods and signatures are unchanged. PostgreSQL must implement: + +```rust +async fn check_health(&self, tenant_id: &Option) -> Result<(), MetastoreError>; +async fn read_latency(&self, tenant_id: &Option) -> std::time::Duration; +async fn stream_exists(&self, stream_name: &str, tenant_id: &Option) -> Result; +async fn is_stream_deleting(&self, stream_name: &str, tenant_id: &Option) -> Result; +async fn delete_stream(&self, stream_name: &str, tenant_id: &Option) -> Result<(), MetastoreError>; +async fn delete_stream_partition(&self, stream_name: &str, date: &str, tenant_id: &Option) -> Result<(), MetastoreError>; +async fn list_tenant_candidates(&self) -> Result, MetastoreError>; +``` + +`delete_stream` and `delete_stream_partition` include physical deletion: PostgreSQL must orchestrate catalog +and data cleanup, just as existing `delete_tenant` already includes both. This +seam introduces no transaction or new deletion protocol. `list_tenant_candidates` +preserves invalid candidates for cleanup; another backend may return only its +registered tenants. The legacy deletion-marker check remains separate from +existence, in the same order. It does not introduce marker publication. + +## Races deliberately preserved + +All stream-setting/stats/retention/first-event/snapshot updates read and blindly +replace the entire stream JSON. Schema commit and refresh merge schemas then +unconditionally publish; concurrent writers can lose fields. Manifest updates +and stream snapshot publication are separate unconditional writes; retention +can race ingestion, settings, stream deletion and queries. RBAC/deployment +metadata, alert state/history, filters and node registration have blind +whole-record writes or separate discovery/write/delete sequences. Tenant +validation/cleanup races concurrent tenant creation. Check-then-load and +check-then-delete remain racy. No CAS conversion, locks, retries, cache, or +transaction was added. Dashboard CAS behavior is unchanged. + +## Guard and evidence + +`metastore::boundary_tests` scans production Rust source before the trailing +`#[cfg(test)]` section. Outside the object-store metastore and four provider +adapters it rejects every raw `put_object`, `upload_file`, conditional PUT and +conditional DELETE, regardless of literal/computed key. It permits exactly one +multipart publisher in `object_storage.rs`, exactly one failed-Parquet DELETE +there, and no raw prefix DELETEs in application code. Tests verify +rejection of computed keys and whitespace-split method calls. This is a +conservative source policy, not runtime provenance or Rust type analysis. It +cannot detect mutations through new APIs, raw object_store clients, macros, +code after the first test section, or changes hidden in allowed adapters. The +existing allowed physical publishers still require review if their arguments +change. It does not prove read routing or transaction safety. + +Compatibility evidence: the nine application bypasses now delegate to exactly +the original provider operations. Provider errors are unwrapped again at the +rerouted legacy call sites, preserving HTTP error mapping and error text. JSON encoders, paths, payloads, write modes, +operation order and provider implementations are unchanged. The LocalFS +lifecycle fixture compares discovery and existence against raw provider results, +asserts exact JSON bytes survive probes, verifies the separate tombstone survives +stream prefix deletion, and compares missing-stream deletion errors. A second +fixture verifies partition deletion removes both Parquet and manifests while +preserving adjacent-date and other-tenant bytes. Existing +schema-refresh/manifest/dashboard/provider tests remain in the full library +suite. No live service or bucket was used; no container was started. No +main-versus-branch MinIO object diff was performed, so provider integration parity +is supported by unchanged delegates and unit tests, not an end-to-end claim. + +Open follow-up: catalog-native deletion/GC, imported legacy file membership, +first/latest timestamp semantics, and removal or semantic exposure of dormant +legacy APIs need later stages. Backend selection and PostgreSQL implementation +are intentionally outside this change. + +Final validation on this branch: + +- `P_BUILD_NO_UI=1 CARGO_BUILD_JOBS=4 cargo clippy --all-targets`: exit 0; + 21 existing library-test warnings, none in changed files. +- `P_BUILD_NO_UI=1 CARGO_BUILD_JOBS=4 cargo test --lib`: exit 0; + **1,379 passed, 0 failed, 4 ignored** (1,383 discovered), 79.65 seconds. +- Both source-guard tests and both LocalFS compatibility fixtures passed. +- `cargo fmt` and `git diff --check` passed. Cargo target remained in this worktree. diff --git a/src/banner.rs b/src/banner.rs index 26d3f3e23..53cf9dfac 100644 --- a/src/banner.rs +++ b/src/banner.rs @@ -88,7 +88,7 @@ fn status_info(config: &Parseable, scheme: &str, id: Uid) { /// - Store (path where the data is stored and its latency) async fn storage_info(config: &Parseable) { let storage = config.storage(); - let latency = storage.get_object_store().get_latency(&None).await; + let latency = config.metastore.read_latency(&None).await; eprintln!( " diff --git a/src/handlers/http/health_check.rs b/src/handlers/http/health_check.rs index 3f06d8c67..e834708fc 100644 --- a/src/handlers/http/health_check.rs +++ b/src/handlers/http/health_check.rs @@ -119,13 +119,7 @@ pub async fn readiness(req: HttpRequest) -> HttpResponse { pub async fn readiness_internal(tenant_id: &Option) -> StatusCode { // Check the object store connection - if PARSEABLE - .storage - .get_object_store() - .check(tenant_id) - .await - .is_ok() - { + if PARSEABLE.metastore.check_health(tenant_id).await.is_ok() { StatusCode::OK } else { StatusCode::SERVICE_UNAVAILABLE diff --git a/src/handlers/http/logstream.rs b/src/handlers/http/logstream.rs index e04a2a945..5b4e42987 100644 --- a/src/handlers/http/logstream.rs +++ b/src/handlers/http/logstream.rs @@ -66,10 +66,12 @@ pub async fn delete( return Err(StreamNotFound(stream_name).into()); } - let objectstore = PARSEABLE.storage.get_object_store(); - // Delete from storage - objectstore.delete_stream(&stream_name, &tenant_id).await?; + PARSEABLE + .metastore + .delete_stream(&stream_name, &tenant_id) + .await + .map_err(crate::metastore::MetastoreError::into_storage_error)?; // 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) { diff --git a/src/handlers/http/modal/query/querier_logstream.rs b/src/handlers/http/modal/query/querier_logstream.rs index 5226fbe07..2655c9254 100644 --- a/src/handlers/http/modal/query/querier_logstream.rs +++ b/src/handlers/http/modal/query/querier_logstream.rs @@ -78,9 +78,12 @@ pub async fn delete( return Err(StreamNotFound(stream_name.clone()).into()); } - let objectstore = PARSEABLE.storage.get_object_store(); // Delete from storage - objectstore.delete_stream(&stream_name, &tenant_id).await?; + PARSEABLE + .metastore + .delete_stream(&stream_name, &tenant_id) + .await + .map_err(crate::metastore::MetastoreError::into_storage_error)?; 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!( diff --git a/src/metastore/metastore_traits.rs b/src/metastore/metastore_traits.rs index 7de32cb04..fa484a51b 100644 --- a/src/metastore/metastore_traits.rs +++ b/src/metastore/metastore_traits.rs @@ -381,6 +381,38 @@ pub trait Metastore: std::fmt::Debug + Send + Sync { node_type: NodeType, ) -> Result; async fn put_node_metadata(&self, obj: &dyn MetastoreObject) -> Result<(), MetastoreError>; + /// Backend health probe; object storage preserves its historical HEAD check. + async fn check_health(&self, tenant_id: &Option) -> Result<(), MetastoreError>; + /// Time a metadata read, including failed reads, for the startup banner. + async fn read_latency(&self, tenant_id: &Option) -> std::time::Duration; + /// Preserve backend-specific stream existence rules without reading bucket metadata at callers. + async fn stream_exists( + &self, + stream_name: &str, + tenant_id: &Option, + ) -> Result; + /// Whether deletion has begun and lazy loading must be suppressed. + async fn is_stream_deleting( + &self, + stream_name: &str, + tenant_id: &Option, + ) -> Result; + /// Remove stream metadata and its data, preserving the backend's deletion order. + async fn delete_stream( + &self, + stream_name: &str, + tenant_id: &Option, + ) -> Result<(), MetastoreError>; + /// Delete a physical date partition together with its manifest metadata. + /// Snapshot removal is performed by the caller before this operation. + async fn delete_stream_partition( + &self, + stream_name: &str, + date: &str, + tenant_id: &Option, + ) -> Result<(), MetastoreError>; + /// Discovery candidates, including invalid entries which startup historically cleans up. + async fn list_tenant_candidates(&self) -> Result, MetastoreError>; async fn list_streams( &self, tenant_id: &Option, diff --git a/src/metastore/metastores/object_store_metastore.rs b/src/metastore/metastores/object_store_metastore.rs index 4a90a0ef1..c9682de58 100644 --- a/src/metastore/metastores/object_store_metastore.rs +++ b/src/metastore/metastores/object_store_metastore.rs @@ -1538,6 +1538,64 @@ impl Metastore for ObjectStoreMetastore { } } + async fn check_health(&self, tenant_id: &Option) -> Result<(), MetastoreError> { + Ok(self.storage.check(tenant_id).await?) + } + + async fn read_latency(&self, tenant_id: &Option) -> std::time::Duration { + self.storage.get_latency(tenant_id).await + } + + async fn stream_exists( + &self, + stream_name: &str, + tenant_id: &Option, + ) -> Result { + Ok(self.storage.stream_exists(stream_name, tenant_id).await?) + } + + async fn is_stream_deleting( + &self, + stream_name: &str, + tenant_id: &Option, + ) -> Result { + Ok(crate::storage::object_storage::is_tombstoned( + self.storage.as_ref(), + stream_name, + tenant_id, + ) + .await?) + } + + async fn delete_stream( + &self, + stream_name: &str, + tenant_id: &Option, + ) -> Result<(), MetastoreError> { + Ok(self.storage.delete_stream(stream_name, tenant_id).await?) + } + + async fn delete_stream_partition( + &self, + stream_name: &str, + date: &str, + tenant_id: &Option, + ) -> Result<(), MetastoreError> { + let path = + crate::storage::object_storage::stream_data_path(stream_name, &[date], tenant_id); + Ok(self.storage.delete_prefix(&path, tenant_id).await?) + } + + async fn list_tenant_candidates(&self) -> Result, MetastoreError> { + Ok(self + .storage + .list_dirs_relative(&RelativePathBuf::from_iter([""]), &None) + .await? + .into_iter() + .filter(|d| !d.starts_with('.')) + .collect()) + } + async fn list_streams( &self, tenant_id: &Option, @@ -2099,3 +2157,109 @@ mod tests { )); } } + +#[cfg(test)] +mod boundary_compatibility_tests { + use super::*; + use crate::storage::{LocalFS, object_storage::tombstone_path}; + + #[tokio::test] + async fn partition_deletion_preserves_tenant_scope_and_adjacent_bytes() { + let dir = tempfile::tempdir().unwrap(); + let storage = Arc::new(LocalFS::new(dir.path().to_owned())); + let metastore = ObjectStoreMetastore { + storage: storage.clone(), + }; + let tenant = Some("tenant".to_owned()); + let keys = [ + "tenant/logs/date=2026-01-01/hour=00/data.parquet", + "tenant/logs/date=2026-01-01/ingestor.node.manifest.json", + "tenant/logs/date=2026-01-02/ingestor.node.manifest.json", + "logs/date=2026-01-01/ingestor.node.manifest.json", + ]; + for key in keys { + storage + .put_object( + RelativePath::new(key), + Bytes::from_static(b"original bytes"), + &tenant, + ) + .await + .unwrap(); + } + metastore + .delete_stream_partition("logs", "date=2026-01-01", &tenant) + .await + .unwrap(); + for key in &keys[..2] { + assert!( + storage + .get_object(RelativePath::new(key), &tenant) + .await + .is_err() + ); + } + for key in &keys[2..] { + assert_eq!( + storage + .get_object(RelativePath::new(key), &tenant) + .await + .unwrap(), + Bytes::from_static(b"original bytes") + ); + } + } + + #[tokio::test] + async fn lifecycle_seams_preserve_localfs_results_and_bytes() { + let dir = tempfile::tempdir().unwrap(); + let storage = Arc::new(LocalFS::new(dir.path().to_owned())); + let metastore = ObjectStoreMetastore { + storage: storage.clone(), + }; + assert!(!metastore.stream_exists("logs", &None).await.unwrap()); + assert!(!metastore.is_stream_deleting("logs", &None).await.unwrap()); + let key = RelativePath::new("logs/.stream/.stream.json"); + let bytes = Bytes::from_static(b"{\"version\":\"v7\",\"unknown\":true}"); + storage.put_object(key, bytes.clone(), &None).await.unwrap(); + assert_eq!( + metastore.stream_exists("logs", &None).await.unwrap(), + storage.stream_exists("logs", &None).await.unwrap() + ); + // Discovery/existence probes never normalize or rewrite stored JSON. + assert_eq!(storage.get_object(key, &None).await.unwrap(), bytes); + storage + .put_object( + &tombstone_path("logs", &None), + Bytes::from_static(b"null"), + &None, + ) + .await + .unwrap(); + assert!(metastore.is_stream_deleting("logs", &None).await.unwrap()); + let mut expected: Vec<_> = storage + .list_dirs_relative(RelativePath::new(""), &None) + .await + .unwrap() + .into_iter() + .filter(|d| !d.starts_with('.')) + .collect(); + let mut actual = metastore.list_tenant_candidates().await.unwrap(); + expected.sort(); + actual.sort(); + assert_eq!(actual, expected); + metastore.delete_stream("logs", &None).await.unwrap(); + assert!(!metastore.stream_exists("logs", &None).await.unwrap()); + // Historical prefix deletion leaves the separate marker untouched. + assert!(metastore.is_stream_deleting("logs", &None).await.unwrap()); + let raw_error = storage + .delete_stream("missing", &None) + .await + .unwrap_err() + .to_string(); + let trait_error = metastore.delete_stream("missing", &None).await.unwrap_err(); + assert!( + matches!(trait_error, MetastoreError::ObjectStorageError(error) if error.to_string() == raw_error) + ); + } +} diff --git a/src/metastore/mod.rs b/src/metastore/mod.rs index acf7b7a97..aaf086657 100644 --- a/src/metastore/mod.rs +++ b/src/metastore/mod.rs @@ -67,6 +67,14 @@ pub enum MetastoreError { } impl MetastoreError { + /// Preserve provider errors at pre-existing ObjectStorage call sites. + pub(crate) fn into_storage_error(self) -> ObjectStorageError { + match self { + Self::ObjectStorageError(error) => error, + error => ObjectStorageError::MetastoreError(Box::new(error.to_detail())), + } + } + pub fn to_detail(&self) -> MetastoreErrorDetail { match self { MetastoreError::Error { @@ -171,3 +179,106 @@ impl MetastoreError { } } } + +#[cfg(test)] +mod boundary_tests { + use std::path::Path; + + // Deliberately conservative: applications cannot issue any raw object PUT, + // regardless of whether the key is a literal, a constant, or computed. + // Provider adapters implement the primitives; only the metastore may use + // them for metadata. The one multipart publisher uploads Parquet data. + fn violations(path: &str, source: &str) -> Vec { + if matches!( + path, + "storage/s3.rs" + | "storage/gcs.rs" + | "storage/azure_blob.rs" + | "storage/localfs.rs" + | "metastore/metastores/object_store_metastore.rs" + ) { + return vec![]; + } + let production = source.split("#[cfg(test)]").next().unwrap(); + let compact: String = production.chars().filter(|c| !c.is_whitespace()).collect(); + let mut errors = Vec::new(); + for operation in [ + ".put_object(", + ".upload_file(", + ".put_metadata_conditional(", + ".delete_metadata_conditional(", + ] { + if compact.contains(operation) { + errors.push(format!( + "{path}: raw {operation} must enter through Metastore" + )); + } + } + for (operation, allowed_path) in [ + (".delete_object(", "storage/object_storage.rs"), + (".delete_prefix(", ""), + ] { + let count = compact.matches(operation).count(); + if count != usize::from(path == allowed_path) { + errors.push(format!("{path}: unexpected raw {operation} ({count})")); + } + } + let multipart = compact.matches(".upload_multipart(").count(); + let allowed = usize::from(path == "storage/object_storage.rs"); + if multipart != allowed { + errors.push(format!( + "{path}: unexpected multipart publisher ({multipart})" + )); + } + errors + } + + fn inspect(root: &Path, path: &Path, errors: &mut Vec) { + for entry in std::fs::read_dir(path).unwrap() { + let path = entry.unwrap().path(); + if path.is_dir() { + inspect(root, &path, errors); + } else if path.extension().is_some_and(|ext| ext == "rs") { + let relative = path.strip_prefix(root).unwrap().to_str().unwrap(); + errors.extend(violations( + relative, + &std::fs::read_to_string(&path).unwrap(), + )); + } + } + } + + #[test] + fn metadata_writes_enter_through_metastore() { + let root = Path::new(env!("CARGO_MANIFEST_DIR")).join("src"); + let mut errors = Vec::new(); + inspect(&root, &root, &mut errors); + assert!(errors.is_empty(), "{}", errors.join("\n")); + } + + #[test] + fn guard_rejects_literal_and_computed_metadata_writes() { + for source in [ + "store.put_object(stream_json_path(name, tenant), bytes, tenant).await", + "store\n . put_object (path, bytes, tenant).await", + "store.put_metadata_conditional(path, bytes, condition, tenant).await", + "store.upload_file(key, local, tenant).await", + ] { + assert!(!violations("application.rs", source).is_empty()); + } + assert!( + violations( + "application.rs", + "#[cfg(test)] mod tests { store.put_object(path); }" + ) + .is_empty() + ); + assert!( + violations( + "metastore/metastores/object_store_metastore.rs", + "store.put_object(path)" + ) + .is_empty() + ); + } +} diff --git a/src/parseable/mod.rs b/src/parseable/mod.rs index 608e25f02..363e3360e 100644 --- a/src/parseable/mod.rs +++ b/src/parseable/mod.rs @@ -31,7 +31,6 @@ use chrono::Utc; use clap::{Parser, error::ErrorKind}; use once_cell::sync::{Lazy, OnceCell}; use regex::Regex; -use relative_path::RelativePathBuf; pub use staging::StagingError; use streams::StreamRef; pub use streams::{Stream, StreamNotFound, Streams}; @@ -72,8 +71,7 @@ use crate::{ static_schema::{StaticSchema, convert_static_schema_to_arrow_schema}, storage::{ ObjectStorage, ObjectStorageError, ObjectStorageProvider, ObjectStoreFormat, Owner, - Permisssion, StorageMetadata, StreamType, object_storage::is_tombstoned, - put_remote_metadata, + Permisssion, StorageMetadata, StreamType, put_remote_metadata, }, tenants::{Service, TENANT_METADATA}, validator, @@ -498,10 +496,20 @@ impl Parseable { 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? { + if self + .metastore + .is_stream_deleting(stream_name, tenant_id) + .await + .map_err(crate::metastore::MetastoreError::into_storage_error)? + { return Ok(false); } - if !storage.stream_exists(stream_name, tenant_id).await? { + if !self + .metastore + .stream_exists(stream_name, tenant_id) + .await + .map_err(crate::metastore::MetastoreError::into_storage_error)? + { return Ok(false); } let (stream_metadata_bytes, schema_bytes) = try_join!( @@ -1235,13 +1243,11 @@ impl Parseable { async fn load_tenants(&self) -> Result<(), anyhow::Error> { let is_multi_tenant = self.options.is_multi_tenant(); - let obj_store = self.storage().get_object_store(); - let dirs: Vec = obj_store - .list_dirs_relative(&RelativePathBuf::from_iter([""]), &None) - .await? - .into_iter() - .filter(|d| !d.starts_with(".")) - .collect(); + let dirs = self + .metastore + .list_tenant_candidates() + .await + .map_err(crate::metastore::MetastoreError::into_storage_error)?; let mut recognized_tenants = vec![]; @@ -1267,9 +1273,11 @@ impl Parseable { ); // delete from storage - if let Err(e) = obj_store - .delete_prefix(&RelativePathBuf::from_iter([&tenant_id]), &None) + if let Err(e) = self + .metastore + .delete_tenant(&tenant_id) .await + .map_err(crate::metastore::MetastoreError::into_storage_error) { tracing::error!( "Unable to delete invalid directory- {tenant_id} with error-\n{e}" @@ -1432,7 +1440,8 @@ mod tests { storage.clone(), ); let stream = server.get_or_create_stream("logs", &None); - let remote_path = RelativePathBuf::from("logs/.stream/.ingestor.other.schema"); + let remote_path = + relative_path::RelativePathBuf::from("logs/.stream/.ingestor.other.schema"); storage .put_object( &remote_path, diff --git a/src/storage/retention.rs b/src/storage/retention.rs index a94be906e..8f88f7044 100644 --- a/src/storage/retention.rs +++ b/src/storage/retention.rs @@ -210,7 +210,6 @@ mod action { use crate::catalog::{ingestor_cleanup_reaches_tenant, remove_manifest_from_snapshot}; use crate::option::Mode; use crate::parseable::PARSEABLE; - use crate::storage::object_storage::stream_data_path; use crate::tenants::TENANT_METADATA; use chrono::{Days, NaiveDate, Utc}; use futures::{StreamExt, stream::FuturesUnordered}; @@ -266,13 +265,13 @@ mod action { for date in dates_to_delete { // The tenant's own data, not a same-named stream of the default tenant. - let path = stream_data_path(&stream_name, &[&date], tenant_id); + let stream_name = &stream_name; delete_tasks.push(async move { PARSEABLE - .storage - .get_object_store() - .delete_prefix(&path, tenant_id) + .metastore + .delete_stream_partition(stream_name, &date, tenant_id) .await + .map_err(crate::metastore::MetastoreError::into_storage_error) }); } From 7544a706d3a1ec2c8110c1dca65e56163dbb950b Mon Sep 17 00:00:00 2001 From: VibeSWE <338779468+VibeSWE@users.noreply.github.com> Date: Wed, 11 Nov 2026 13:34:21 -0600 Subject: [PATCH 2/2] ci: build Quest from a pinned source commit The floating quay.io/parseablehq/quest:main image now expects 202 from stream deletion. This server still deletes synchronously and returns 200, so TestSmokeRbacBasic and TestSmokeRoles fail on main and on every pull request. The previous image digest is no longer available from the registry, so build Quest from the last source commit before that change. The compose files take the image and pull policy from QUEST_IMAGE and QUEST_PULL_POLICY and keep their previous defaults. Refs #151 Co-Authored-By: Claude Fable 5.2 --- .github/workflows/integration-test.yaml | 15 +++++++++++++++ docker-compose-distributed-test-with-kafka.yaml | 4 ++-- docker-compose-distributed-test.yaml | 4 ++-- docker-compose-gcs-distributed-test.yaml | 4 ++-- docker-compose-test-with-kafka.yaml | 4 ++-- docker-compose-test.yaml | 4 ++-- 6 files changed, 25 insertions(+), 10 deletions(-) diff --git a/.github/workflows/integration-test.yaml b/.github/workflows/integration-test.yaml index 123496873..9c4997522 100644 --- a/.github/workflows/integration-test.yaml +++ b/.github/workflows/integration-test.yaml @@ -67,6 +67,13 @@ jobs: PARSEABLE_TEST_IMAGE: parseable-ci:${{ github.sha }} COMPOSE_FILE: ${{ matrix.compose }} QUEST_MODE: ${{ inputs.full && 'load' || 'smoke' }} + # Quest is built from a fixed source commit instead of pulling the + # floating quay.io/parseablehq/quest:main image. Later Quest commits + # expect 202 from stream deletion, which this server does not return yet. + # Remove the pin when issue #151 is done. + QUEST_SOURCE_REF: f4a026eecde1badde8776ff366a743f9bbca2e67 + QUEST_IMAGE: quest-ci:pinned + QUEST_PULL_POLICY: never steps: - uses: actions/checkout@v4 - uses: actions/download-artifact@v4 @@ -84,6 +91,14 @@ jobs: run: | docker compose up --no-build --wait --wait-timeout 180 "$SERVER" docker compose exec -T "$SERVER" sh -ec 'curl --fail --silent --show-error --location http://localhost:8000/next/ | grep -q "Parseable ยท Observability"' + - name: Build Quest from the pinned source commit + run: | + src="$RUNNER_TEMP/quest-src" + git init --quiet "$src" + git -C "$src" fetch --quiet --depth 1 https://github.com/parseablehq/quest.git "$QUEST_SOURCE_REF" + git -C "$src" checkout --quiet FETCH_HEAD + test "$(git -C "$src" rev-parse HEAD)" = "$QUEST_SOURCE_REF" + docker build --tag "$QUEST_IMAGE" "$src" - name: Run Quest run: docker compose up --no-build --exit-code-from quest - name: Capture service logs diff --git a/docker-compose-distributed-test-with-kafka.yaml b/docker-compose-distributed-test-with-kafka.yaml index f173e22b0..1e22cf39c 100644 --- a/docker-compose-distributed-test-with-kafka.yaml +++ b/docker-compose-distributed-test-with-kafka.yaml @@ -194,8 +194,8 @@ services: quest: platform: linux/amd64 - image: quay.io/parseablehq/quest:main - pull_policy: always + image: ${QUEST_IMAGE:-quay.io/parseablehq/quest:main} + pull_policy: ${QUEST_PULL_POLICY:-always} command: [ "load", diff --git a/docker-compose-distributed-test.yaml b/docker-compose-distributed-test.yaml index b46810649..c3bcdfc4a 100644 --- a/docker-compose-distributed-test.yaml +++ b/docker-compose-distributed-test.yaml @@ -109,8 +109,8 @@ services: quest: platform: linux/amd64 - image: quay.io/parseablehq/quest:main - pull_policy: always + image: ${QUEST_IMAGE:-quay.io/parseablehq/quest:main} + pull_policy: ${QUEST_PULL_POLICY:-always} environment: - QUEST_EDITION=oss command: diff --git a/docker-compose-gcs-distributed-test.yaml b/docker-compose-gcs-distributed-test.yaml index a90e53e2f..66bf18f69 100644 --- a/docker-compose-gcs-distributed-test.yaml +++ b/docker-compose-gcs-distributed-test.yaml @@ -79,8 +79,8 @@ services: quest: platform: linux/amd64 - image: quay.io/parseablehq/quest:main - pull_policy: always + image: ${QUEST_IMAGE:-quay.io/parseablehq/quest:main} + pull_policy: ${QUEST_PULL_POLICY:-always} command: [ "load", diff --git a/docker-compose-test-with-kafka.yaml b/docker-compose-test-with-kafka.yaml index ff8f09f2e..79d337e59 100644 --- a/docker-compose-test-with-kafka.yaml +++ b/docker-compose-test-with-kafka.yaml @@ -68,9 +68,9 @@ services: max_attempts: 3 quest: - image: quay.io/parseablehq/quest:main + image: ${QUEST_IMAGE:-quay.io/parseablehq/quest:main} platform: linux/amd64 - pull_policy: always + pull_policy: ${QUEST_PULL_POLICY:-always} command: [ "load", "http://parseable:8000", diff --git a/docker-compose-test.yaml b/docker-compose-test.yaml index ddef891fb..5fc46d7aa 100644 --- a/docker-compose-test.yaml +++ b/docker-compose-test.yaml @@ -64,9 +64,9 @@ services: max_attempts: 3 quest: - image: quay.io/parseablehq/quest:main + image: ${QUEST_IMAGE:-quay.io/parseablehq/quest:main} platform: linux/amd64 - pull_policy: always + pull_policy: ${QUEST_PULL_POLICY:-always} environment: - QUEST_EDITION=oss command: [