fix(pchronicle): fix distributed S3 manifest writes

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<GatewaySplitTemplate>,
-    manifest_write_mode: ObjectStoreManifestWriteMode,
 ) -> anyhow::Result<(Arc<dyn CaptureEventSink>, 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<F>(

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<GatewaySplitTemplate>,
-        manifest_write_mode: ObjectStoreManifestWriteMode,
     ) -> Result<Self> {
         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(),

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<PathBuf>,
 
-    /// 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<GatewayObjectStoreManifestMode> 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 {

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::<Vec<_>>();
         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::<Vec<_>>()
             .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(())
     }

crates/persisting-pchronicle-cli/src/tests.rs

@@ -4427,7 +4427,6 @@ fn serve_args_with_storage(storage: Vec<String>) -> 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);
 

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}"

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 {