diff --git a/crates/persisting-pchronicle-cli/src/gateway_capture.rs b/crates/persisting-pchronicle-cli/src/gateway_capture.rs index 8dc1deca..4ff5fc8b 100644 --- a/crates/persisting-pchronicle-cli/src/gateway_capture.rs +++ b/crates/persisting-pchronicle-cli/src/gateway_capture.rs @@ -5,11 +5,9 @@ use std::sync::Arc; use persisting_gateway::record::EventRecord; use persisting_gateway::session::storage::CaptureRoute; use persisting_gateway::sink::{CallbackSink, CaptureEventSink}; -#[cfg(test)] -use persisting_pchronicle::storage::raw_event_append_queue; use persisting_pchronicle::storage::{ - ObjectStoreManifestWriteMode, RawEventAppendOutcome, RawEventAppendSender, - RawEventAppendWorker, StoryCoords, raw_event_append_queue_with_manifest_write_mode, + RawEventAppendOutcome, RawEventAppendSender, RawEventAppendWorker, StoryCoords, + raw_event_append_queue, }; use crate::gateway_partition::{GatewayPartitionRouter, GatewaySplitTemplate}; @@ -33,15 +31,12 @@ pub(crate) fn gateway_capture_sink( gateway_capture_sink_with_factory(dataset_uri, default_agent_id, None, raw_event_append_queue) } -pub(crate) fn gateway_capture_sink_with_manifest_write_mode( +pub(crate) fn gateway_capture_sink_with_split( dataset_uri: &str, default_agent_id: &str, split: Option, - manifest_write_mode: ObjectStoreManifestWriteMode, ) -> anyhow::Result<(Arc, GatewayCaptureWriter)> { - gateway_capture_sink_with_factory(dataset_uri, default_agent_id, split, move || { - raw_event_append_queue_with_manifest_write_mode(manifest_write_mode) - }) + gateway_capture_sink_with_factory(dataset_uri, default_agent_id, split, raw_event_append_queue) } fn gateway_capture_sink_with_factory( diff --git a/crates/persisting-pchronicle-cli/src/gateway_ingest.rs b/crates/persisting-pchronicle-cli/src/gateway_ingest.rs index a5dc9eb1..65f61146 100644 --- a/crates/persisting-pchronicle-cli/src/gateway_ingest.rs +++ b/crates/persisting-pchronicle-cli/src/gateway_ingest.rs @@ -12,8 +12,8 @@ use axum::routing::{get, post}; use axum::{Json, Router}; use persisting_events::{EventRecord, TrajectoryAppendResponse}; use persisting_pchronicle::storage::{ - ObjectStoreManifestWriteMode, RawEventAppendOutcome, RawEventAppendSender, - RawEventAppendWorker, StoryCoords, raw_event_append_queue_with_manifest_write_mode, + RawEventAppendOutcome, RawEventAppendSender, RawEventAppendWorker, StoryCoords, + raw_event_append_queue, }; use serde::{Deserialize, Serialize}; @@ -65,7 +65,6 @@ impl PreparedIngestGateway { listen: std::net::SocketAddr, dataset_uri: String, split: Option, - manifest_write_mode: ObjectStoreManifestWriteMode, ) -> Result { anyhow::ensure!( listen.ip().is_loopback(), @@ -78,8 +77,7 @@ impl PreparedIngestGateway { .local_addr() .context("read pChronicle ingest Gateway listen address")? .to_string(); - let (sender, worker) = - raw_event_append_queue_with_manifest_write_mode(manifest_write_mode)?; + let (sender, worker) = raw_event_append_queue()?; Ok(Self { listener, endpoint, @@ -636,10 +634,7 @@ mod tests { #[test] fn append_routes_to_user_partition_and_is_durable() { let temporary = tempfile::tempdir().unwrap(); - let (sender, worker) = raw_event_append_queue_with_manifest_write_mode( - ObjectStoreManifestWriteMode::Conditional, - ) - .unwrap(); + let (sender, worker) = raw_event_append_queue().unwrap(); let state = IngestState { partitions: GatewayPartitionRouter::new( temporary.path().to_string_lossy(), diff --git a/crates/persisting-pchronicle-cli/src/lib.rs b/crates/persisting-pchronicle-cli/src/lib.rs index b3751837..bea8dd02 100644 --- a/crates/persisting-pchronicle-cli/src/lib.rs +++ b/crates/persisting-pchronicle-cli/src/lib.rs @@ -47,8 +47,8 @@ use persisting_pchronicle::storage::{ AutomaticProjectionInspection, AutomaticProjectionState, CatalogErrorPolicy, CatalogSnapshotOptions, CatalogSourceKind, CatalogSourceStatus, CatalogStorylineKey, DEFAULT_DATASET_NAME, DatasetCatalogSnapshot, DatasetLocation, DatasetMount, DiscoveredSource, - EventFactSnapshot, ObjectStoreManifestWriteMode, StorylineLanceStore, - StorylineProjectionBuildOutcome, automatic_projection_inventory, build_storyline_projection, + EventFactSnapshot, StorylineLanceStore, StorylineProjectionBuildOutcome, + automatic_projection_inventory, build_storyline_projection, inspect_automatic_storyline_projection, probe_canonical_event_store, }; use serde::{Deserialize, Serialize}; @@ -960,17 +960,6 @@ struct ServeArgs { #[arg(long, value_name = "DIRECTORY", requires = "gateway_config")] gateway_state: Option, - /// Object-store manifest publication contract used by Gateway capture. - #[arg( - long, - value_enum, - default_value_t, - requires = "gateway_mode", - value_name = "MODE", - hide = true - )] - gateway_object_store_manifest_mode: GatewayObjectStoreManifestMode, - /// Also maintain Gateway's live AgenticMD projection. #[arg(long, requires = "gateway_config")] gateway_stream_markdown: bool, @@ -1163,23 +1152,6 @@ enum EchoEncoding { Base64, } -#[derive(Debug, Clone, Copy, Default, PartialEq, Eq, ValueEnum)] -enum GatewayObjectStoreManifestMode { - #[default] - Conditional, - /// One Gateway process owns the Dataset; conditional object replacement is unavailable. - SingleWriter, -} - -impl From for ObjectStoreManifestWriteMode { - fn from(mode: GatewayObjectStoreManifestMode) -> Self { - match mode { - GatewayObjectStoreManifestMode::Conditional => Self::Conditional, - GatewayObjectStoreManifestMode::SingleWriter => Self::SingleWriter, - } - } -} - #[derive(Debug, Deserialize)] #[serde(deny_unknown_fields)] struct WarehouseFile { @@ -1701,21 +1673,11 @@ async fn prepare_gateway( .map(gateway_partition::GatewaySplitTemplate::parse) .transpose()?; let local_dataset = local_dataset_path(dataset_uri)?; - if args.gateway_object_store_manifest_mode == GatewayObjectStoreManifestMode::SingleWriter { - anyhow::ensure!( - local_dataset.is_none(), - "--gateway-object-store-manifest-mode single-writer requires an object-store Dataset" - ); - } if let Some(listen) = args.gateway { - let gateway = gateway_ingest::PreparedIngestGateway::bind( - listen, - dataset_uri.to_string(), - split, - args.gateway_object_store_manifest_mode.into(), - ) - .await?; + let gateway = + gateway_ingest::PreparedIngestGateway::bind(listen, dataset_uri.to_string(), split) + .await?; return Ok(Some(PreparedGateway::Ingest(gateway))); } @@ -1753,11 +1715,10 @@ async fn prepare_gateway( .local_addr() .context("read pChronicle Gateway admin listen address")? .to_string(); - let (sink, writer) = gateway_capture::gateway_capture_sink_with_manifest_write_mode( + let (sink, writer) = gateway_capture::gateway_capture_sink_with_split( dataset_uri, &config.agent_id, split.clone(), - args.gateway_object_store_manifest_mode.into(), )?; Ok(Some(PreparedGateway::Proxy(Box::new( PreparedProxyGateway { diff --git a/crates/persisting-pchronicle-cli/src/projection_supervisor.rs b/crates/persisting-pchronicle-cli/src/projection_supervisor.rs index c8de7d31..db8c9d87 100644 --- a/crates/persisting-pchronicle-cli/src/projection_supervisor.rs +++ b/crates/persisting-pchronicle-cli/src/projection_supervisor.rs @@ -130,21 +130,33 @@ impl ProjectionSupervisor { ) }) .collect(); - let mut failures = inventory.errors.len(); + let mut failures = inventory + .errors + .iter() + .map(|error| format!("{}: source discovery failed", error.source_path)) + .collect::>(); let outcomes = stream::iter(inventory.targets) .map(|target| async move { + let source_path = target.source_path.clone(); maintain_automatic_storyline_projection(&target) .await .map(|_| ()) + .map_err(|error| (source_path, error)) }) .buffer_unordered(self.options.max_concurrent.max(1)) .collect::>() .await; - failures = - failures.saturating_add(outcomes.iter().filter(|result| result.is_err()).count()); + failures.extend(outcomes.into_iter().filter_map(|outcome| match outcome { + Ok(()) => None, + Err((source_path, error)) => { + Some(format!("{}: {error:#}", sanitize_log_field(&source_path))) + } + })); anyhow::ensure!( - failures == 0, - "automatic Storyline projection startup failed for {failures} source(s)" + failures.is_empty(), + "automatic Storyline projection startup failed for {} source(s): {}", + failures.len(), + failures.join("; ") ); let converged = self.discover().await?; @@ -446,6 +458,7 @@ mod tests { let error = supervisor.converge_before_readiness().await.unwrap_err(); assert!(error.to_string().contains("startup failed for 1 source")); + assert!(error.to_string().contains("b/events.lance")); assert_eq!(std::fs::read(projection_b.join("CURRENT"))?, before); Ok(()) } diff --git a/crates/persisting-pchronicle-cli/src/tests.rs b/crates/persisting-pchronicle-cli/src/tests.rs index b865ab89..a5d51cce 100644 --- a/crates/persisting-pchronicle-cli/src/tests.rs +++ b/crates/persisting-pchronicle-cli/src/tests.rs @@ -4427,7 +4427,6 @@ fn serve_args_with_storage(storage: Vec) -> ServeArgs { gateway_split: None, gateway_split_idle_seconds: 1800, gateway_state: None, - gateway_object_store_manifest_mode: GatewayObjectStoreManifestMode::default(), gateway_stream_markdown: false, debug: false, catalog_config: None, @@ -4953,8 +4952,6 @@ fn serve_gateway_options_are_explicit_and_scoped() -> Result<()> { "s3://capture-bucket/events", "--gateway-state", ".gateway-state", - "--gateway-object-store-manifest-mode", - "single-writer", "--gateway-stream-markdown", "--gateway-debug", ])?; @@ -4968,10 +4965,6 @@ fn serve_gateway_options_are_explicit_and_scoped() -> Result<()> { Some("s3://capture-bucket/events") ); assert_eq!(args.gateway_state, Some(PathBuf::from(".gateway-state"))); - assert_eq!( - args.gateway_object_store_manifest_mode, - GatewayObjectStoreManifestMode::SingleWriter - ); assert!(args.gateway_stream_markdown); assert!(args.debug); diff --git a/crates/persisting-pchronicle-cli/tests/binary_contract.rs b/crates/persisting-pchronicle-cli/tests/binary_contract.rs index 38975286..35c27592 100644 --- a/crates/persisting-pchronicle-cli/tests/binary_contract.rs +++ b/crates/persisting-pchronicle-cli/tests/binary_contract.rs @@ -124,12 +124,7 @@ fn serve_help_exposes_only_the_canonical_dataset_surface() -> Result<()> { "serve help omits {option}: {stdout}" ); } - for legacy in [ - "--warehouse-config", - "--storage", - "--gateway-object-store-manifest-mode", - "--catalog-query-worker", - ] { + for legacy in ["--warehouse-config", "--storage", "--catalog-query-worker"] { assert!( !stdout.contains(legacy), "serve help exposes compatibility option {legacy}: {stdout}" diff --git a/crates/persisting-pchronicle/src/append_queue.rs b/crates/persisting-pchronicle/src/append_queue.rs index f832f13f..25a96f93 100644 --- a/crates/persisting-pchronicle/src/append_queue.rs +++ b/crates/persisting-pchronicle/src/append_queue.rs @@ -10,7 +10,7 @@ use anyhow::Context; use crate::formats::EventRecord; use crate::layout::StoryCoords; use crate::store::compact_sealed_event_segment; -use crate::store::{ObjectStoreManifestWriteMode, RawEventLanceAppender, raw_event_lance_path}; +use crate::store::{RawEventLanceAppender, raw_event_lance_path}; pub const DEFAULT_RAW_EVENT_QUEUE_CAPACITY: usize = 256; pub const DEFAULT_RAW_EVENT_BATCH_SIZE: usize = 256; @@ -191,18 +191,6 @@ pub fn raw_event_append_queue() -> anyhow::Result<(RawEventAppendSender, RawEven raw_event_append_queue_with_capacity(DEFAULT_RAW_EVENT_QUEUE_CAPACITY) } -pub fn raw_event_append_queue_with_manifest_write_mode( - manifest_write_mode: ObjectStoreManifestWriteMode, -) -> anyhow::Result<(RawEventAppendSender, RawEventAppendWorker)> { - raw_event_append_queue_with_options( - DEFAULT_RAW_EVENT_QUEUE_CAPACITY, - DEFAULT_RAW_EVENT_COMPACTION_THRESHOLD, - DEFAULT_RAW_EVENT_TARGET_ROWS_PER_FRAGMENT, - DEFAULT_RAW_EVENT_HIERARCHY_FANOUT, - manifest_write_mode, - ) -} - pub fn raw_event_append_queue_with_capacity( capacity: usize, ) -> anyhow::Result<(RawEventAppendSender, RawEventAppendWorker)> { @@ -211,7 +199,6 @@ pub fn raw_event_append_queue_with_capacity( DEFAULT_RAW_EVENT_COMPACTION_THRESHOLD, DEFAULT_RAW_EVENT_TARGET_ROWS_PER_FRAGMENT, DEFAULT_RAW_EVENT_HIERARCHY_FANOUT, - ObjectStoreManifestWriteMode::Conditional, ) } @@ -220,7 +207,6 @@ fn raw_event_append_queue_with_options( compaction_threshold: usize, target_rows_per_fragment: usize, hierarchy_fanout: usize, - manifest_write_mode: ObjectStoreManifestWriteMode, ) -> anyhow::Result<(RawEventAppendSender, RawEventAppendWorker)> { if capacity == 0 { anyhow::bail!("pChronicle append queue capacity must be greater than zero"); @@ -255,7 +241,6 @@ fn raw_event_append_queue_with_options( compaction_threshold, target_rows_per_fragment, hierarchy_fanout, - manifest_write_mode, ) } }) @@ -278,15 +263,13 @@ fn run_append_worker( compaction_threshold: usize, target_rows_per_fragment: usize, hierarchy_fanout: usize, - manifest_write_mode: ObjectStoreManifestWriteMode, ) -> anyhow::Result<()> { let runtime = tokio::runtime::Builder::new_multi_thread() .worker_threads(2) .enable_all() .build() .context("create pChronicle append worker runtime")?; - let mut appender = - RawEventLanceAppender::default().with_object_store_manifest_write_mode(manifest_write_mode); + let mut appender = RawEventLanceAppender::default(); let (maintenance_tx, mut maintenance_rx) = tokio::sync::mpsc::channel(DEFAULT_RAW_EVENT_MAINTENANCE_CAPACITY); let maintenance_task = runtime.spawn(async move { @@ -645,34 +628,6 @@ mod tests { worker.finish().unwrap(); } - #[test] - fn single_writer_manifest_mode_publishes_object_store_events() { - let storage = format!( - "shared-memory://append-single-writer-{}/dataset", - uuid::Uuid::new_v4() - ); - let coords = StoryCoords::new(storage, "agent", "session", None); - let (sender, worker) = raw_event_append_queue_with_manifest_write_mode( - ObjectStoreManifestWriteMode::SingleWriter, - ) - .unwrap(); - - assert_eq!( - sender.append_durable(coords.clone(), event()).unwrap(), - RawEventAppendOutcome::Accepted - ); - worker.finish().unwrap(); - - let runtime = tokio::runtime::Builder::new_current_thread() - .enable_all() - .build() - .unwrap(); - let replay = runtime - .block_on(RawEventLanceStore.replay(&coords, 0, None)) - .unwrap(); - assert_eq!(replay.records.len(), 1); - } - #[test] fn durable_append_isolates_one_partition_failure() { let dir = tempfile::tempdir().unwrap(); @@ -746,14 +701,7 @@ mod tests { let dir = tempfile::tempdir().unwrap(); let storage = dir.path().join("store"); let coords = StoryCoords::new(storage.to_string_lossy(), "agent", "session", None); - let (sender, worker) = raw_event_append_queue_with_options( - 32, - 2, - 2, - 2, - ObjectStoreManifestWriteMode::Conditional, - ) - .unwrap(); + let (sender, worker) = raw_event_append_queue_with_options(32, 2, 2, 2).unwrap(); // 8 rows become four L0 segments, two L1 segments, and finally one L2 // segment. Each merge preserves append order and total row count. @@ -823,7 +771,6 @@ mod tests { DEFAULT_RAW_EVENT_COMPACTION_THRESHOLD, DEFAULT_RAW_EVENT_TARGET_ROWS_PER_FRAGMENT, DEFAULT_RAW_EVENT_HIERARCHY_FANOUT, - ObjectStoreManifestWriteMode::Conditional, ) }); let worker = RawEventAppendWorker { @@ -897,7 +844,6 @@ mod tests { DEFAULT_RAW_EVENT_COMPACTION_THRESHOLD, DEFAULT_RAW_EVENT_TARGET_ROWS_PER_FRAGMENT, DEFAULT_RAW_EVENT_HIERARCHY_FANOUT, - ObjectStoreManifestWriteMode::Conditional, ) }); let worker = RawEventAppendWorker { diff --git a/crates/persisting-pchronicle/src/storage.rs b/crates/persisting-pchronicle/src/storage.rs index b524f303..3a6b7a8e 100644 --- a/crates/persisting-pchronicle/src/storage.rs +++ b/crates/persisting-pchronicle/src/storage.rs @@ -9,7 +9,6 @@ pub use crate::append_queue::{ DEFAULT_RAW_EVENT_MAINTENANCE_CAPACITY, DEFAULT_RAW_EVENT_QUEUE_CAPACITY, DEFAULT_RAW_EVENT_TARGET_ROWS_PER_FRAGMENT, RawEventAppendOutcome, RawEventAppendSender, RawEventAppendWorker, raw_event_append_queue, raw_event_append_queue_with_capacity, - raw_event_append_queue_with_manifest_write_mode, }; pub use crate::layout::{ StoryCoords, StoryLocationPartial, is_subagent_session_storage_key, @@ -38,16 +37,15 @@ pub use crate::store::{ DatasetLocation, DatasetLocationKind, DatasetMount, DiscoveredSource, EventFactSnapshot, EventLogLayoutStats, EventWriterFence, ExportOutcome, LanceMaintenanceOptions, LanceMaintenanceReport, LeaseAcquireOutcome, ManifestKind, ManifestStats, NamespacePath, - ObjectStoreManifestWriteMode, PhysicalColumn, PhysicalDataFile, PhysicalFileLayout, - PhysicalFragment, PhysicalLayout, PhysicalPage, PhysicalPagePreview, PhysicalPageQuery, - PhysicalSource, PhysicalTable, ProjectionSourceSnapshot, RawEventLanceAppender, - RawEventLanceStore, ReplayOutcome, RunControlStore, StorylineContentOptions, - StorylineContentReadMode, StorylineDataSource, StorylineDataSourceOptions, StorylineLanceStore, - StorylineMaintenanceReport, StorylineProjectionLineage, StorylineStreamImportReport, - StorylineTablePaths, TrajectoryStats, attempt_registry_now_ms, distinct_session_ids_in_run, - export_source_dirs, export_story_bundle, inspect_physical_file, inspect_physical_layout, - inspect_physical_page, list_physical_sources, load_manifest, raw_event_lance_path, - write_compact_jsonl_manifest, + PhysicalColumn, PhysicalDataFile, PhysicalFileLayout, PhysicalFragment, PhysicalLayout, + PhysicalPage, PhysicalPagePreview, PhysicalPageQuery, PhysicalSource, PhysicalTable, + ProjectionSourceSnapshot, RawEventLanceAppender, RawEventLanceStore, ReplayOutcome, + RunControlStore, StorylineContentOptions, StorylineContentReadMode, StorylineDataSource, + StorylineDataSourceOptions, StorylineLanceStore, StorylineMaintenanceReport, + StorylineProjectionLineage, StorylineStreamImportReport, StorylineTablePaths, TrajectoryStats, + attempt_registry_now_ms, distinct_session_ids_in_run, export_source_dirs, export_story_bundle, + inspect_physical_file, inspect_physical_layout, inspect_physical_page, list_physical_sources, + load_manifest, raw_event_lance_path, write_compact_jsonl_manifest, }; // Compatibility exports; new callers should use `crate::search`. diff --git a/crates/persisting-pchronicle/src/store/dataset_write_lock.rs b/crates/persisting-pchronicle/src/store/dataset_write_lock.rs index 39caef67..56a99a62 100644 --- a/crates/persisting-pchronicle/src/store/dataset_write_lock.rs +++ b/crates/persisting-pchronicle/src/store/dataset_write_lock.rs @@ -1,4 +1,4 @@ -//! Dataset-scoped write serialization for the single-writer storage engine. +//! Dataset-scoped write serialization. use std::fs::{File, OpenOptions}; use std::path::Path; @@ -22,9 +22,9 @@ impl Drop for DatasetWriteGuard { } /// Serialize writers within the process and, for local datasets, across -/// processes. Object-store deployments retain the documented single-writer -/// contract; Lance transactions provide atomic publication but not a global -/// distributed mutex. +/// processes. Object-store deployments rely on conditional publication of +/// the visibility manifest for cross-process safety; Lance transactions +/// provide atomic publication but not a global distributed mutex. pub(crate) async fn acquire(uri: &str) -> Result { let process = root_write_lock::for_root(uri).lock_owned().await; let local_file = if super::events::is_object_store_uri(uri) { diff --git a/crates/persisting-pchronicle/src/store/events/manifest.rs b/crates/persisting-pchronicle/src/store/events/manifest.rs index 34f51319..79b24b6e 100644 --- a/crates/persisting-pchronicle/src/store/events/manifest.rs +++ b/crates/persisting-pchronicle/src/store/events/manifest.rs @@ -20,15 +20,6 @@ const MANIFEST_FILE: &str = "_manifest.json"; const MANIFEST_LOCK_FILE: &str = "_manifest.lock"; const CAS_RETRIES: usize = 64; -#[derive(Debug, Clone, Copy, Default, PartialEq, Eq)] -pub enum ObjectStoreManifestWriteMode { - #[default] - Conditional, - /// Use only when one process owns the object-store Dataset. This supports - /// S3-compatible providers that cannot conditionally replace an object. - SingleWriter, -} - #[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] pub struct EventWriterFence { pub epoch: u64, @@ -104,25 +95,10 @@ pub(super) async fn activate( root_uri: &str, requested: Option<&EventWriterFence>, auto_writer_id: &str, -) -> Result> { - activate_with_mode( - root_uri, - requested, - auto_writer_id, - ObjectStoreManifestWriteMode::Conditional, - ) - .await -} - -pub(super) async fn activate_with_mode( - root_uri: &str, - requested: Option<&EventWriterFence>, - auto_writer_id: &str, - write_mode: ObjectStoreManifestWriteMode, ) -> Result> { let requested = requested.cloned(); let auto_writer_id = auto_writer_id.to_string(); - mutate_with_mode(root_uri, write_mode, move |current| { + mutate(root_uri, move |current| { let next_fence = match (&requested, current) { (Some(fence), Some(manifest)) if fence == &manifest.active_writer => { return Ok(ManifestMutation::Unchanged(manifest.clone())); @@ -166,29 +142,13 @@ pub(super) async fn activate_with_mode( .await } -#[cfg(test)] pub(super) async fn publish_segment( root_uri: &str, fence: &EventWriterFence, segment: EventSegment, -) -> Result> { - publish_segment_with_mode( - root_uri, - fence, - segment, - ObjectStoreManifestWriteMode::Conditional, - ) - .await -} - -pub(super) async fn publish_segment_with_mode( - root_uri: &str, - fence: &EventWriterFence, - segment: EventSegment, - write_mode: ObjectStoreManifestWriteMode, ) -> Result> { let fence = fence.clone(); - mutate_with_mode(root_uri, write_mode, move |current| { + mutate(root_uri, move |current| { let current = current.context("event manifest disappeared during publish")?; if let Some(conflict) = active_writer_conflict(current, &fence) { return Ok(ManifestMutation::Conflict(conflict)); @@ -282,34 +242,16 @@ pub(super) async fn replace_segments( /// Atomically replace one contiguous immutable segment group while preserving /// its position in append order. Exact descriptor matching prevents a stale /// compactor from overwriting a segment version published by another task. -#[cfg(test)] pub(super) async fn replace_segment_group( root_uri: &str, fence: &EventWriterFence, expected: &[EventSegment], replacement: EventSegment, -) -> Result> { - replace_segment_group_with_mode( - root_uri, - fence, - expected, - replacement, - ObjectStoreManifestWriteMode::Conditional, - ) - .await -} - -pub(super) async fn replace_segment_group_with_mode( - root_uri: &str, - fence: &EventWriterFence, - expected: &[EventSegment], - replacement: EventSegment, - write_mode: ObjectStoreManifestWriteMode, ) -> Result> { anyhow::ensure!(!expected.is_empty(), "segment replacement group is empty"); let expected = expected.to_vec(); let fence = fence.clone(); - mutate_with_mode(root_uri, write_mode, move |current| { + mutate(root_uri, move |current| { let current = current.context("event manifest disappeared during segment merge")?; if let Some(conflict) = active_writer_conflict(current, &fence) { return Ok(ManifestMutation::Conflict(conflict)); @@ -487,23 +429,6 @@ enum ManifestMutation { } async fn mutate(root_uri: &str, mutation: F) -> Result> -where - T: Send + 'static, - F: Fn(Option<&EventManifest>) -> Result> + Send + Sync + 'static, -{ - mutate_with_mode( - root_uri, - ObjectStoreManifestWriteMode::Conditional, - mutation, - ) - .await -} - -async fn mutate_with_mode( - root_uri: &str, - write_mode: ObjectStoreManifestWriteMode, - mutation: F, -) -> Result> where T: Send + 'static, F: Fn(Option<&EventManifest>) -> Result> + Send + Sync + 'static, @@ -560,14 +485,9 @@ where }; validate_manifest(&manifest)?; let bytes = serde_json::to_vec_pretty(&manifest)?; - let result = match (write_mode, current.as_ref().map(|(_, version)| version)) { - (_, None) => store.write_create(path, bytes).await, - (ObjectStoreManifestWriteMode::Conditional, Some(version)) => { - store.write_match(path, bytes, version).await - } - (ObjectStoreManifestWriteMode::SingleWriter, Some(_)) => { - store.write_overwrite(path, bytes).await - } + let result = match current.as_ref().map(|(_, version)| version) { + None => store.write_create(path, bytes).await, + Some(version) => store.write_match(path, bytes, version).await, }; match result { Ok(_) => return Ok(ManifestWriteOutcome::Applied(value)), diff --git a/crates/persisting-pchronicle/src/store/events/mod.rs b/crates/persisting-pchronicle/src/store/events/mod.rs index 325d1d7b..888b3b39 100644 --- a/crates/persisting-pchronicle/src/store/events/mod.rs +++ b/crates/persisting-pchronicle/src/store/events/mod.rs @@ -29,8 +29,8 @@ use lance_index::scalar::{BuiltinIndexType, ScalarIndexParams}; pub use self::datafusion::{DATAFUSION_EVENTS_TABLE, EventFactSnapshot, RawEventDataSource}; use self::manifest as raw_event_manifest; +pub use self::manifest::EventWriterFence; use self::manifest::{EventManifest, EventSegment, EventWriterConflict, ManifestWriteOutcome}; -pub use self::manifest::{EventWriterFence, ObjectStoreManifestWriteMode}; pub use self::rows::{ event_records_from_batch, event_rows_from_batch, event_rows_to_batch, raw_event_arrow_schema, }; @@ -101,14 +101,12 @@ pub struct EventLogLayoutStats { pub struct RawEventLanceAppender { requested_fence: Option, auto_writer_id: String, - manifest_write_mode: ObjectStoreManifestWriteMode, datasets: BTreeMap, } #[derive(Debug)] struct CachedRawDataset { fence: EventWriterFence, - manifest_write_mode: ObjectStoreManifestWriteMode, segment_id: String, segment_uri: String, dataset: Option, @@ -160,7 +158,6 @@ impl EventAppendBatchReport { pub(crate) struct SealedEventSegment { root_uri: String, fence: EventWriterFence, - manifest_write_mode: ObjectStoreManifestWriteMode, segment: EventSegment, } @@ -169,7 +166,6 @@ impl Default for RawEventLanceAppender { Self { requested_fence: None, auto_writer_id: format!("auto-{}", uuid::Uuid::new_v4()), - manifest_write_mode: ObjectStoreManifestWriteMode::Conditional, datasets: BTreeMap::new(), } } @@ -184,19 +180,10 @@ impl RawEventLanceAppender { Self { auto_writer_id: fence.writer_id.clone(), requested_fence: Some(fence), - manifest_write_mode: ObjectStoreManifestWriteMode::Conditional, datasets: BTreeMap::new(), } } - pub fn with_object_store_manifest_write_mode( - mut self, - mode: ObjectStoreManifestWriteMode, - ) -> Self { - self.manifest_write_mode = mode; - self - } - /// Activate this writer before accepting data. A newer activation fences /// every older appender at the manifest publication boundary. pub async fn activate(&mut self, session: &StoryCoords) -> Result { @@ -214,19 +201,13 @@ impl RawEventLanceAppender { async fn new_state(&self, uri: &str) -> Result { let manifest = manifest_write_applied( - raw_event_manifest::activate_with_mode( - uri, - self.requested_fence.as_ref(), - &self.auto_writer_id, - self.manifest_write_mode, - ) - .await?, + raw_event_manifest::activate(uri, self.requested_fence.as_ref(), &self.auto_writer_id) + .await?, )?; let fence = manifest.active_writer.clone(); let segment_id = format!("e{}-{}", fence.epoch, uuid::Uuid::new_v4()); Ok(CachedRawDataset { fence, - manifest_write_mode: self.manifest_write_mode, segment_uri: raw_event_manifest::segment_uri(uri, &segment_id), segment_id, dataset: None, @@ -359,7 +340,6 @@ impl RawEventLanceAppender { sealed.push(SealedEventSegment { root_uri: uri.clone(), fence: state.fence.clone(), - manifest_write_mode: state.manifest_write_mode, segment: EventSegment { id: state.segment_id.clone(), version: dataset.version_id(), @@ -405,7 +385,7 @@ async fn append_event_group( .context("event segment row count overflow")?; state.pending_fragments = state.pending_fragments.saturating_add(1); let manifest = manifest_write_applied( - raw_event_manifest::publish_segment_with_mode( + raw_event_manifest::publish_segment( &uri, &state.fence, EventSegment { @@ -415,7 +395,6 @@ async fn append_event_group( level: 0, sealed: false, }, - state.manifest_write_mode, ) .await?, )?; @@ -463,21 +442,10 @@ pub(crate) async fn compact_sealed_event_segment( } published_segment.sealed = true; manifest_write_applied( - raw_event_manifest::publish_segment_with_mode( - &sealed.root_uri, - &sealed.fence, - published_segment, - sealed.manifest_write_mode, - ) - .await?, + raw_event_manifest::publish_segment(&sealed.root_uri, &sealed.fence, published_segment) + .await?, )?; - compact_event_hierarchy_locked( - &sealed.root_uri, - &sealed.fence, - hierarchy_fanout, - sealed.manifest_write_mode, - ) - .await?; + compact_event_hierarchy_locked(&sealed.root_uri, &sealed.fence, hierarchy_fanout).await?; Ok(()) } @@ -498,7 +466,6 @@ async fn compact_event_hierarchy_locked( root_uri: &str, fence: &EventWriterFence, fanout: usize, - manifest_write_mode: ObjectStoreManifestWriteMode, ) -> Result<()> { loop { let manifest = raw_event_manifest::read(root_uri) @@ -530,14 +497,7 @@ async fn compact_event_hierarchy_locked( replacement.level = next_level; replacement.sealed = true; manifest_write_applied( - raw_event_manifest::replace_segment_group_with_mode( - root_uri, - fence, - &group, - replacement, - manifest_write_mode, - ) - .await?, + raw_event_manifest::replace_segment_group(root_uri, fence, &group, replacement).await?, )?; } } diff --git a/crates/persisting-pchronicle/src/store/events/tests.rs b/crates/persisting-pchronicle/src/store/events/tests.rs index b5fa918a..2ea73979 100644 --- a/crates/persisting-pchronicle/src/store/events/tests.rs +++ b/crates/persisting-pchronicle/src/store/events/tests.rs @@ -486,7 +486,7 @@ async fn event_schema_mismatch_is_rejected() { #[tokio::test] async fn one_cached_appender_preserves_physical_append_order() { - let storage = remote_storage("cached-single-writer"); + let storage = remote_storage("cached-appender"); let session = flat_session(&storage, "agent", "session"); let mut writer = RawEventLanceAppender::default(); diff --git a/crates/persisting-pchronicle/src/store/mod.rs b/crates/persisting-pchronicle/src/store/mod.rs index 76ee0980..775f18b8 100644 --- a/crates/persisting-pchronicle/src/store/mod.rs +++ b/crates/persisting-pchronicle/src/store/mod.rs @@ -92,9 +92,9 @@ pub use event_row::{EventRow, event_record_to_event_row, event_row_to_event_reco #[cfg(feature = "lance-store")] pub use events::{ DATAFUSION_EVENTS_TABLE, EventFactSnapshot, EventLogLayoutStats, EventWriterFence, - LanceMaintenanceOptions, LanceMaintenanceReport, ObjectStoreManifestWriteMode, - RawEventDataSource, RawEventLanceAppender, distinct_session_ids_in_run, event_rows_from_batch, - maintain as maintain_raw_events, raw_event_arrow_schema, + LanceMaintenanceOptions, LanceMaintenanceReport, RawEventDataSource, RawEventLanceAppender, + distinct_session_ids_in_run, event_rows_from_batch, maintain as maintain_raw_events, + raw_event_arrow_schema, }; #[cfg(feature = "lance-store")] pub(crate) use events::{SealedEventSegment, compact_sealed_event_segment}; diff --git a/crates/persisting-pchronicle/src/store/opendal_store.rs b/crates/persisting-pchronicle/src/store/opendal_store.rs index 10a3cbd8..36152241 100644 --- a/crates/persisting-pchronicle/src/store/opendal_store.rs +++ b/crates/persisting-pchronicle/src/store/opendal_store.rs @@ -118,12 +118,38 @@ impl Store { let condition = expected.condition().ok_or_else(|| { anyhow!("OpenDAL backend did not return an ETag/version for conditional write") })?; - self.operator - .write_with(path, bytes) + let result = self + .operator + .write_with(path, bytes.clone()) .if_match(condition) - .await - .map(|_| ()) - .map_err(Into::into) + .await; + match result { + Ok(_) => Ok(()), + // Some S3-compatible gateways compare the If-Match header against + // their unquoted ETag, so a correctly quoted condition always + // fails with 412. One retry with the unquoted form still proves + // the stored ETag matched; real contention fails both attempts. + Err(error) + if error.kind() == ErrorKind::ConditionNotMatch + && let Some(unquoted) = unquoted_etag(condition) => + { + let retry = self + .operator + .write_with(path, bytes) + .if_match(unquoted) + .await; + match retry { + Ok(_) => Ok(()), + // Preserve the original conditional conflict when the + // gateway rejects the compatibility form itself. + Err(retry_error) if retry_error.kind() != ErrorKind::ConditionNotMatch => { + Err(error.into()) + } + Err(retry_error) => Err(retry_error.into()), + } + } + Err(error) => Err(error.into()), + } } pub(crate) async fn write_overwrite(&self, path: &str, bytes: Vec) -> Result<()> { @@ -187,6 +213,13 @@ pub(crate) fn is_conflict(error: &opendal::Error) -> bool { ) } +/// Strip one pair of surrounding double quotes from a conditional-write ETag. +/// Returns `None` for unquoted or empty conditions. +fn unquoted_etag(condition: &str) -> Option<&str> { + let inner = condition.strip_prefix('"')?.strip_suffix('"')?; + (!inner.is_empty()).then_some(inner) +} + pub(crate) fn version(metadata: &Metadata) -> Version { Version { etag: metadata.etag().map(ToOwned::to_owned), @@ -219,3 +252,16 @@ fn normalize_uri(uri: &str) -> Result { } Ok(parsed.to_string()) } + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn unquoted_etag_strips_one_quote_pair() { + assert_eq!(unquoted_etag("\"abc\""), Some("abc")); + assert_eq!(unquoted_etag("\"\""), None); + assert_eq!(unquoted_etag("abc"), None); + assert_eq!(unquoted_etag("\"abc"), None); + } +} diff --git a/crates/persisting-pchronicle/tests/s3_storage.rs b/crates/persisting-pchronicle/tests/s3_storage.rs index 7d96ed94..42eba1b3 100644 --- a/crates/persisting-pchronicle/tests/s3_storage.rs +++ b/crates/persisting-pchronicle/tests/s3_storage.rs @@ -9,8 +9,8 @@ use persisting_pchronicle::document::{DocumentFormat, decode_json_storylines}; use persisting_pchronicle::model::{EventIdentity, EventRecord, StorylineDocument}; use persisting_pchronicle::query::{ChronicleQueryEngine, ChronicleQueryExecutionOptions}; use persisting_pchronicle::storage::{ - LanceMaintenanceOptions, ObjectStoreManifestWriteMode, RawEventLanceAppender, - RawEventLanceStore, StoryCoords, StorylineLanceStore, raw_event_lance_path, + LanceMaintenanceOptions, RawEventLanceAppender, RawEventLanceStore, StoryCoords, + StorylineLanceStore, raw_event_lance_path, }; use std::io::{Read, Write}; use std::process::{Command, Stdio}; @@ -277,44 +277,6 @@ async fn run_append_scale_contract(root: &str) -> Result<()> { Ok(()) } -async fn run_single_writer_manifest_contract(root: &str) -> Result<()> { - let event_root = format!("{root}/event-single-writer"); - let session = StoryCoords::new( - &event_root, - "contract-agent", - "single-writer-story", - Some("single-writer-run".into()), - ); - let mut writer = RawEventLanceAppender::default() - .with_object_store_manifest_write_mode(ObjectStoreManifestWriteMode::SingleWriter); - writer - .append_event_batch(&[(session.clone(), event("first"))]) - .await?; - let mut second = event("second"); - second.seq = 1; - writer - .append_event_batch(&[(session.clone(), second)]) - .await?; - writer.finish(); - - let replay = RawEventLanceStore.read_events(&session, 0, None).await?; - assert_eq!(replay.len(), 2); - let engine = ChronicleQueryEngine::open( - DocumentFormat::CanonicalEvent, - raw_event_lance_path(&session)?, - ChronicleQueryExecutionOptions::default(), - ) - .await?; - let output = engine - .query_jsonl("SELECT COUNT(*) AS rows FROM events") - .await?; - assert_eq!( - serde_json::from_str::(output.trim())?["rows"], - 2 - ); - Ok(()) -} - async fn cleanup(root: &str) -> Result<()> { if let Err(error) = Operator::from_uri(root)? .delete_with(".") @@ -492,21 +454,3 @@ async fn s3_append_scale_snapshot_and_maintenance_contract() -> Result<()> { )), } } - -#[tokio::test] -#[ignore = "requires PCHRONICLE_S3_TEST_URI and a single-writer S3-compatible bucket"] -async fn s3_single_writer_manifest_contract() -> Result<()> { - let root = unique_root()?; - let contract_result = run_single_writer_manifest_contract(&root).await; - let cleanup_result = cleanup(&root).await; - match (contract_result, cleanup_result) { - (Ok(()), Ok(())) => Ok(()), - (Err(error), Ok(())) => Err(error), - (Ok(()), Err(error)) => { - Err(error).context("S3 single-writer contract passed but cleanup failed") - } - (Err(error), Err(cleanup_error)) => Err(error).context(format!( - "S3 single-writer contract cleanup also failed: {cleanup_error:#}" - )), - } -}