From b46a0ac48baeb3b7e37f2cdf6c04c63c280173c5 Mon Sep 17 00:00:00 2001 From: Reiase Date: Fri, 2 Oct 2026 21:59:01 +0800 Subject: [PATCH 1/2] Improve pChronicle cache reuse and tolerant batch imports --- .../src/exchange/decode.rs | 76 +++- .../src/server/catalog_worker.rs | 7 +- crates/persisting-pchronicle-cli/src/tests.rs | 5 +- .../src/formats/actf/mod.rs | 74 +++- .../src/store/blockcache/adapter.rs | 296 +++++++++++++-- .../src/store/blockcache/block.rs | 342 +++++++++++++++--- .../src/store/blockcache/config.rs | 11 +- .../src/store/files/actf_reader.rs | 6 +- .../src/store/files/actf_stream.rs | 25 +- .../src/store/files/tests.rs | 38 ++ .../en/pchronicle/guides/troubleshooting.md | 41 +++ .../zh/pchronicle/guides/troubleshooting.md | 35 ++ 12 files changed, 869 insertions(+), 87 deletions(-) diff --git a/crates/persisting-pchronicle-cli/src/exchange/decode.rs b/crates/persisting-pchronicle-cli/src/exchange/decode.rs index 2c4c75ff5..bc8f20da0 100644 --- a/crates/persisting-pchronicle-cli/src/exchange/decode.rs +++ b/crates/persisting-pchronicle-cli/src/exchange/decode.rs @@ -438,7 +438,8 @@ pub(crate) fn decode_import_source( format!("{} is not UTF-8: {error}", diagnostic_path.display()), ) })?; - let allow_skip = requested_format == ExchangeFormat::Auto && logical_source_path.is_some(); + // Batch imports keep readable sources regardless of the requested format. + let allow_skip = logical_source_path.is_some(); let format = match resolve_import_format( requested_format, suggested_format, @@ -840,6 +841,79 @@ mod tests { use super::*; use std::path::Path; + #[test] + fn directory_import_skips_unreadable_sources_for_every_format() { + for format in [ + ExchangeFormat::Atif, + ExchangeFormat::Actf, + ExchangeFormat::OpenaiMessages, + ExchangeFormat::Storyline, + ExchangeFormat::Codex, + ExchangeFormat::ClaudeCode, + ] { + let outcome = decode_import_source( + format, + None, + ImportOutputFormat::Preserve, + Some(Path::new("input")), + Some(Path::new("bad.json")), + Some(Path::new("bad.json")), + b"not json", + &mut Default::default(), + ) + .unwrap(); + assert!( + matches!(outcome, DecodeImportOutcome::Skipped { .. }), + "{format:?}" + ); + } + for (format, input) in [ + ( + ExchangeFormat::Atif, + include_bytes!("../../assets/onboard/support-ticket.json").as_slice(), + ), + ( + ExchangeFormat::Actf, + include_bytes!("../../assets/onboard/code-repair.actf.json").as_slice(), + ), + ( + ExchangeFormat::OpenaiMessages, + include_bytes!("../../assets/onboard/training.json").as_slice(), + ), + ] { + let outcome = decode_import_source( + format, + None, + ImportOutputFormat::Preserve, + Some(Path::new("input")), + Some(Path::new("source.json")), + Some(Path::new("source.json")), + input, + &mut Default::default(), + ) + .unwrap(); + match outcome { + DecodeImportOutcome::Imported(decoded) => { + assert!(!decoded.storylines.is_empty(), "{format:?}"); + } + DecodeImportOutcome::Skipped { reason, .. } => panic!("{format:?}: {reason}"), + } + } + assert!( + decode_import_source( + ExchangeFormat::Actf, + None, + ImportOutputFormat::Preserve, + None, + None, + None, + b"{}", + &mut Default::default(), + ) + .is_err() + ); + } + #[test] fn visible_json_extensions() { assert!(is_visible_json_file(Path::new("a.json"))); diff --git a/crates/persisting-pchronicle-cli/src/server/catalog_worker.rs b/crates/persisting-pchronicle-cli/src/server/catalog_worker.rs index ff4f96fab..9dd9b476d 100644 --- a/crates/persisting-pchronicle-cli/src/server/catalog_worker.rs +++ b/crates/persisting-pchronicle-cli/src/server/catalog_worker.rs @@ -314,6 +314,7 @@ fn command( "SSL_CERT_DIR", "PCHRONICLE_QUERY_MEMORY_LIMIT", "PCHRONICLE_LANCE_CACHE_CAPACITY_BYTES", + "PCHRONICLE_LANCE_CACHE_BLOCK_SIZE_BYTES", // Workers issue the reads, so admission tuning that never reaches them // tunes nothing. "PCHRONICLE_OBJECT_STORE_CONCURRENCY", @@ -339,7 +340,11 @@ impl Worker { // and every scope can share them. Keeping them beside the per-scope // caches rather than inside one means a reader does not refetch what // another worker already paid for. - let blocks = root.join("blocks"); + let blocks = std::path::absolute( + std::env::var_os("PCHRONICLE_LANCE_CACHE_DIR") + .map(PathBuf::from) + .unwrap_or_else(|| root.join("blocks")), + )?; let mut builder = std::fs::DirBuilder::new(); builder.recursive(true); #[cfg(unix)] diff --git a/crates/persisting-pchronicle-cli/src/tests.rs b/crates/persisting-pchronicle-cli/src/tests.rs index 6cd953848..46d7de40b 100644 --- a/crates/persisting-pchronicle-cli/src/tests.rs +++ b/crates/persisting-pchronicle-cli/src/tests.rs @@ -3121,7 +3121,8 @@ async fn directory_import_dedupes_unknown_warnings_across_sources() -> Result<() } #[tokio::test] -async fn directory_import_skips_invalid_json_and_publishes_valid_sources() -> Result<()> { +async fn directory_import_explicit_format_skips_invalid_json_and_publishes_valid_sources() +-> Result<()> { let temp = tempfile::tempdir()?; let input = temp.path().join("input"); fs::create_dir_all(&input)?; @@ -3136,6 +3137,8 @@ async fn directory_import_skips_invalid_json_and_publishes_valid_sources() -> Re let mut argv = vec![ "pchronicle".to_owned(), "import".to_owned(), + "--format".to_owned(), + "atif".to_owned(), "--from".to_owned(), input.to_string_lossy().into_owned(), "--output".to_owned(), diff --git a/crates/persisting-pchronicle/src/formats/actf/mod.rs b/crates/persisting-pchronicle/src/formats/actf/mod.rs index b06689783..7b58be82c 100644 --- a/crates/persisting-pchronicle/src/formats/actf/mod.rs +++ b/crates/persisting-pchronicle/src/formats/actf/mod.rs @@ -176,8 +176,7 @@ fn decode_json( let envelope = take_unknown_fields_envelope(&mut value)?; let mut document: ActfDocument = serde_json::from_value(value).map_err(|error| InputIssue::invalid(error.to_string()))?; - normalize_solved_at(&mut document.solved_at); - reconcile_document_tool_lists(&mut document); + document.normalize_for_import(); document.validate()?; let mut stories = actf_to_storylines(&document).map_err(|error| InputIssue::invalid(error.to_string()))?; @@ -218,11 +217,15 @@ fn decode_json( pub struct ActfDocument { pub task_id: String, /// Some error dumps emit numeric categories (`2`) instead of strings. - #[serde(deserialize_with = "stringish")] + #[serde(default, deserialize_with = "stringish")] pub category: String, + #[serde(default, deserialize_with = "null_as_default")] pub k: u64, + #[serde(default, deserialize_with = "null_as_default")] pub correct: bool, + #[serde(default, deserialize_with = "null_as_default")] pub attempts_tried: u64, + #[serde(default, deserialize_with = "null_as_default")] pub solved_at: Value, pub attempts: BTreeMap, #[serde(flatten)] @@ -231,11 +234,13 @@ pub struct ActfDocument { #[derive(Debug, Clone, PartialEq, Serialize, Deserialize)] pub struct ActfAttempt { + #[serde(default, deserialize_with = "null_as_default")] pub correct: bool, #[serde(default)] pub final_answer: Value, #[serde(default)] pub ground_truth: Value, + #[serde(default)] pub trajectory: ActfTrajectory, #[serde(default, deserialize_with = "null_as_empty_string")] pub status: String, @@ -269,6 +274,16 @@ pub struct ActfTrajectory { pub extra: Map, } +impl Default for ActfTrajectory { + fn default() -> Self { + Self::from_event_log(Vec::new()) + } +} + +fn default_schema_version() -> String { + ACTF_SCHEMA_VERSION.into() +} + impl ActfTrajectory { pub fn from_event_log(events: Vec) -> Self { let started_at = events @@ -334,6 +349,7 @@ enum ActfTrajectoryWire { /// object shape, not this array. Events(Vec), Canonical { + #[serde(default = "default_schema_version")] schema_version: String, #[serde(default)] steps: Vec, @@ -396,7 +412,9 @@ impl<'de> Deserialize<'de> for ActfTrajectory { #[derive(Debug, Clone, PartialEq, Serialize, Deserialize)] pub struct ActfStep { pub step_id: i64, + #[serde(default, deserialize_with = "null_as_default")] pub assistant_content: ActfAssistantContent, + #[serde(default, deserialize_with = "null_as_default")] pub metric: ActfMetric, #[serde(default, deserialize_with = "null_as_empty_string")] pub system_prompt: String, @@ -424,7 +442,7 @@ impl ActfStep { } } -#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)] +#[derive(Debug, Default, Clone, PartialEq, Serialize, Deserialize)] pub struct ActfAssistantContent { #[serde(default, deserialize_with = "null_as_empty_string")] pub content: String, @@ -436,7 +454,7 @@ pub struct ActfAssistantContent { pub extra: Map, } -#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)] +#[derive(Debug, Default, Clone, PartialEq, Serialize, Deserialize)] pub struct ActfMetric { #[serde(default)] pub prompt_tokens_len: Value, @@ -559,13 +577,23 @@ fn reconcile_document_tool_lists(document: &mut ActfDocument) { } impl ActfDocument { + /// Recover optional metadata from the attempts that are actually present. + pub(crate) fn normalize_for_import(&mut self) { + if self.category.trim().is_empty() { + self.category = "unknown".into(); + } + self.attempts_tried = self.attempts.len() as u64; + self.k = self.k.max(self.attempts_tried); + normalize_solved_at(&mut self.solved_at); + reconcile_document_tool_lists(self); + } + #[cfg(any(test, feature = "lance-store"))] pub fn from_json_str(input: &str) -> InputResult { let sanitized = super::common::sanitize_json_nonfinite(input); let mut document: Self = serde_json::from_str(sanitized.as_ref()) .map_err(|error| InputIssue::invalid(error.to_string()))?; - normalize_solved_at(&mut document.solved_at); - reconcile_document_tool_lists(&mut document); + document.normalize_for_import(); document.validate()?; Ok(document) } @@ -762,6 +790,38 @@ mod tests { .unwrap() } + #[test] + fn imports_incomplete_actf_metadata_without_losing_steps() { + let mut value = serde_json::to_value(fixture()).unwrap(); + for key in ["category", "correct", "solved_at"] { + value.as_object_mut().unwrap().remove(key); + } + value["k"] = json!(0); + value["attempts_tried"] = json!(99); + let attempt = value["attempts"]["1"].as_object_mut().unwrap(); + attempt.remove("correct"); + let trajectory = attempt["trajectory"].as_object_mut().unwrap(); + trajectory.remove("schema_version"); + let step = trajectory["steps"][0].as_object_mut().unwrap(); + step.remove("metric"); + let stories = crate::document::decode_json_storylines( + DocumentFormat::Actf, + &value.to_string(), + "incomplete.actf.json", + ) + .unwrap(); + assert_eq!(stories.len(), 1); + assert_eq!(stories[0].turns.len(), 1); + assert_eq!( + stories[0].turns[0].reasoning_content.as_deref(), + Some("inspect") + ); + assert_eq!(stories[0].turns[0].tool_calls.as_ref().unwrap().len(), 1); + let document = ActfDocument::from_json_str(&value.to_string()).unwrap(); + assert_eq!(document.category, "unknown"); + assert_eq!((document.k, document.attempts_tried), (1, 1)); + } + #[test] fn accepts_null_or_empty_object_trajectory_as_empty_event_log() { for trajectory in [json!(null), json!({})] { diff --git a/crates/persisting-pchronicle/src/store/blockcache/adapter.rs b/crates/persisting-pchronicle/src/store/blockcache/adapter.rs index 772473084..b9083ca3e 100644 --- a/crates/persisting-pchronicle/src/store/blockcache/adapter.rs +++ b/crates/persisting-pchronicle/src/store/blockcache/adapter.rs @@ -1,5 +1,5 @@ //! Lance bridge and ObjectStore read-through adapter. Only validated range reads -//! are cached; HEAD and full GET retain the backend's native semantics/stream. +//! are cached; explicit HEAD retains the backend's native freshness semantics. use super::{BlockCache, CacheConfig}; use crate::store::object_store_io_gate::{self as io_gate, IoKind}; @@ -13,7 +13,7 @@ use object_store::{ }; use std::{ ops::Range, - sync::{Arc, Mutex}, + sync::{Arc, Mutex, OnceLock}, time::{Duration, Instant}, }; @@ -69,6 +69,8 @@ async fn remote_write_request( struct GatedMultipartUpload { inner: Arc>>, io_scope: String, + store: CachedObjectStore, + path: Path, } impl std::fmt::Debug for GatedMultipartUpload { @@ -95,7 +97,11 @@ impl MultipartUpload for GatedMultipartUpload { async fn complete(&mut self) -> ObjectResult { let mut inner = self.inner.lock().await; - remote_write_request(&self.io_scope, inner.complete()).await + let result = remote_write_request(&self.io_scope, inner.complete()).await; + if result.is_ok() { + self.store.forget_head(&self.path); + } + result } async fn abort(&mut self) -> ObjectResult<()> { @@ -112,10 +118,34 @@ pub struct LanceCacheWrapper { } fn backend_identity_from_env() -> String { - let backend: std::collections::BTreeMap<_, _> = std::env::vars() + backend_identity(std::env::vars()) +} + +fn backend_identity(vars: impl Iterator) -> String { + let backend: std::collections::BTreeMap<_, _> = vars .filter(|(key, _)| { key.starts_with("AWS_") || key.starts_with("AZURE_") || key.starts_with("GOOGLE_") }) + .map(|(key, value)| { + // Workers give these missing files a new temporary HOME each time. + // Hash actual configuration, not an incidental pathname. Keep the + // path on other I/O errors so distinct unreadable configs stay apart. + let value = if matches!( + key.as_str(), + "AWS_CONFIG_FILE" | "AWS_SHARED_CREDENTIALS_FILE" + ) { + match std::fs::read(&value) { + Ok(bytes) => blake3::hash(&bytes).to_hex().to_string(), + Err(error) if error.kind() == std::io::ErrorKind::NotFound => { + "".into() + } + Err(_) => value, + } + } else { + value + }; + (key, value) + }) .collect(); blake3::hash(&serde_json::to_vec(&backend).unwrap_or_default()) .to_hex() @@ -143,13 +173,33 @@ impl lance_io::object_store::WrappingObjectStore for LanceCacheWrapper { } pub fn lance_wrapper(capacity_bytes: u64) -> Arc { + type WrapperKey = (CacheConfig, String); + static WRAPPERS: OnceLock< + Mutex>>, + > = OnceLock::new(); let mut config = CacheConfig::from_env(); config.capacity_bytes = capacity_bytes; - Arc::new(LanceCacheWrapper { - capacity_bytes, - block_size_bytes: config.block_size_bytes, - root: config.root, - }) + let mut wrappers = WRAPPERS + .get_or_init(Default::default) + .lock() + .unwrap_or_else(std::sync::PoisonError::into_inner); + let key = (config.clone(), backend_identity_from_env()); + // Lance keys its object-store cache by wrapper Arc identity. + if !wrappers.contains_key(&key) && wrappers.len() >= 64 { + if let Some(old) = wrappers.keys().next().cloned() { + wrappers.remove(&old); + } + } + wrappers + .entry(key) + .or_insert_with(|| { + Arc::new(LanceCacheWrapper { + capacity_bytes, + block_size_bytes: config.block_size_bytes, + root: config.root, + }) + }) + .clone() } pub fn lance_store_params(capacity_bytes: u64) -> lance_io::object_store::ObjectStoreParams { @@ -180,9 +230,10 @@ struct HeadMemo { entries: std::collections::HashMap, } -/// A path plus the version asked for, which is all that changes the answer; +/// The cache/backend namespace, path and version identify the answer; /// remaining conditions are checked against the metadata by the caller. -type HeadKey = (String, Option); +type MetadataNamespace = (std::path::PathBuf, String); +type HeadKey = (MetadataNamespace, String, Option); impl HeadMemo { fn get(&self, key: &HeadKey) -> Option<(ObjectMeta, Attributes)> { @@ -190,8 +241,9 @@ impl HeadMemo { (fetched.elapsed() < HEAD_TTL).then(|| (meta.clone(), attributes.clone())) } - fn forget(&mut self, path: &str) { - self.entries.retain(|(known, _), _| known != path); + fn forget(&mut self, namespace: &MetadataNamespace, path: &str) { + self.entries + .retain(|(scope, known, _), _| scope != namespace || known != path); } fn insert(&mut self, key: HeadKey, meta: ObjectMeta, attributes: Attributes) { @@ -213,6 +265,7 @@ pub struct CachedObjectStore { block_size_bytes: u64, store_uri: String, io_scope: String, + metadata_namespace: MetadataNamespace, heads: Arc>, } @@ -221,16 +274,27 @@ impl CachedObjectStore { /// as well as bucket. Use a digest for any sensitive configuration. pub fn new(inner: Arc, config: CacheConfig, store_uri: String) -> Self { let block_size_bytes = config.block_size_bytes; + let metadata_namespace = (config.root.clone(), store_uri.clone()); + static HEADS: OnceLock>> = OnceLock::new(); Self { inner, cache: BlockCache::new(config), block_size_bytes, io_scope: io_gate::scope_key(&store_uri), store_uri, - heads: Arc::new(Mutex::new(HeadMemo::default())), + metadata_namespace, + heads: HEADS.get_or_init(Default::default).clone(), } } + pub fn stats(&self) -> super::CacheStats { + self.cache.stats() + } + + fn forget_head(&self, path: &Path) { + self.heads().forget(&self.metadata_namespace, path.as_ref()); + } + fn heads(&self) -> std::sync::MutexGuard<'_, HeadMemo> { self.heads .lock() @@ -250,7 +314,13 @@ impl CachedObjectStore { self.cache .get_or_fetch(&file, (range.end - range.start) as usize, async { remote_request(&self.io_scope, async { - let result = self.inner.get_opts(path, options.clone()).await?; + let result = match self.inner.get_opts(path, options.clone()).await { + Ok(result) => result, + Err(error) => { + self.forget_head(path); + return Err(error); + } + }; // Do not publish mismatched bytes even if a compatible backend // ignores If-Match or the requested VersionId. if result.range != range @@ -266,7 +336,7 @@ impl CachedObjectStore { { // Remembered metadata describes bytes the backend no // longer serves; the next read must ask again. - self.heads().forget(path.as_ref()); + self.forget_head(path); return Err(object_store::Error::Precondition { path: path.to_string(), source: "object changed while reading cached block".into(), @@ -289,7 +359,11 @@ impl std::fmt::Display for CachedObjectStore { #[async_trait] impl ObjectStore for CachedObjectStore { async fn put_opts(&self, p: &Path, b: PutPayload, o: PutOptions) -> ObjectResult { - remote_write_request(&self.io_scope, self.inner.put_opts(p, b, o)).await + let result = remote_write_request(&self.io_scope, self.inner.put_opts(p, b, o)).await; + if result.is_ok() { + self.forget_head(p); + } + result } async fn put_multipart_opts( &self, @@ -301,16 +375,30 @@ impl ObjectStore for CachedObjectStore { Ok(Box::new(GatedMultipartUpload { inner: Arc::new(tokio::sync::Mutex::new(upload)), io_scope: self.io_scope.clone(), + store: self.clone(), + path: p.clone(), })) } async fn get_opts(&self, p: &Path, o: GetOptions) -> ObjectResult { - if o.head || o.range.is_none() { + // Lance rewrites this hint after commits; keep discovery fresh even + // while immutable data and versioned manifests use cached metadata. + if p.filename() == Some("latest_version_hint.json") { return remote_request(&self.io_scope, self.inner.get_opts(p, o)).await; } + let head_key = ( + self.metadata_namespace.clone(), + p.to_string(), + o.version.clone(), + ); + if o.head { + let head = remote_request(&self.io_scope, self.inner.get_opts(p, o)).await?; + self.heads() + .insert(head_key, head.meta.clone(), head.attributes.clone()); + return Ok(head); + } // Metadata for the requested version. Conditions beyond the version are // checked locally below, so remembered metadata reaches the same verdict // a conditional round trip would have returned. - let head_key = (p.to_string(), o.version.clone()); let remembered = self.heads().get(&head_key); let (head_meta, head_attributes) = match remembered { Some(head) => head, @@ -350,9 +438,18 @@ impl ObjectStore for CachedObjectStore { if o.version.is_some() && o.version != head_meta.version { return remote_request(&self.io_scope, self.inner.get_opts(p, o)).await; } - let Some(range) = o.range.as_ref() else { - return remote_request(&self.io_scope, self.inner.get_opts(p, o)).await; - }; + if head_meta.size == 0 && o.range.is_none() { + return Ok(GetResult { + payload: GetResultPayload::Stream(futures::stream::empty().boxed()), + meta: head_meta, + range: 0..0, + attributes: head_attributes, + }); + } + let range = o + .range + .clone() + .unwrap_or_else(|| (0..head_meta.size).into()); let range = range .as_range(head_meta.size) @@ -435,14 +532,17 @@ impl ObjectStore for CachedObjectStore { p: BoxStream<'static, ObjectResult>, ) -> BoxStream<'static, ObjectResult> { let uri = self.io_scope.clone(); + let this = self.clone(); let stream = self.inner.delete_stream(p); futures::stream::unfold(stream, move |mut stream| { let uri = uri.clone(); + let this = this.clone(); async move { let _permit = io_gate::acquire(&uri, IoKind::Write).await; match stream.next().await { Some(result) => { - if result.is_ok() { + if let Ok(path) = &result { + this.forget_head(path); io_gate::note_success(&uri); } else if result .as_ref() @@ -489,7 +589,11 @@ impl ObjectStore for CachedObjectStore { remote_request(&self.io_scope, self.inner.list_with_delimiter(p)).await } async fn copy_opts(&self, a: &Path, b: &Path, o: CopyOptions) -> ObjectResult<()> { - remote_write_request(&self.io_scope, self.inner.copy_opts(a, b, o)).await + let result = remote_write_request(&self.io_scope, self.inner.copy_opts(a, b, o)).await; + if result.is_ok() { + self.forget_head(b); + } + result } } @@ -498,6 +602,58 @@ mod tests { use super::*; use object_store::{ObjectStoreExt, memory::InMemory}; + #[test] + fn backend_namespace_survives_worker_home_changes_but_isolates_credentials() { + let root = tempfile::tempdir().unwrap(); + let vars = |home: &str, key: &str| { + vec![ + ( + "AWS_CONFIG_FILE".into(), + root.path() + .join(home) + .join("config") + .to_string_lossy() + .into_owned(), + ), + ( + "AWS_SHARED_CREDENTIALS_FILE".into(), + root.path() + .join(home) + .join("credentials") + .to_string_lossy() + .into_owned(), + ), + ("AWS_ACCESS_KEY_ID".into(), key.into()), + ("AWS_ENDPOINT".into(), "https://s3.example.test".into()), + ] + }; + assert_eq!( + backend_identity(vars("worker-1", "one").into_iter()), + backend_identity(vars("worker-2", "one").into_iter()) + ); + assert_ne!( + backend_identity(vars("worker-1", "one").into_iter()), + backend_identity(vars("worker-2", "two").into_iter()) + ); + std::fs::create_dir(root.path().join("worker-1")).unwrap(); + std::fs::write( + root.path().join("worker-1/config"), + b"different backend config", + ) + .unwrap(); + assert_ne!( + backend_identity(vars("worker-1", "one").into_iter()), + backend_identity(vars("worker-2", "one").into_iter()) + ); + } + + #[test] + fn lance_opens_reuse_the_wrapper_identity() { + let a = lance_wrapper(1234567); + let b = lance_wrapper(1234567); + assert!(Arc::ptr_eq(&a, &b)); + } + #[tokio::test] async fn range_cache_preserves_head_and_full_get_semantics() { let root = tempfile::tempdir().unwrap(); @@ -523,6 +679,35 @@ mod tests { cached.get(&path).await.unwrap().bytes().await.unwrap(), Bytes::from_static(b"0123456789") ); + let empty = Path::from("empty.lance"); + inner.put(&empty, Bytes::new().into()).await.unwrap(); + assert!( + cached + .get(&empty) + .await + .unwrap() + .bytes() + .await + .unwrap() + .is_empty() + ); + let hint = Path::from("dataset/_versions/latest_version_hint.json"); + inner + .put(&hint, Bytes::from_static(b"one").into()) + .await + .unwrap(); + assert_eq!( + cached.get(&hint).await.unwrap().bytes().await.unwrap(), + Bytes::from_static(b"one") + ); + inner + .put(&hint, Bytes::from_static(b"two").into()) + .await + .unwrap(); + assert_eq!( + cached.get(&hint).await.unwrap().bytes().await.unwrap(), + Bytes::from_static(b"two") + ); } /// Counts what actually leaves for the backend. @@ -616,4 +801,67 @@ mod tests { "a fully cached range must not reach the backend" ); } + #[tokio::test] + async fn full_reads_reuse_blocks_across_opens_and_mutations_invalidate_metadata() { + let root = tempfile::tempdir().unwrap(); + let memory = Arc::new(InMemory::new()); + let path = Path::from("dataset/data.lance"); + memory + .put(&path, Bytes::from_static(b"0123456789").into()) + .await + .unwrap(); + let gets = Arc::new(std::sync::atomic::AtomicUsize::new(0)); + let inner: Arc = Arc::new(Counted { + inner: memory, + gets: gets.clone(), + }); + let config = CacheConfig::new(root.path().into(), 1024, 4); + let first = + CachedObjectStore::new(inner.clone(), config.clone(), "s3://test-bucket".into()); + first.head(&path).await.unwrap(); + assert_eq!( + first.get(&path).await.unwrap().bytes().await.unwrap(), + Bytes::from_static(b"0123456789") + ); + // The explicit HEAD filled metadata, followed by exactly three blocks. + assert_eq!(gets.load(std::sync::atomic::Ordering::Relaxed), 4); + drop(first); + let second = CachedObjectStore::new(inner, config, "s3://test-bucket".into()); + assert_eq!( + second.get(&path).await.unwrap().bytes().await.unwrap(), + Bytes::from_static(b"0123456789") + ); + assert_eq!( + gets.load(std::sync::atomic::Ordering::Relaxed), + 4, + "warm reopen must not download or re-HEAD" + ); + assert_eq!(second.stats().hit_bytes, 10); + assert_eq!(second.stats().downloaded_bytes, 10); + second + .put(&path, Bytes::from_static(b"abcdefghij").into()) + .await + .unwrap(); + assert_eq!( + second.get_range(&path, 0..4).await.unwrap(), + Bytes::from_static(b"abcd") + ); + assert!(gets.load(std::sync::atomic::Ordering::Relaxed) > 4); + second.copy(&path, &Path::from("copy.lance")).await.unwrap(); + assert_eq!( + second + .get(&Path::from("copy.lance")) + .await + .unwrap() + .bytes() + .await + .unwrap(), + Bytes::from_static(b"abcdefghij") + ); + second.delete(&path).await.unwrap(); + assert!(matches!( + second.get_range(&path, 0..4).await, + Err(object_store::Error::NotFound { .. }) + )); + } } diff --git a/crates/persisting-pchronicle/src/store/blockcache/block.rs b/crates/persisting-pchronicle/src/store/blockcache/block.rs index 05f68c5bb..db648a99d 100644 --- a/crates/persisting-pchronicle/src/store/blockcache/block.rs +++ b/crates/persisting-pchronicle/src/store/blockcache/block.rs @@ -5,7 +5,7 @@ use std::{ future::Future, path::{Path, PathBuf}, sync::{ - Arc, + Arc, Mutex, OnceLock, atomic::{AtomicU64, Ordering}, }, }; @@ -15,6 +15,11 @@ pub struct CacheStats { pub hits: u64, pub misses: u64, pub evictions: u64, + pub hit_bytes: u64, + pub downloaded_bytes: u64, + pub coalesced_reads: u64, + pub write_errors: u64, + pub resident_bytes: u64, } #[derive(Debug, Default)] @@ -22,30 +27,62 @@ struct Counters { hits: AtomicU64, misses: AtomicU64, evictions: AtomicU64, + hit_bytes: AtomicU64, + downloaded_bytes: AtomicU64, + coalesced_reads: AtomicU64, + write_errors: AtomicU64, + resident_bytes: AtomicU64, } #[derive(Debug, Clone)] pub struct BlockCache { config: CacheConfig, - counters: Arc, - flights: Arc>>>>, - last_trim_ms: Arc, + shared: Arc, +} + +#[derive(Debug, Default)] +struct Shared { + counters: Counters, + // ponytail: downloads coalesce within a process; add per-block file locks + // if multiple worker processes contend on the same cold working set. + flights: tokio::sync::Mutex>>>, + last_trim_ms: AtomicU64, + trim_lock: tokio::sync::Mutex<()>, + accesses: Mutex>, + writes_since_trim: AtomicU64, } impl BlockCache { pub fn new(config: CacheConfig) -> Self { - Self { - config, - counters: Arc::new(Counters::default()), - flights: Arc::new(tokio::sync::Mutex::new(HashMap::new())), - last_trim_ms: Arc::new(AtomicU64::new(0)), + static CACHES: OnceLock>>> = OnceLock::new(); + let mut caches = CACHES + .get_or_init(Default::default) + .lock() + .unwrap_or_else(std::sync::PoisonError::into_inner); + // ponytail: retain 64 cache configurations per process; use an LRU of + // configurations if a service actually rotates through more roots. + if !caches.contains_key(&config) && caches.len() >= 64 { + if let Some(key) = caches.keys().next().cloned() { + caches.remove(&key); + } } + let shared = caches.entry(config.clone()).or_default().clone(); + Self { config, shared } } pub fn stats(&self) -> CacheStats { CacheStats { - hits: self.counters.hits.load(Ordering::Relaxed), - misses: self.counters.misses.load(Ordering::Relaxed), - evictions: self.counters.evictions.load(Ordering::Relaxed), + hits: self.shared.counters.hits.load(Ordering::Relaxed), + misses: self.shared.counters.misses.load(Ordering::Relaxed), + evictions: self.shared.counters.evictions.load(Ordering::Relaxed), + hit_bytes: self.shared.counters.hit_bytes.load(Ordering::Relaxed), + downloaded_bytes: self + .shared + .counters + .downloaded_bytes + .load(Ordering::Relaxed), + coalesced_reads: self.shared.counters.coalesced_reads.load(Ordering::Relaxed), + write_errors: self.shared.counters.write_errors.load(Ordering::Relaxed), + resident_bytes: self.shared.counters.resident_bytes.load(Ordering::Relaxed), } } pub fn block_path(&self, key: &str, block: u64) -> PathBuf { @@ -69,19 +106,20 @@ impl BlockCache { where F: Future>, { - // Reclaim stale files after a restart as well as after new downloads. - self.schedule_trim(); match tokio::fs::read(path).await { Ok(bytes) if bytes.len() == expected => { - self.counters.hits.fetch_add(1, Ordering::Relaxed); + self.record_hit(path, bytes.len()).await; + // Refresh recency before restart cleanup can choose victims. + self.schedule_trim(); return Ok(Bytes::from(bytes)); } - _ => { - self.counters.misses.fetch_add(1, Ordering::Relaxed); - } + _ => self.schedule_trim(), }; let flight = { - let mut flights = self.flights.lock().await; + let mut flights = self.shared.flights.lock().await; + // Cancelled fetches leave only the map's Arc; reclaim them on the + // next miss rather than retaining every cancelled object path. + flights.retain(|_, flight| Arc::strong_count(flight) > 1); flights .entry(path.to_path_buf()) .or_insert_with(|| Arc::new(tokio::sync::Mutex::new(()))) @@ -92,9 +130,11 @@ impl BlockCache { if let Ok(bytes) = tokio::fs::read(path).await && bytes.len() == expected { - self.counters.hits.fetch_add(1, Ordering::Relaxed); + self.shared.counters.coalesced_reads.fetch_add(1, Ordering::Relaxed); + self.record_hit(path, bytes.len()).await; return Ok(Bytes::from(bytes)); } + self.shared.counters.misses.fetch_add(1, Ordering::Relaxed); let bytes = fetch.await?; if bytes.len() != expected { return Err(object_store::Error::Generic { @@ -102,21 +142,35 @@ impl BlockCache { source: format!("block length {}, expected {expected}", bytes.len()).into(), }); } - if tokio::fs::create_dir_all(path.parent().unwrap_or(&self.config.root)) - .await - .is_ok() - { - let tmp = path.with_extension(format!("{}.tmp", std::process::id())); - if tokio::fs::write(&tmp, &bytes).await.is_ok() { - let _ = tokio::fs::rename(&tmp, path).await; - self.schedule_trim(); + self.shared.counters.downloaded_bytes.fetch_add(bytes.len() as u64, Ordering::Relaxed); + tracing::debug!(target: "pchronicle.block_cache", block = %path.display(), bytes = bytes.len(), "block cache miss"); + // A block larger than the entire budget would evict useful data and + // immediately evict itself. Serve it without polluting the cache. + if expected as u64 <= self.config.capacity_bytes { + let tmp = path.with_extension(format!("{}.{}.tmp", std::process::id(), uuid::Uuid::new_v4())); + let write = async { + tokio::fs::create_dir_all(path.parent().unwrap_or(&self.config.root)).await?; + tokio::fs::write(&tmp, &bytes).await?; + tokio::fs::rename(&tmp, path).await + }.await; + match write { + Ok(()) => { + self.accesses().insert(path.to_path_buf(), std::time::SystemTime::now()); + self.shared.writes_since_trim.fetch_add(bytes.len() as u64, Ordering::Relaxed); + self.schedule_trim(); + } + Err(error) => { + let _ = tokio::fs::remove_file(&tmp).await; + self.shared.counters.write_errors.fetch_add(1, Ordering::Relaxed); + tracing::warn!(target: "pchronicle.block_cache", %error, "cache write failed; serving downloaded bytes"); + } } } Ok(bytes) } .await; drop(flight_guard); - let mut flights = self.flights.lock().await; + let mut flights = self.shared.flights.lock().await; if Arc::strong_count(&flight) == 2 && flights .get(path) @@ -126,13 +180,43 @@ impl BlockCache { } result } + fn accesses(&self) -> std::sync::MutexGuard<'_, HashMap> { + self.shared + .accesses + .lock() + .unwrap_or_else(std::sync::PoisonError::into_inner) + } + + async fn record_hit(&self, path: &Path, size: usize) { + self.shared.counters.hits.fetch_add(1, Ordering::Relaxed); + self.shared + .counters + .hit_bytes + .fetch_add(size as u64, Ordering::Relaxed); + let now = std::time::SystemTime::now(); + self.accesses().insert(path.to_path_buf(), now); + tracing::debug!(target: "pchronicle.block_cache", block = %path.display(), bytes = size, "block cache hit"); + // Persist recency so the hot working set survives worker restarts. + let path = path.to_path_buf(); + let _ = tokio::task::spawn_blocking(move || { + std::fs::File::options() + .write(true) + .open(path)? + .set_times(std::fs::FileTimes::new().set_modified(now)) + }) + .await; + } + fn schedule_trim(&self) { let now = std::time::SystemTime::now() .duration_since(std::time::UNIX_EPOCH) .map_or(0, |duration| duration.as_millis() as u64); - let last = self.last_trim_ms.load(Ordering::Relaxed); - if now.saturating_sub(last) < 5_000 + let last = self.shared.last_trim_ms.load(Ordering::Relaxed); + let full = self.shared.writes_since_trim.load(Ordering::Relaxed) + >= (self.config.capacity_bytes / 10).max(1); + if (!full && now.saturating_sub(last) < 5_000) || self + .shared .last_trim_ms .compare_exchange(last, now, Ordering::Relaxed, Ordering::Relaxed) .is_err() @@ -140,9 +224,23 @@ impl BlockCache { return; } let cache = self.clone(); - tokio::spawn(async move { cache.trim().await }); + tokio::spawn(async move { + let Ok(_guard) = cache.shared.trim_lock.try_lock() else { + return; + }; + cache.trim_inner().await; + }); } pub async fn trim(&self) { + let _guard = self.shared.trim_lock.lock().await; + self.trim_inner().await; + } + + async fn trim_inner(&self) { + self.shared.writes_since_trim.swap(0, Ordering::Relaxed); + // ponytail: reconcile by scanning disk, including other workers' files; + // use a persistent block index if millions of blocks make this costly. + let scan_started = std::time::SystemTime::now(); let mut files = Vec::new(); let mut total = 0; let mut dirs = vec![self.config.root.clone()]; @@ -153,31 +251,188 @@ impl BlockCache { while let Ok(Some(entry)) = entries.next_entry().await { let path = entry.path(); if let Ok(meta) = entry.metadata().await { + if meta.file_type().is_symlink() { + continue; + } if meta.is_dir() { dirs.push(path); - } else if meta.is_file() && path.extension().is_none_or(|ext| ext != "tmp") { + } else if meta.is_file() && path.extension().is_some_and(|ext| ext == "tmp") { + let owned = + path.file_name() + .and_then(|name| name.to_str()) + .is_some_and(|name| { + name.get(..64).is_some_and(|hash| { + hash.bytes().all(|c| c.is_ascii_hexdigit()) + }) && name.as_bytes().get(64) == Some(&b'.') + }); + if owned + && meta + .modified() + .ok() + .and_then(|time| time.elapsed().ok()) + .is_some_and(|age| age > std::time::Duration::from_secs(300)) + { + let _ = tokio::fs::remove_file(path).await; + } + } else if meta.is_file() { total += meta.len(); - files.push((meta.modified().ok(), meta.len(), path)); + let modified = meta.modified().ok(); + let recent = self.accesses().get(&path).copied().max(modified); + files.push((recent, modified, meta.len(), path)); } } } } - files.sort_by_key(|(mtime, _, _)| *mtime); - for (_, size, path) in files { - if total <= self.config.capacity_bytes { + let present = files + .iter() + .map(|(_, _, _, path)| path.clone()) + .collect::>(); + self.accesses() + .retain(|path, time| present.contains(path) || *time >= scan_started); + files.sort_by_key(|(recent, _, _, _)| *recent); + // Evict at 90% to an 80% target, leaving space for the next downloads. + let reserve = self.config.capacity_bytes / 10; + let target = if total >= self.config.capacity_bytes - reserve { + self.config.capacity_bytes - 2 * reserve + } else { + self.config.capacity_bytes + }; + for (recent, modified, size, path) in files { + if total <= target { break; } - if tokio::fs::remove_file(path).await.is_ok() { + if self.shared.flights.lock().await.contains_key(&path) { + continue; + } + // Another reader/process may have refreshed this block since the + // scan; do not delete its new working set based on an old snapshot. + if self + .accesses() + .get(&path) + .copied() + .is_some_and(|now| Some(now) > recent) + || tokio::fs::metadata(&path) + .await + .ok() + .and_then(|meta| meta.modified().ok()) + > modified + { + continue; + } + if tokio::fs::remove_file(&path).await.is_ok() { + self.accesses().remove(&path); total -= size; - self.counters.evictions.fetch_add(1, Ordering::Relaxed); + self.shared + .counters + .evictions + .fetch_add(1, Ordering::Relaxed); } } + self.shared + .counters + .resident_bytes + .store(total, Ordering::Relaxed); + tracing::debug!(target: "pchronicle.block_cache", root = %self.config.root.display(), capacity_bytes = self.config.capacity_bytes, stats = ?self.stats(), "block cache status"); } } #[cfg(test)] mod tests { use super::*; + #[tokio::test] + async fn lru_survives_losing_the_in_memory_recency() { + let dir = tempfile::tempdir().unwrap(); + let cache = BlockCache::new(CacheConfig::new(dir.path().into(), 8, 4)); + cache.shared.last_trim_ms.store(u64::MAX, Ordering::Relaxed); + let hot = dir.path().join("hot"); + let cold = dir.path().join("cold"); + for (path, age) in [(&hot, 1), (&cold, 2)] { + std::fs::write(path, b"data").unwrap(); + std::fs::File::options() + .write(true) + .open(path) + .unwrap() + .set_times( + std::fs::FileTimes::new() + .set_modified(std::time::UNIX_EPOCH + std::time::Duration::from_secs(age)), + ) + .unwrap(); + } + cache + .get_or_fetch(&hot, 4, async { panic!("hot block must hit") }) + .await + .unwrap(); + let fresh = dir.path().join("fresh"); + std::fs::write(&fresh, b"data").unwrap(); + // Simulate a restart: only the on-disk access times remain. + cache.accesses().clear(); + cache.trim().await; + assert!(hot.exists()); + assert!(fresh.exists()); + assert!(!cold.exists()); + assert_eq!(cache.stats().resident_bytes, 8); + assert_eq!(cache.stats().evictions, 1); + } + + #[tokio::test] + async fn high_watermark_eviction_leaves_room_before_capacity_is_exceeded() { + let dir = tempfile::tempdir().unwrap(); + let cache = BlockCache::new(CacheConfig::new(dir.path().into(), 100, 30)); + for name in ["a", "b", "c"] { + std::fs::write(dir.path().join(name), [0; 30]).unwrap(); + } + cache.trim().await; + assert_eq!(cache.stats().evictions, 1); + assert_eq!(cache.stats().resident_bytes, 60); + } + + #[tokio::test] + async fn independently_opened_caches_coalesce_downloads_and_share_stats() { + let dir = tempfile::tempdir().unwrap(); + let config = CacheConfig::new(dir.path().into(), 1024, 4); + let first = BlockCache::new(config.clone()); + let second = BlockCache::new(config); + let path = first.block_path("object", 0); + let downloads = AtomicU64::new(0); + let fetch = || async { + downloads.fetch_add(1, Ordering::Relaxed); + tokio::time::sleep(std::time::Duration::from_millis(20)).await; + Ok(Bytes::from_static(b"data")) + }; + let (a, b) = tokio::join!( + first.get_or_fetch(&path, 4, fetch()), + second.get_or_fetch(&path, 4, fetch()) + ); + assert_eq!(a.unwrap(), b.unwrap()); + assert_eq!(downloads.load(Ordering::Relaxed), 1); + assert_eq!(first.stats().misses, 1); + assert_eq!(second.stats().downloaded_bytes, 4); + assert_eq!(second.stats().hits, 1); + } + + #[tokio::test] + async fn cache_write_failure_and_oversized_blocks_do_not_fail_reads() { + let blocked = tempfile::NamedTempFile::new().unwrap(); + let cache = BlockCache::new(CacheConfig::new(blocked.path().into(), 1024, 4)); + let path = cache.block_path("object", 0); + assert_eq!( + cache + .get_or_fetch(&path, 4, async { Ok(Bytes::from_static(b"data")) }) + .await + .unwrap(), + Bytes::from_static(b"data") + ); + assert_eq!(cache.stats().write_errors, 1); + let dir = tempfile::tempdir().unwrap(); + let cache = BlockCache::new(CacheConfig::new(dir.path().into(), 2, 4)); + let path = cache.block_path("oversized", 0); + cache + .get_or_fetch(&path, 4, async { Ok(Bytes::from_static(b"data")) }) + .await + .unwrap(); + assert!(!path.exists()); + } + #[tokio::test] async fn hit_miss_and_corrupt_are_observable() { let dir = tempfile::tempdir().unwrap(); @@ -205,12 +460,17 @@ mod tests { .unwrap(), Bytes::from_static(b"efgh") ); + let mut stats = cache.stats(); + stats.resident_bytes = 0; // Background reconciliation is asynchronous. assert_eq!( - cache.stats(), + stats, CacheStats { hits: 1, misses: 2, - evictions: 0 + evictions: 0, + hit_bytes: 4, + downloaded_bytes: 8, + ..Default::default() } ); } diff --git a/crates/persisting-pchronicle/src/store/blockcache/config.rs b/crates/persisting-pchronicle/src/store/blockcache/config.rs index 6e7ac3c01..41c10e240 100644 --- a/crates/persisting-pchronicle/src/store/blockcache/config.rs +++ b/crates/persisting-pchronicle/src/store/blockcache/config.rs @@ -2,9 +2,9 @@ use std::path::PathBuf; pub const DEFAULT_CAPACITY_BYTES: u64 = 512 * 1024 * 1024; pub const SERVE_CAPACITY_BYTES: u64 = 8 * DEFAULT_CAPACITY_BYTES; -pub const DEFAULT_BLOCK_SIZE_BYTES: u64 = 8 * 1024 * 1024; +pub const DEFAULT_BLOCK_SIZE_BYTES: u64 = 1024 * 1024; -#[derive(Debug, Clone)] +#[derive(Debug, Clone, PartialEq, Eq, Hash)] pub struct CacheConfig { pub root: PathBuf, pub capacity_bytes: u64, @@ -27,7 +27,12 @@ impl CacheConfig { .and_then(|v| v.parse().ok()) .filter(|v| *v > 0) .unwrap_or(DEFAULT_CAPACITY_BYTES); - Self::new(default_cache_dir(), capacity, DEFAULT_BLOCK_SIZE_BYTES) + let block_size = std::env::var("PCHRONICLE_LANCE_CACHE_BLOCK_SIZE_BYTES") + .ok() + .and_then(|v| v.parse().ok()) + .filter(|v| *v > 0) + .unwrap_or(DEFAULT_BLOCK_SIZE_BYTES); + Self::new(default_cache_dir(), capacity, block_size) } } diff --git a/crates/persisting-pchronicle/src/store/files/actf_reader.rs b/crates/persisting-pchronicle/src/store/files/actf_reader.rs index 09baad47b..918981848 100644 --- a/crates/persisting-pchronicle/src/store/files/actf_reader.rs +++ b/crates/persisting-pchronicle/src/store/files/actf_reader.rs @@ -79,12 +79,13 @@ fn push_actf_stories_from_slice( stories: &mut Vec, ) -> io::Result<()> { let mut deserializer = serde_json::Deserializer::from_slice(record); - let document = ActfDocument::deserialize(&mut deserializer) + let mut document = ActfDocument::deserialize(&mut deserializer) .and_then(|document| { deserializer.end()?; Ok(document) }) .map_err(|error| io::Error::new(io::ErrorKind::InvalidData, error))?; + document.normalize_for_import(); document .validate() .map_err(|error| io::Error::new(io::ErrorKind::InvalidData, error.to_string()))?; @@ -104,11 +105,12 @@ fn extend_actf_stories( fn deserialize_actf_document(reader: R) -> Result { let mut deserializer = serde_json::Deserializer::from_reader(reader); - let document = + let mut document = ActfDocument::deserialize(&mut deserializer).context("deserialize ACTF document")?; deserializer .end() .context("finish ACTF document deserialization")?; + document.normalize_for_import(); document .validate() .map_err(|error| anyhow::anyhow!(error.to_string()))?; diff --git a/crates/persisting-pchronicle/src/store/files/actf_stream.rs b/crates/persisting-pchronicle/src/store/files/actf_stream.rs index 187e2d80a..56b321c3b 100644 --- a/crates/persisting-pchronicle/src/store/files/actf_stream.rs +++ b/crates/persisting-pchronicle/src/store/files/actf_stream.rs @@ -225,7 +225,7 @@ impl<'de> Visitor<'de> for ProjectedActfAttemptVisitor<'_> { } } } - trajectory.ok_or_else(|| de::Error::missing_field("trajectory")) + trajectory.ok_or_else(|| de::Error::custom(ACTF_TRAJECTORY_NOT_PROJECTABLE)) } } @@ -233,8 +233,7 @@ struct ProjectedActfTrajectorySeed<'a> { scan: &'a FileScanSpec, } -pub(super) const ACTF_TRAJECTORY_NOT_PROJECTABLE: &str = - "ACTF trajectory is an event log; use full decode"; +pub(super) const ACTF_TRAJECTORY_NOT_PROJECTABLE: &str = "ACTF trajectory requires full decode"; impl<'de> DeserializeSeed<'de> for ProjectedActfTrajectorySeed<'_> { type Value = ProjectedActfAttempt; @@ -258,6 +257,10 @@ impl<'de> Visitor<'de> for ProjectedActfTrajectoryVisitor<'_> { formatter.write_str("an ACTF trajectory object or event-log array") } + fn visit_unit(self) -> std::result::Result { + Err(de::Error::custom(ACTF_TRAJECTORY_NOT_PROJECTABLE)) + } + fn visit_seq(self, mut sequence: A) -> std::result::Result where A: SeqAccess<'de>, @@ -281,7 +284,7 @@ impl<'de> Visitor<'de> for ProjectedActfTrajectoryVisitor<'_> { } } } - steps.ok_or_else(|| de::Error::missing_field("steps")) + steps.ok_or_else(|| de::Error::custom(ACTF_TRAJECTORY_NOT_PROJECTABLE)) } } @@ -322,7 +325,7 @@ impl<'de> Visitor<'de> for ProjectedActfStepsVisitor<'_> { steps.push(step); } if steps.is_empty() { - return Err(de::Error::custom("ACTF trajectory steps must not be empty")); + return Err(de::Error::custom(ACTF_TRAJECTORY_NOT_PROJECTABLE)); } Ok(ProjectedActfAttempt { steps }) } @@ -414,7 +417,8 @@ impl<'de> Visitor<'de> for ProjectedActfStepVisitor<'_> { Ok(ProjectedActfStep { step_id: step_id.ok_or_else(|| de::Error::missing_field("step_id"))?, - started_at: started_at.ok_or_else(|| de::Error::missing_field("started_at"))?, + started_at: started_at + .ok_or_else(|| de::Error::custom(ACTF_TRAJECTORY_NOT_PROJECTABLE))?, content, reasoning_content, tools_nonempty, @@ -440,7 +444,7 @@ impl<'de> DeserializeSeed<'de> for ProjectedActfAssistantContentSeed<'_> { where D: serde::Deserializer<'de>, { - deserializer.deserialize_map(ProjectedActfAssistantContentVisitor { scan: self.scan }) + deserializer.deserialize_any(ProjectedActfAssistantContentVisitor { scan: self.scan }) } } @@ -455,6 +459,13 @@ impl<'de> Visitor<'de> for ProjectedActfAssistantContentVisitor<'_> { formatter.write_str("ACTF assistant_content object") } + fn visit_unit(self) -> std::result::Result { + Ok(ProjectedActfAssistantContent { + content: String::new(), + reasoning_content: None, + }) + } + fn visit_map(self, mut map: A) -> std::result::Result where A: MapAccess<'de>, diff --git a/crates/persisting-pchronicle/src/store/files/tests.rs b/crates/persisting-pchronicle/src/store/files/tests.rs index 064045a8e..c46a9a56f 100644 --- a/crates/persisting-pchronicle/src/store/files/tests.rs +++ b/crates/persisting-pchronicle/src/store/files/tests.rs @@ -248,6 +248,44 @@ async fn full_atif_array_reports_bounded_input_buffer_peak() { assert!(peak < 2 * 64 * 1024, "peak={peak}"); } +#[tokio::test] +async fn queries_incomplete_actf_with_projected_and_full_decode() { + for trajectory in [ + serde_json::json!({"steps":[{"step_id":1,"assistant_content":null}]}), + serde_json::json!(null), + ] { + let input = tempfile::NamedTempFile::with_suffix(".json").unwrap(); + std::fs::write( + input.path(), + serde_json::json!({ + "task_id":"incomplete", + "attempts":{"1":{"trajectory":trajectory}} + }) + .to_string(), + ) + .unwrap(); + let manifest = LocalQueryManifest::for_format(input.path(), DocumentFormat::Actf).unwrap(); + let source = FileTrajectoryDataSource::from_manifest(manifest).unwrap(); + let context = SessionContext::new(); + source.register(&context).unwrap(); + for sql in [ + "SELECT session_id FROM runs", + "SELECT session_id, message_json FROM steps", + ] { + let batches = context.sql(sql).await.unwrap().collect().await.unwrap(); + let expected = if sql.contains("runs") || !trajectory.is_null() { + 1 + } else { + 0 + }; + assert_eq!( + batches.iter().map(RecordBatch::num_rows).sum::(), + expected + ); + } + } +} + #[tokio::test] async fn queries_actf_event_log_trajectory_as_steps() { let input = tempfile::NamedTempFile::with_suffix(".json").unwrap(); diff --git a/docs/src/en/pchronicle/guides/troubleshooting.md b/docs/src/en/pchronicle/guides/troubleshooting.md index 75d4a5364..b32156c59 100644 --- a/docs/src/en/pchronicle/guides/troubleshooting.md +++ b/docs/src/en/pchronicle/guides/troubleshooting.md @@ -74,6 +74,47 @@ exchange guide to import into a Dataset pChronicle can normalize. Import does not invent missing lineage or Evidence; preserve the original Source alongside the normalized view when provenance matters. +## Verify the local block cache + +Remote Lance reads use a persistent, version-keyed disk cache. Blocks are 1 MiB +by default; range and full-file reads share the same blocks. Reopening a Dataset +in the same process reuses metadata and concurrent downloads. Cache hits update +persisted access times, so eviction retains recently used blocks after a worker +restart. At 90% capacity, background eviction targets 80% usage, leaving room for new downloads. + +| Environment variable | Default | Purpose | +| --- | --- | --- | +| `PCHRONICLE_LANCE_CACHE_DIR` | OS cache directory, under `pchronicle/blocks` | Shared cache location; also honored by catalog workers | +| `PCHRONICLE_LANCE_CACHE_CAPACITY_BYTES` | `536870912` (512 MiB) | Disk capacity target | +| `PCHRONICLE_LANCE_CACHE_BLOCK_SIZE_BYTES` | `1048576` (1 MiB) | Aligned download/cache block size | + +Run the same remote query twice with debug logging: + +```bash +pchronicle --log-level debug query @prod \ + --sql 'SELECT COUNT(*) FROM dataset.steps' +``` + +Look for `pchronicle.block_cache`: `block cache miss` records downloaded blocks, +`block cache hit` records disk reads, and `block cache status` reports hits, +misses, hit/downloaded bytes, coalesced reads, evictions, write errors and disk +usage at the latest reconciliation. `CachedObjectStore::stats()` exposes the +same counters, shared by stores with the same cache configuration in a process. +A second CLI process reuses disk blocks but still discovers remote metadata. + +Metadata is reused for up to 60 seconds. Explicit HEAD requests and Lance's +mutable version hint retain remote freshness; successful writes, multipart +completion, copies and deletes invalidate remembered metadata. Local disk write +failures are logged and the downloaded bytes are still returned. A block larger +than the configured capacity is served without caching. + +This uses the aligned-block and persistent LRU principles described in +[Doris file cache internals](https://doris.apache.org/docs/3.x/compute-storage-decoupled/file-cache/file-cache-internals/). +Eviction reconciles disk contents periodically, including files written by other +workers. Downloads are coalesced within a process; separate processes can still +download the same cold block concurrently. Priority queues and prewarming are +not implemented. + ## Before opening an issue Include the pChronicle version, Dataset path or pin name (without credentials), diff --git a/docs/src/zh/pchronicle/guides/troubleshooting.md b/docs/src/zh/pchronicle/guides/troubleshooting.md index 30887049c..57cb323a3 100644 --- a/docs/src/zh/pchronicle/guides/troubleshooting.md +++ b/docs/src/zh/pchronicle/guides/troubleshooting.md @@ -68,6 +68,41 @@ CI 中使用 `--file` 和明确的输出限制。需要更大预算时，应在 可以规范化的 Dataset。导入不会补造缺失的 lineage 或 Evidence；需要追溯时，请保留原始 Source 与规范化视图。 +## 确认本地块缓存是否生效 + +远端 Lance 读取使用按对象版本区分的持久化磁盘缓存，默认块大小为 1 MiB。 +范围读取和整文件读取复用同一套块；同一进程重新打开 Dataset 会复用元数据并合并并发下载。 +命中会更新持久化访问时间，worker 重启后仍能优先保留近期使用的块。 +占用达到容量的 90% 后，后台淘汰以 80% 为目标，为后续下载预留空间。 + +| 环境变量 | 默认值 | 用途 | +| --- | --- | --- | +| `PCHRONICLE_LANCE_CACHE_DIR` | 系统缓存目录下的 `pchronicle/blocks` | 共享缓存目录，catalog worker 也遵循此设置 | +| `PCHRONICLE_LANCE_CACHE_CAPACITY_BYTES` | `536870912`（512 MiB） | 磁盘缓存容量目标 | +| `PCHRONICLE_LANCE_CACHE_BLOCK_SIZE_BYTES` | `1048576`（1 MiB） | 对齐下载与缓存的块大小 | + +打开 debug 日志，重复执行同一条远端查询： + +```bash +pchronicle --log-level debug query @prod \ + --sql 'SELECT COUNT(*) FROM dataset.steps' +``` + +查看 `pchronicle.block_cache` 日志：`block cache miss` 表示下载块， +`block cache hit` 表示从磁盘命中；`block cache status` 给出命中、未命中、 +命中/下载字节、合并读取、淘汰、写入失败次数，以及最近一次磁盘扫描统计的占用。 +`CachedObjectStore::stats()` 提供同样的指标；同一进程内相同缓存配置共享计数。 +另一个 CLI 进程会复用磁盘块，仍会查询远端元数据。 + +元数据最多复用 60 秒。显式 HEAD 和 Lance 的可变版本提示仍查询远端； +写入、分片上传完成、复制或删除成功后，会使相关元数据失效。 +本地缓存写入失败会记录日志，已下载的数据仍正常返回；大于总容量的块直接返回而不缓存。 + +实现参考了 [Doris 文件缓存](https://doris.apache.org/docs/3.x/compute-storage-decoupled/file-cache/file-cache-internals/) +的对齐分块和持久化 LRU 思路。淘汰定期扫描磁盘，也统计其他 worker 写入的文件。 +下载合并作用于同一进程；不同进程仍可能并发下载同一个冷块。 +尚未实现优先级队列与预热。 + ## 提交 issue 前 请提供 pChronicle 版本、Dataset 路径或 pin 名称（不要包含凭据）、`status --format json` 输出、 From f210c3346def7c9f1c5e64a0a39ef1705ec6db39 Mon Sep 17 00:00:00 2001 From: Reiase Date: Fri, 2 Oct 2026 22:10:53 +0800 Subject: [PATCH 2/2] Fix Clippy conditions in shared cache registries --- .../src/store/blockcache/adapter.rs | 9 +++++---- .../persisting-pchronicle/src/store/blockcache/block.rs | 9 +++++---- 2 files changed, 10 insertions(+), 8 deletions(-) diff --git a/crates/persisting-pchronicle/src/store/blockcache/adapter.rs b/crates/persisting-pchronicle/src/store/blockcache/adapter.rs index b9083ca3e..371ae3c22 100644 --- a/crates/persisting-pchronicle/src/store/blockcache/adapter.rs +++ b/crates/persisting-pchronicle/src/store/blockcache/adapter.rs @@ -185,10 +185,11 @@ pub fn lance_wrapper(capacity_bytes: u64) -> Arc= 64 { - if let Some(old) = wrappers.keys().next().cloned() { - wrappers.remove(&old); - } + if !wrappers.contains_key(&key) + && wrappers.len() >= 64 + && let Some(old) = wrappers.keys().next().cloned() + { + wrappers.remove(&old); } wrappers .entry(key) diff --git a/crates/persisting-pchronicle/src/store/blockcache/block.rs b/crates/persisting-pchronicle/src/store/blockcache/block.rs index db648a99d..7f03dd061 100644 --- a/crates/persisting-pchronicle/src/store/blockcache/block.rs +++ b/crates/persisting-pchronicle/src/store/blockcache/block.rs @@ -61,10 +61,11 @@ impl BlockCache { .unwrap_or_else(std::sync::PoisonError::into_inner); // ponytail: retain 64 cache configurations per process; use an LRU of // configurations if a service actually rotates through more roots. - if !caches.contains_key(&config) && caches.len() >= 64 { - if let Some(key) = caches.keys().next().cloned() { - caches.remove(&key); - } + if !caches.contains_key(&config) + && caches.len() >= 64 + && let Some(key) = caches.keys().next().cloned() + { + caches.remove(&key); } let shared = caches.entry(config.clone()).or_default().clone(); Self { config, shared }