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: [ 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) }); }