From 9b4997d7b9e4780bf70f45130b54d48b786469c9 Mon Sep 17 00:00:00 2001 From: Sam Horradarn Date: Fri, 4 Sep 2026 13:11:38 +0200 Subject: [PATCH 1/8] feat(simulation): let the simulation store S3-style object tag sets on xorbs MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit `DeletionControlableClient` gains `get_xorb_tag_set` / `set_xorb_tag_set`, carrying an `ObjectTagSet` of ordered `(key, value)` pairs. Writes replace the whole set, as S3 `PutObjectTagging` does. This is deliberately separate from the existing 32-byte `ObjectTag`: writing a tag set leaves the object's bytes, and so its `ObjectTag`, untouched. That is the property a caller needs to record something about an object without invalidating anything keyed on its content — xet-garbage-collection uses it to distinguish a xorb re-uploaded since its inventory snapshot from one that has sat untouched, which it previously could only infer from the ETag moving. Implemented for both backing clients and exposed over the local server as `GET`/`PUT /simulation/xorbs/{hash}/tag_set`, so `SimulationControlClient` picks it up like the other deletion controls. `MemoryClient` holds the sets beside `xorbs`; `LocalClient` writes a `.tagset` JSON sidecar, matching the existing `.gctag` convention of keeping GC state next to the object rather than in it. Tagging an absent xorb is an error rather than leaving an orphan sidecar. Tests cover the round trip, wholesale replacement, that the `ObjectTag` and both xorb listings are unaffected, and the absent-xorb error, for both clients. Co-Authored-By: Claude Opus 5 --- .../simulation/deletion_controls.rs | 13 +++ .../src/cas_client/simulation/local_client.rs | 91 ++++++++++++++++++- .../local_server/simulation_control_client.rs | 30 +++++- .../local_server/simulation_handlers.rs | 32 ++++++- .../local_server/simulation_types.rs | 9 +- .../cas_client/simulation/memory_client.rs | 62 ++++++++++++- xet_client/src/cas_client/simulation/mod.rs | 2 +- 7 files changed, 231 insertions(+), 8 deletions(-) diff --git a/xet_client/src/cas_client/simulation/deletion_controls.rs b/xet_client/src/cas_client/simulation/deletion_controls.rs index 35f4585cb..a965ae4c4 100644 --- a/xet_client/src/cas_client/simulation/deletion_controls.rs +++ b/xet_client/src/cas_client/simulation/deletion_controls.rs @@ -10,6 +10,13 @@ use crate::error::Result; /// to reduce false matches when objects are rapidly rewritten. pub type ObjectTag = [u8; 32]; +/// S3-style `(key, value)` tag set attached to an object. +/// +/// Distinct from [`ObjectTag`]: writing this leaves the object's bytes, and so its +/// [`ObjectTag`], untouched. That is what lets a caller record something about an object +/// without invalidating anything keyed on its content. +pub type ObjectTagSet = Vec<(String, String)>; + /// Trait for clients that support deletion and integrity operations on shards and file entries. /// /// Implemented by `LocalClient` (disk-backed) and `MemoryClient` (in-memory). @@ -48,6 +55,12 @@ pub trait DeletionControlableClient: Send + Sync { /// Returns `Ok(true)` if deleted, `Ok(false)` if the tag did not match. async fn delete_xorb_if_tag_matches(&self, hash: &MerkleHash, tag: &ObjectTag) -> Result; + /// Returns a XORB's tag set, empty if it has none. + async fn get_xorb_tag_set(&self, hash: &MerkleHash) -> Result; + + /// Replaces a XORB's tag set wholesale, as S3 `PutObjectTagging` does. + async fn set_xorb_tag_set(&self, hash: &MerkleHash, tags: ObjectTagSet) -> Result<()>; + /// Returns all shard hashes with their associated object tags. async fn list_shards_with_tags(&self) -> Result>; diff --git a/xet_client/src/cas_client/simulation/local_client.rs b/xet_client/src/cas_client/simulation/local_client.rs index 7b78e8236..6d4cfd1a2 100644 --- a/xet_client/src/cas_client/simulation/local_client.rs +++ b/xet_client/src/cas_client/simulation/local_client.rs @@ -31,7 +31,7 @@ use xet_runtime::core::XetContext; use xet_runtime::fd_diagnostics::{report_fd_count, track_fd_scope}; use xet_runtime::file_utils::SafeFileCreator; -use super::deletion_controls::ObjectTag; +use super::deletion_controls::{ObjectTag, ObjectTagSet}; use super::direct_access_client::DirectAccessClient; use super::xorb_utils::{self, REFERENCE_INSTANT, duration_to_expiration_secs_ceil}; use crate::cas_client::Client; @@ -375,6 +375,16 @@ impl LocalClient { PathBuf::from(name) } + /// Path of a xorb's S3-style tag set: `.tagset`, holding JSON. + /// A sidecar so writing tags cannot disturb the bytes the [`ObjectTag`] is + /// derived from. + fn tag_set_xorb_path(&self, hash: &MerkleHash) -> PathBuf { + let canonical = self.get_path_for_entry(hash); + let mut name = canonical.into_os_string(); + name.push(".tagset"); + PathBuf::from(name) + } + /// Path used to park a tagged-for-deletion shard: `.mdb.gctag`. fn gctag_shard_path(&self, hash: &MerkleHash) -> PathBuf { let canonical = self.shard_dir.join(shard_file_name(hash)); @@ -1053,6 +1063,33 @@ impl super::DeletionControlableClient for LocalClient { Ok(true) } + async fn get_xorb_tag_set(&self, hash: &MerkleHash) -> Result { + if !self.get_path_for_entry(hash).exists() { + return Err(ClientError::Other(format!("XORB not found: {}", hash.hex()))); + } + let path = self.tag_set_xorb_path(hash); + if !path.exists() { + return Ok(ObjectTagSet::new()); + } + let raw = std::fs::read(&path)?; + serde_json::from_slice(&raw) + .map_err(|e| ClientError::Other(format!("invalid tag set at {}: {e}", path.display()))) + } + + async fn set_xorb_tag_set(&self, hash: &MerkleHash, tags: ObjectTagSet) -> Result<()> { + if !self.get_path_for_entry(hash).exists() { + return Err(ClientError::Other(format!("XORB not found: {}", hash.hex()))); + } + let path = self.tag_set_xorb_path(hash); + let raw = serde_json::to_vec(&tags).map_err(|e| ClientError::Other(format!("serialize tag set: {e}")))?; + #[cfg(windows)] + if path.exists() { + Self::clear_readonly(&path); + } + std::fs::write(&path, raw)?; + Ok(()) + } + async fn list_shards_with_tags(&self) -> Result> { let mut ret = Vec::new(); for (hash, path) in self.shard_file_paths()? { @@ -2278,6 +2315,58 @@ mod tests { .expect("Integrity should pass: the old shard's stale file entry with dangling xorb refs is not consulted"); } + /// A tag set is stored, replaced wholesale on the next write (as S3 + /// `PutObjectTagging` does), and never disturbs the xorb's `ObjectTag` or + /// its appearance in the xorb listing — the sidecar must not read as a xorb. + #[tokio::test] + async fn test_xorb_tag_set_round_trip_leaves_object_tag_intact() { + let client = LocalClient::temporary(test_context()).await.unwrap(); + let file = client.upload_random_file(&[(1, (0, 2))], 2048).await.unwrap(); + let xorb_hash = file.terms[0].xorb_hash; + + let before = client.list_xorbs_and_tags().await.unwrap(); + let (_, tag_before) = before.iter().find(|(h, _)| *h == xorb_hash).unwrap(); + assert!(client.get_xorb_tag_set(&xorb_hash).await.unwrap().is_empty(), "a fresh xorb has no tag set"); + + client + .set_xorb_tag_set(&xorb_hash, vec![("last-upload".to_string(), "1234".to_string())]) + .await + .unwrap(); + assert_eq!( + client.get_xorb_tag_set(&xorb_hash).await.unwrap(), + vec![("last-upload".to_string(), "1234".to_string())] + ); + + // A second write replaces the set rather than merging into it. + client + .set_xorb_tag_set(&xorb_hash, vec![("other".to_string(), "x".to_string())]) + .await + .unwrap(); + assert_eq!( + client.get_xorb_tag_set(&xorb_hash).await.unwrap(), + vec![("other".to_string(), "x".to_string())], + "PutObjectTagging semantics replace the whole set" + ); + + // Both listing paths must ignore the sidecar, not just the tagged one. + let listed = client.list_xorbs().await.unwrap(); + assert_eq!(listed, vec![xorb_hash], "the .tagset sidecar must not be listed as a xorb"); + + let after = client.list_xorbs_and_tags().await.unwrap(); + assert_eq!(after.len(), before.len(), "the sidecar must not appear as an extra tagged xorb"); + let (_, tag_after) = after.iter().find(|(h, _)| *h == xorb_hash).unwrap(); + assert_eq!(tag_before, tag_after, "tagging must not move the ObjectTag"); + } + + /// Tagging an absent xorb is an error rather than creating an orphan sidecar. + #[tokio::test] + async fn test_xorb_tag_set_requires_the_xorb_to_exist() { + let client = LocalClient::temporary(test_context()).await.unwrap(); + let missing = MerkleHash::default(); + assert!(client.get_xorb_tag_set(&missing).await.is_err()); + assert!(client.set_xorb_tag_set(&missing, vec![]).await.is_err()); + } + /// Tests that list_xorbs_and_tags tags change after file re-creation with a timestamp delay. #[tokio::test] async fn test_list_xorbs_and_tags_timestamp_changes() { diff --git a/xet_client/src/cas_client/simulation/local_server/simulation_control_client.rs b/xet_client/src/cas_client/simulation/local_server/simulation_control_client.rs index 750cab89a..77835546c 100644 --- a/xet_client/src/cas_client/simulation/local_server/simulation_control_client.rs +++ b/xet_client/src/cas_client/simulation/local_server/simulation_control_client.rs @@ -11,12 +11,12 @@ use xet_runtime::core::XetContext; use super::simulation_types::{ FetchTermDataRequest, FetchTermDataResponse, FileShardsEntry, FileSizeResponse, HashWithTag, TagDeleteRequest, - TagDeleteResponse, XorbExistsResponse, XorbLengthResponse, XorbRangesRequest, XorbRangesResponse, + TagDeleteResponse, TagSetBody, XorbExistsResponse, XorbLengthResponse, XorbRangesRequest, XorbRangesResponse, XorbRawLengthResponse, }; use crate::cas_client::RemoteClient; use crate::cas_client::interface::Client; -use crate::cas_client::simulation::deletion_controls::ObjectTag; +use crate::cas_client::simulation::deletion_controls::{ObjectTag, ObjectTagSet}; use crate::cas_client::simulation::xorb_utils::duration_to_expiration_secs_ceil; use crate::cas_client::simulation::{DeletionControlableClient, DirectAccessClient}; use crate::cas_types::{ @@ -582,6 +582,32 @@ impl DeletionControlableClient for SimulationControlClient { Ok(result.deleted) } + async fn get_xorb_tag_set(&self, hash: &MerkleHash) -> Result { + let hex = HexMerkleHash::from(*hash); + let resp = self + .http_client + .get(self.sim_url(&format!("/xorbs/{hex}/tag_set"))) + .send() + .await + .map_err(|e| ClientError::Other(e.to_string()))?; + let resp = Self::check_status(resp).await?; + let body: TagSetBody = resp.json().await.map_err(|e| ClientError::Other(e.to_string()))?; + Ok(body.tags) + } + + async fn set_xorb_tag_set(&self, hash: &MerkleHash, tags: ObjectTagSet) -> Result<()> { + let hex = HexMerkleHash::from(*hash); + let resp = self + .http_client + .put(self.sim_url(&format!("/xorbs/{hex}/tag_set"))) + .json(&TagSetBody { tags }) + .send() + .await + .map_err(|e| ClientError::Other(e.to_string()))?; + Self::check_status(resp).await?; + Ok(()) + } + async fn list_shards_with_tags(&self) -> Result> { let resp = self .http_client diff --git a/xet_client/src/cas_client/simulation/local_server/simulation_handlers.rs b/xet_client/src/cas_client/simulation/local_server/simulation_handlers.rs index 6179805b8..8577daeb6 100644 --- a/xet_client/src/cas_client/simulation/local_server/simulation_handlers.rs +++ b/xet_client/src/cas_client/simulation/local_server/simulation_handlers.rs @@ -10,8 +10,8 @@ use http::header::RANGE; use super::handlers::{FileRangeVariant, ServerState, error_to_response, parse_range_header}; use super::simulation_types::{ ConfigDelayRangeRequest, ConfigDurationRequest, FetchTermDataRequest, FetchTermDataResponse, FileShardsEntry, - FileSizeResponse, HashWithTag, TagDeleteRequest, TagDeleteResponse, XorbExistsResponse, XorbLengthResponse, - XorbRangesRequest, XorbRangesResponse, XorbRawLengthResponse, + FileSizeResponse, HashWithTag, TagDeleteRequest, TagDeleteResponse, TagSetBody, XorbExistsResponse, + XorbLengthResponse, XorbRangesRequest, XorbRangesResponse, XorbRawLengthResponse, }; use crate::cas_types::{FileRange, HexMerkleHash}; @@ -38,6 +38,7 @@ pub fn simulation_routes() -> Router { // DeletionControlableClient routes .route("/xorbs_with_tags", get(list_xorbs_and_tags)) .route("/xorbs/{hash}/tag_delete", post(delete_xorb_if_tag_matches)) + .route("/xorbs/{hash}/tag_set", get(get_xorb_tag_set).put(set_xorb_tag_set)) .route("/shards", get(list_shard_entries)) .route("/shards_with_tags", get(list_shards_with_tags)) .route("/shards/{hash}", get(get_shard_bytes).delete(delete_shard_entry)) @@ -230,6 +231,33 @@ async fn delete_xorb_if_tag_matches( } } +async fn get_xorb_tag_set( + State(state): State, + Path(HexMerkleHash(hash)): Path, +) -> Response { + let Some(dc) = &state.deletion_client else { + return not_implemented(); + }; + match dc.get_xorb_tag_set(&hash).await { + Ok(tags) => Json(TagSetBody { tags }).into_response(), + Err(e) => error_to_response(e), + } +} + +async fn set_xorb_tag_set( + State(state): State, + Path(HexMerkleHash(hash)): Path, + Json(body): Json, +) -> Response { + let Some(dc) = &state.deletion_client else { + return not_implemented(); + }; + match dc.set_xorb_tag_set(&hash, body.tags).await { + Ok(()) => StatusCode::NO_CONTENT.into_response(), + Err(e) => error_to_response(e), + } +} + async fn list_shards_with_tags(State(state): State) -> Response { let Some(dc) = &state.deletion_client else { return not_implemented(); diff --git a/xet_client/src/cas_client/simulation/local_server/simulation_types.rs b/xet_client/src/cas_client/simulation/local_server/simulation_types.rs index 273bdff6e..9d455e2e4 100644 --- a/xet_client/src/cas_client/simulation/local_server/simulation_types.rs +++ b/xet_client/src/cas_client/simulation/local_server/simulation_types.rs @@ -1,7 +1,7 @@ use serde::{Deserialize, Serialize}; use xet_core_structures::merklehash::MerkleHash; -use crate::cas_client::simulation::deletion_controls::ObjectTag; +use crate::cas_client::simulation::deletion_controls::{ObjectTag, ObjectTagSet}; use crate::cas_types::XorbReconstructionFetchInfo; #[derive(Debug, Serialize, Deserialize)] @@ -78,3 +78,10 @@ pub struct TagDeleteRequest { pub struct TagDeleteResponse { pub deleted: bool, } + +/// Body of a tag-set read or write. Ordered pairs rather than a map so the +/// wire form matches S3's tag set, where key order is preserved. +#[derive(Debug, Serialize, Deserialize)] +pub struct TagSetBody { + pub tags: ObjectTagSet, +} diff --git a/xet_client/src/cas_client/simulation/memory_client.rs b/xet_client/src/cas_client/simulation/memory_client.rs index 39d264fb4..541daa08a 100644 --- a/xet_client/src/cas_client/simulation/memory_client.rs +++ b/xet_client/src/cas_client/simulation/memory_client.rs @@ -27,7 +27,7 @@ use super::super::interface::{ShardUploadProgressCallback, ShardUploadProgressTy use super::super::progress_tracked_streams::ProgressCallback; use super::client_testing_utils::{FileTermReference, RandomFileContents}; #[cfg(not(target_family = "wasm"))] -use super::deletion_controls::ObjectTag; +use super::deletion_controls::{ObjectTag, ObjectTagSet}; use super::direct_access_client::DirectAccessClient; use super::random_xorb::RandomXorb; use super::xorb_utils::{self, REFERENCE_INSTANT, duration_to_expiration_secs_ceil}; @@ -89,6 +89,9 @@ pub struct MemoryClient { /// Shard hash currently tagged for lifecycle deletion. Shard data is /// retained in `shard` so a re-upload can clear the tag. gc_tagged_shard: RwLock>, + /// S3-style tag sets per XORB. Held separately from `xorbs` so writing one + /// cannot perturb the bytes the [`ObjectTag`] is derived from. + xorb_tag_sets: RwLock>, } impl MemoryClient { @@ -108,6 +111,7 @@ impl MemoryClient { lifecycle_tag_deletion: AtomicBool::new(false), gc_tagged_xorbs: RwLock::new(HashSet::new()), gc_tagged_shard: RwLock::new(None), + xorb_tag_sets: RwLock::new(MerkleHashMap::new()), }) } @@ -1235,6 +1239,21 @@ impl super::DeletionControlableClient for MemoryClient { Ok(true) } + async fn get_xorb_tag_set(&self, hash: &MerkleHash) -> Result { + if !self.xorbs.read().await.contains_key(hash) { + return Err(ClientError::Other(format!("XORB not found: {}", hash.hex()))); + } + Ok(self.xorb_tag_sets.read().await.get(hash).cloned().unwrap_or_default()) + } + + async fn set_xorb_tag_set(&self, hash: &MerkleHash, tags: ObjectTagSet) -> Result<()> { + if !self.xorbs.read().await.contains_key(hash) { + return Err(ClientError::Other(format!("XORB not found: {}", hash.hex()))); + } + self.xorb_tag_sets.write().await.insert(*hash, tags); + Ok(()) + } + async fn list_shards_with_tags(&self) -> Result> { let shard = self.shard.read().await; let Some((shard_hash, shard_bytes)) = Self::current_shard_hash_and_bytes(&shard)? else { @@ -1618,6 +1637,47 @@ mod tests { assert!(client.xorbs.read().await.get(&xorb_hash).is_none()); } + /// Tag sets round-trip, replace wholesale on the next write, and leave the + /// xorb's `ObjectTag` alone — the same contract `LocalClient` provides. + #[tokio::test] + async fn test_xorb_tag_set_round_trip_leaves_object_tag_intact() { + let client = new_deletion_client(); + let file = client.upload_random_file(&[(1, (0, 2))], 2048).await.unwrap(); + let xorb_hash = file.terms[0].xorb_hash; + + let before = client.list_xorbs_and_tags().await.unwrap(); + assert!(client.get_xorb_tag_set(&xorb_hash).await.unwrap().is_empty()); + + client + .set_xorb_tag_set(&xorb_hash, vec![("last-upload".to_string(), "1234".to_string())]) + .await + .unwrap(); + assert_eq!( + client.get_xorb_tag_set(&xorb_hash).await.unwrap(), + vec![("last-upload".to_string(), "1234".to_string())] + ); + + client + .set_xorb_tag_set(&xorb_hash, vec![("other".to_string(), "x".to_string())]) + .await + .unwrap(); + assert_eq!( + client.get_xorb_tag_set(&xorb_hash).await.unwrap(), + vec![("other".to_string(), "x".to_string())], + "PutObjectTagging semantics replace the whole set" + ); + + assert_eq!(client.list_xorbs_and_tags().await.unwrap(), before, "tagging must not move the ObjectTag"); + } + + #[tokio::test] + async fn test_xorb_tag_set_requires_the_xorb_to_exist() { + let client = new_deletion_client(); + let missing = MerkleHash::default(); + assert!(client.get_xorb_tag_set(&missing).await.is_err()); + assert!(client.set_xorb_tag_set(&missing, vec![]).await.is_err()); + } + #[tokio::test] async fn test_lifecycle_tag_verify_integrity_flags_tagged_xorb() { let client = new_deletion_client(); diff --git a/xet_client/src/cas_client/simulation/mod.rs b/xet_client/src/cas_client/simulation/mod.rs index d39799555..80523b175 100644 --- a/xet_client/src/cas_client/simulation/mod.rs +++ b/xet_client/src/cas_client/simulation/mod.rs @@ -30,7 +30,7 @@ mod deletion_controls; mod local_client; #[cfg(not(target_family = "wasm"))] -pub use deletion_controls::{DeletionControlableClient, ObjectTag}; +pub use deletion_controls::{DeletionControlableClient, ObjectTag, ObjectTagSet}; #[cfg(not(target_family = "wasm"))] pub use local_client::LocalClient; From a79dbff2b3697e31db0321ebc242e90ef2448f7f Mon Sep 17 00:00:00 2001 From: Sam Horradarn Date: Fri, 4 Sep 2026 13:19:22 +0200 Subject: [PATCH 2/8] refactor(simulation): rename the conditional-deletion tag to etag MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit With `ObjectTagSet` alongside it, "tag" now meant three things in this module: the 32-byte compare-and-delete value, the S3-style key/value set, and the `gc-delete` lifecycle marker. The first is an ETag stand-in, so name it one: `ObjectTag` -> `ObjectETag`, `list_xorbs_and_tags` -> `list_xorbs_and_etags`, `delete_xorb_if_tag_matches` -> `delete_xorb_if_etag_matches` and the shard equivalents, `HashWithTag` -> `HashWithETag`, `TagDelete{Request,Response}` -> `ETagDelete{Request,Response}`, plus the local bindings and doc prose. Simulation routes move with them: `/xorbs_with_etags`, `/shards_with_etags`, `/{hash}/etag_delete`. Both ends of that wire are in this crate, so they change together. The lifecycle markers — `gc_tagged_*`, `lifecycle_tag_deletion`, `.gctag` — keep their names. They are the third concept and unrelated to either of the others. Breaking for `simulation`-feature consumers; xet-garbage-collection is the known one and updates with its lockfile bump. Co-Authored-By: Claude Opus 5 --- .../simulation/deletion_controls.rs | 29 ++++---- .../simulation/deletion_unit_testing.rs | 48 ++++++------- .../src/cas_client/simulation/local_client.rs | 50 ++++++------- .../local_server/simulation_control_client.rs | 40 +++++------ .../local_server/simulation_handlers.rs | 42 +++++------ .../local_server/simulation_types.rs | 12 ++-- .../cas_client/simulation/memory_client.rs | 72 +++++++++---------- xet_client/src/cas_client/simulation/mod.rs | 2 +- 8 files changed, 149 insertions(+), 146 deletions(-) diff --git a/xet_client/src/cas_client/simulation/deletion_controls.rs b/xet_client/src/cas_client/simulation/deletion_controls.rs index a965ae4c4..cfe0a6adc 100644 --- a/xet_client/src/cas_client/simulation/deletion_controls.rs +++ b/xet_client/src/cas_client/simulation/deletion_controls.rs @@ -4,16 +4,17 @@ use xet_core_structures::merklehash::MerkleHash; use crate::error::Result; -/// An opaque 32-byte tag used for conditional deletion (compare-and-delete). +/// An opaque 32-byte etag used for conditional deletion (compare-and-delete), standing in +/// for S3's ETag. /// /// Implementations should derive this from object metadata/content with enough entropy /// to reduce false matches when objects are rapidly rewritten. -pub type ObjectTag = [u8; 32]; +pub type ObjectETag = [u8; 32]; /// S3-style `(key, value)` tag set attached to an object. /// -/// Distinct from [`ObjectTag`]: writing this leaves the object's bytes, and so its -/// [`ObjectTag`], untouched. That is what lets a caller record something about an object +/// Distinct from [`ObjectETag`]: writing this leaves the object's bytes, and so its +/// [`ObjectETag`], untouched. That is what lets a caller record something about an object /// without invalidating anything keyed on its content. pub type ObjectTagSet = Vec<(String, String)>; @@ -48,12 +49,12 @@ pub trait DeletionControlableClient: Send + Sync { /// Deletes a XORB by hash. async fn delete_xorb(&self, hash: &MerkleHash); - /// Returns all XORB hashes with their associated object tags. - async fn list_xorbs_and_tags(&self) -> Result>; + /// Returns all XORB hashes with their associated object etags. + async fn list_xorbs_and_etags(&self) -> Result>; - /// Deletes a XORB only if its current tag matches the provided tag. - /// Returns `Ok(true)` if deleted, `Ok(false)` if the tag did not match. - async fn delete_xorb_if_tag_matches(&self, hash: &MerkleHash, tag: &ObjectTag) -> Result; + /// Deletes a XORB only if its current etag matches the provided etag. + /// Returns `Ok(true)` if deleted, `Ok(false)` if the etag did not match. + async fn delete_xorb_if_etag_matches(&self, hash: &MerkleHash, etag: &ObjectETag) -> Result; /// Returns a XORB's tag set, empty if it has none. async fn get_xorb_tag_set(&self, hash: &MerkleHash) -> Result; @@ -61,12 +62,12 @@ pub trait DeletionControlableClient: Send + Sync { /// Replaces a XORB's tag set wholesale, as S3 `PutObjectTagging` does. async fn set_xorb_tag_set(&self, hash: &MerkleHash, tags: ObjectTagSet) -> Result<()>; - /// Returns all shard hashes with their associated object tags. - async fn list_shards_with_tags(&self) -> Result>; + /// Returns all shard hashes with their associated object etags. + async fn list_shards_with_etags(&self) -> Result>; - /// Deletes a shard only if its current tag matches the provided tag. - /// Returns `Ok(true)` if deleted, `Ok(false)` if the tag did not match. - async fn delete_shard_if_tag_matches(&self, hash: &MerkleHash, tag: &ObjectTag) -> Result; + /// Deletes a shard only if its current etag matches the provided etag. + /// Returns `Ok(true)` if deleted, `Ok(false)` if the etag did not match. + async fn delete_shard_if_etag_matches(&self, hash: &MerkleHash, etag: &ObjectETag) -> Result; /// Verifies referential integrity of all shards on disk. async fn verify_integrity(&self) -> Result<()>; diff --git a/xet_client/src/cas_client/simulation/deletion_unit_testing.rs b/xet_client/src/cas_client/simulation/deletion_unit_testing.rs index 798f57a93..15f1e26cb 100644 --- a/xet_client/src/cas_client/simulation/deletion_unit_testing.rs +++ b/xet_client/src/cas_client/simulation/deletion_unit_testing.rs @@ -13,7 +13,7 @@ use std::sync::Arc; use xet_core_structures::merklehash::MerkleHash; use super::client_testing_utils::RandomFileContents; -use super::deletion_controls::ObjectTag; +use super::deletion_controls::ObjectETag; use super::{ClientTestingUtils, DeletionControlableClient, DirectAccessClient}; /// Runs all common DeletionControlableClient tests using a factory that creates fresh clients. @@ -340,19 +340,19 @@ async fn test_verify_integrity_after_file_deletion(client: Arc) { - assert!(client.list_xorbs_and_tags().await.unwrap().is_empty()); + assert!(client.list_xorbs_and_etags().await.unwrap().is_empty()); let file = client.upload_random_file(&[(1, (0, 3))], 2048).await.unwrap(); let expected_xorbs = expected_xorb_hashes(&[&file]); - let xorbs_and_tags = client.list_xorbs_and_tags().await.unwrap(); - let listed_hashes: HashSet = xorbs_and_tags.iter().map(|(h, _)| *h).collect(); + let xorbs_and_etags = client.list_xorbs_and_etags().await.unwrap(); + let listed_hashes: HashSet = xorbs_and_etags.iter().map(|(h, _)| *h).collect(); assert_eq!(listed_hashes, expected_xorbs); - let zero_tag: ObjectTag = [0u8; 32]; - for (_, tag) in &xorbs_and_tags { + let zero_tag: ObjectETag = [0u8; 32]; + for (_, tag) in &xorbs_and_etags { assert_ne!(tag, &zero_tag, "Tag should be non-zero"); } } @@ -362,32 +362,32 @@ async fn test_delete_xorb_if_tag_matches(client: Arc) { - assert!(client.list_shards_with_tags().await.unwrap().is_empty()); + assert!(client.list_shards_with_etags().await.unwrap().is_empty()); client.upload_random_file(&[(1, (0, 3))], 2048).await.unwrap(); - let shards_and_tags = client.list_shards_with_tags().await.unwrap(); + let shards_and_tags = client.list_shards_with_etags().await.unwrap(); let shard_entries = client.list_shard_entries().await.unwrap(); let listed_hashes: HashSet = shards_and_tags.iter().map(|(h, _)| *h).collect(); let expected_hashes: HashSet = shard_entries.into_iter().collect(); assert_eq!(listed_hashes, expected_hashes); - let zero_tag: ObjectTag = [0u8; 32]; + let zero_tag: ObjectETag = [0u8; 32]; for (_, tag) in &shards_and_tags { assert_ne!(tag, &zero_tag, "Tag should be non-zero"); } @@ -397,16 +397,16 @@ async fn test_list_shards_with_tags(client: Arc) { let _file = client.upload_random_file(&[(1, (0, 2))], 2048).await.unwrap(); - let shards_and_tags = client.list_shards_with_tags().await.unwrap(); + let shards_and_tags = client.list_shards_with_etags().await.unwrap(); assert!(!shards_and_tags.is_empty()); let (shard_hash, correct_tag) = &shards_and_tags[0]; - let wrong_tag: ObjectTag = [0xFFu8; 32]; - let deleted = client.delete_shard_if_tag_matches(shard_hash, &wrong_tag).await.unwrap(); - assert!(!deleted, "Wrong tag should not delete the shard"); - assert!(!client.list_shard_entries().await.unwrap().is_empty(), "Shard should still exist after wrong tag"); + let wrong_tag: ObjectETag = [0xFFu8; 32]; + let deleted = client.delete_shard_if_etag_matches(shard_hash, &wrong_tag).await.unwrap(); + assert!(!deleted, "Wrong etag should not delete the shard"); + assert!(!client.list_shard_entries().await.unwrap().is_empty(), "Shard should still exist after wrong etag"); - let deleted = client.delete_shard_if_tag_matches(shard_hash, correct_tag).await.unwrap(); - assert!(deleted, "Correct tag should delete the shard"); - assert!(client.list_shard_entries().await.unwrap().is_empty(), "Shard should be gone after correct tag"); + let deleted = client.delete_shard_if_etag_matches(shard_hash, correct_tag).await.unwrap(); + assert!(deleted, "Correct etag should delete the shard"); + assert!(client.list_shard_entries().await.unwrap().is_empty(), "Shard should be gone after correct etag"); } diff --git a/xet_client/src/cas_client/simulation/local_client.rs b/xet_client/src/cas_client/simulation/local_client.rs index 6d4cfd1a2..319890c97 100644 --- a/xet_client/src/cas_client/simulation/local_client.rs +++ b/xet_client/src/cas_client/simulation/local_client.rs @@ -31,7 +31,7 @@ use xet_runtime::core::XetContext; use xet_runtime::fd_diagnostics::{report_fd_count, track_fd_scope}; use xet_runtime::file_utils::SafeFileCreator; -use super::deletion_controls::{ObjectTag, ObjectTagSet}; +use super::deletion_controls::{ObjectETag, ObjectTagSet}; use super::direct_access_client::DirectAccessClient; use super::xorb_utils::{self, REFERENCE_INSTANT, duration_to_expiration_secs_ceil}; use crate::cas_client::Client; @@ -376,7 +376,7 @@ impl LocalClient { } /// Path of a xorb's S3-style tag set: `.tagset`, holding JSON. - /// A sidecar so writing tags cannot disturb the bytes the [`ObjectTag`] is + /// A sidecar so writing tags cannot disturb the bytes the [`ObjectETag`] is /// derived from. fn tag_set_xorb_path(&self, hash: &MerkleHash) -> PathBuf { let canonical = self.get_path_for_entry(hash); @@ -432,11 +432,11 @@ impl LocalClient { } } - /// Builds an `ObjectTag` from file metadata at the given path. + /// Builds an `ObjectETag` from file metadata at the given path. /// /// We hash multiple metadata fields to increase entropy and reduce false /// matches during rapid rewrite/delete races. - fn object_tag_from_path(path: &Path) -> Result { + fn object_etag_from_path(path: &Path) -> Result { let meta = std::fs::metadata(path).map_err(ClientError::internal)?; let modified = meta.modified().map_err(ClientError::internal)?; let modified_nanos = modified.duration_since(std::time::UNIX_EPOCH).unwrap_or_default().as_nanos(); @@ -1006,7 +1006,7 @@ impl super::DeletionControlableClient for LocalClient { } } - async fn list_xorbs_and_tags(&self) -> Result> { + async fn list_xorbs_and_etags(&self) -> Result> { let mut ret = Vec::new(); for entry in self.xorb_dir.read_dir().map_err(ClientError::internal)? { let entry = entry.map_err(ClientError::internal)?; @@ -1017,15 +1017,15 @@ impl super::DeletionControlableClient for LocalClient { if let Some(pos) = name.rfind('.') { let hex = &name[(pos + 1)..]; if let Ok(hash) = MerkleHash::from_hex(hex) { - let tag = Self::object_tag_from_path(&path)?; - ret.push((hash, tag)); + let etag = Self::object_etag_from_path(&path)?; + ret.push((hash, etag)); } } } Ok(ret) } - async fn delete_xorb_if_tag_matches(&self, hash: &MerkleHash, tag: &ObjectTag) -> Result { + async fn delete_xorb_if_etag_matches(&self, hash: &MerkleHash, etag: &ObjectETag) -> Result { let file_path = self.get_path_for_entry(hash); // Atomically move the file out of the namespace before checking the @@ -1036,7 +1036,7 @@ impl super::DeletionControlableClient for LocalClient { return Err(ClientError::XORBNotFound(*hash)); } - let current_tag = match Self::object_tag_from_path(&tmp_path) { + let current_etag = match Self::object_etag_from_path(&tmp_path) { Ok(t) => t, Err(e) => { Self::restore_from_tmp(&tmp_path, &file_path); @@ -1044,7 +1044,7 @@ impl super::DeletionControlableClient for LocalClient { }, }; - if ¤t_tag != tag { + if ¤t_etag != etag { Self::restore_from_tmp(&tmp_path, &file_path); return Ok(false); } @@ -1090,16 +1090,16 @@ impl super::DeletionControlableClient for LocalClient { Ok(()) } - async fn list_shards_with_tags(&self) -> Result> { + async fn list_shards_with_etags(&self) -> Result> { let mut ret = Vec::new(); for (hash, path) in self.shard_file_paths()? { - let tag = Self::object_tag_from_path(&path)?; - ret.push((hash, tag)); + let etag = Self::object_etag_from_path(&path)?; + ret.push((hash, etag)); } Ok(ret) } - async fn delete_shard_if_tag_matches(&self, hash: &MerkleHash, tag: &ObjectTag) -> Result { + async fn delete_shard_if_etag_matches(&self, hash: &MerkleHash, etag: &ObjectETag) -> Result { let path = self.shard_path_for_hash(hash)?; let tmp_path = path.with_extension(format!("gc_del_{:x}", rand::random::())); @@ -1107,7 +1107,7 @@ impl super::DeletionControlableClient for LocalClient { return Err(ClientError::Other(format!("Shard not found: {}", hash.hex()))); } - let current_tag = match Self::object_tag_from_path(&tmp_path) { + let current_etag = match Self::object_etag_from_path(&tmp_path) { Ok(t) => t, Err(e) => { Self::restore_from_tmp(&tmp_path, &path); @@ -1115,7 +1115,7 @@ impl super::DeletionControlableClient for LocalClient { }, }; - if ¤t_tag != tag { + if ¤t_etag != etag { Self::restore_from_tmp(&tmp_path, &path); return Ok(false); } @@ -1643,7 +1643,7 @@ impl Client for LocalClient { // Always rewrite: even if the xorb already exists, the file must be // re-created so its filesystem metadata (mtime/ctime) changes, producing - // a new tag for delete_xorb_if_tag_matches. SafeFileCreator uses + // a new etag for delete_xorb_if_etag_matches. SafeFileCreator uses // temp-file + atomic rename, so concurrent readers are safe. // Reconstruct footer if not present @@ -2316,7 +2316,7 @@ mod tests { } /// A tag set is stored, replaced wholesale on the next write (as S3 - /// `PutObjectTagging` does), and never disturbs the xorb's `ObjectTag` or + /// `PutObjectTagging` does), and never disturbs the xorb's `ObjectETag` or /// its appearance in the xorb listing — the sidecar must not read as a xorb. #[tokio::test] async fn test_xorb_tag_set_round_trip_leaves_object_tag_intact() { @@ -2324,7 +2324,7 @@ mod tests { let file = client.upload_random_file(&[(1, (0, 2))], 2048).await.unwrap(); let xorb_hash = file.terms[0].xorb_hash; - let before = client.list_xorbs_and_tags().await.unwrap(); + let before = client.list_xorbs_and_etags().await.unwrap(); let (_, tag_before) = before.iter().find(|(h, _)| *h == xorb_hash).unwrap(); assert!(client.get_xorb_tag_set(&xorb_hash).await.unwrap().is_empty(), "a fresh xorb has no tag set"); @@ -2352,10 +2352,10 @@ mod tests { let listed = client.list_xorbs().await.unwrap(); assert_eq!(listed, vec![xorb_hash], "the .tagset sidecar must not be listed as a xorb"); - let after = client.list_xorbs_and_tags().await.unwrap(); + let after = client.list_xorbs_and_etags().await.unwrap(); assert_eq!(after.len(), before.len(), "the sidecar must not appear as an extra tagged xorb"); let (_, tag_after) = after.iter().find(|(h, _)| *h == xorb_hash).unwrap(); - assert_eq!(tag_before, tag_after, "tagging must not move the ObjectTag"); + assert_eq!(tag_before, tag_after, "tagging must not move the ObjectETag"); } /// Tagging an absent xorb is an error rather than creating an orphan sidecar. @@ -2367,15 +2367,15 @@ mod tests { assert!(client.set_xorb_tag_set(&missing, vec![]).await.is_err()); } - /// Tests that list_xorbs_and_tags tags change after file re-creation with a timestamp delay. + /// Tests that list_xorbs_and_etags etags change after file re-creation with a timestamp delay. #[tokio::test] - async fn test_list_xorbs_and_tags_timestamp_changes() { + async fn test_list_xorbs_and_etags_timestamp_changes() { let client = LocalClient::temporary(test_context()).await.unwrap(); let file1 = client.upload_random_file(&[(1, (0, 2))], 2048).await.unwrap(); let xorb_hash = file1.terms[0].xorb_hash; - let tags1 = client.list_xorbs_and_tags().await.unwrap(); + let tags1 = client.list_xorbs_and_etags().await.unwrap(); let (_, tag1) = tags1.iter().find(|(h, _)| *h == xorb_hash).unwrap(); // Delete and wait 1 second so the filesystem timestamp advances. @@ -2386,7 +2386,7 @@ mod tests { let file2 = client.upload_random_file(&[(1, (0, 2))], 2048).await.unwrap(); let xorb_hash2 = file2.terms[0].xorb_hash; - let tags2 = client.list_xorbs_and_tags().await.unwrap(); + let tags2 = client.list_xorbs_and_etags().await.unwrap(); let (_, tag2) = tags2.iter().find(|(h, _)| *h == xorb_hash2).unwrap(); assert_ne!(tag1, tag2, "Tags should differ after re-creation with timestamp delay"); diff --git a/xet_client/src/cas_client/simulation/local_server/simulation_control_client.rs b/xet_client/src/cas_client/simulation/local_server/simulation_control_client.rs index 77835546c..02a6fceb4 100644 --- a/xet_client/src/cas_client/simulation/local_server/simulation_control_client.rs +++ b/xet_client/src/cas_client/simulation/local_server/simulation_control_client.rs @@ -10,13 +10,13 @@ use xet_core_structures::xorb_object::XorbObject; use xet_runtime::core::XetContext; use super::simulation_types::{ - FetchTermDataRequest, FetchTermDataResponse, FileShardsEntry, FileSizeResponse, HashWithTag, TagDeleteRequest, - TagDeleteResponse, TagSetBody, XorbExistsResponse, XorbLengthResponse, XorbRangesRequest, XorbRangesResponse, - XorbRawLengthResponse, + ETagDeleteRequest, ETagDeleteResponse, FetchTermDataRequest, FetchTermDataResponse, FileShardsEntry, + FileSizeResponse, HashWithETag, TagSetBody, XorbExistsResponse, XorbLengthResponse, XorbRangesRequest, + XorbRangesResponse, XorbRawLengthResponse, }; use crate::cas_client::RemoteClient; use crate::cas_client::interface::Client; -use crate::cas_client::simulation::deletion_controls::{ObjectTag, ObjectTagSet}; +use crate::cas_client::simulation::deletion_controls::{ObjectETag, ObjectTagSet}; use crate::cas_client::simulation::xorb_utils::duration_to_expiration_secs_ceil; use crate::cas_client::simulation::{DeletionControlableClient, DirectAccessClient}; use crate::cas_types::{ @@ -554,22 +554,22 @@ impl DeletionControlableClient for SimulationControlClient { let _ = self.http_client.delete(&url).send().await; } - async fn list_xorbs_and_tags(&self) -> Result> { + async fn list_xorbs_and_etags(&self) -> Result> { let resp = self .http_client - .get(self.sim_url("/xorbs_with_tags")) + .get(self.sim_url("/xorbs_with_etags")) .send() .await .map_err(|e| ClientError::Other(e.to_string()))?; let resp = Self::check_status(resp).await?; - let entries: Vec = resp.json().await.map_err(|e| ClientError::Other(e.to_string()))?; - Ok(entries.into_iter().map(|e| (e.hash, e.tag)).collect()) + let entries: Vec = resp.json().await.map_err(|e| ClientError::Other(e.to_string()))?; + Ok(entries.into_iter().map(|e| (e.hash, e.etag)).collect()) } - async fn delete_xorb_if_tag_matches(&self, hash: &MerkleHash, tag: &ObjectTag) -> Result { + async fn delete_xorb_if_etag_matches(&self, hash: &MerkleHash, etag: &ObjectETag) -> Result { let hex = HexMerkleHash::from(*hash); - let url = self.sim_url(&format!("/xorbs/{hex}/tag_delete")); - let body = TagDeleteRequest { tag: *tag }; + let url = self.sim_url(&format!("/xorbs/{hex}/etag_delete")); + let body = ETagDeleteRequest { etag: *etag }; let resp = self .http_client .post(&url) @@ -578,7 +578,7 @@ impl DeletionControlableClient for SimulationControlClient { .await .map_err(|e| ClientError::Other(e.to_string()))?; let resp = Self::check_status(resp).await?; - let result: TagDeleteResponse = resp.json().await.map_err(|e| ClientError::Other(e.to_string()))?; + let result: ETagDeleteResponse = resp.json().await.map_err(|e| ClientError::Other(e.to_string()))?; Ok(result.deleted) } @@ -608,22 +608,22 @@ impl DeletionControlableClient for SimulationControlClient { Ok(()) } - async fn list_shards_with_tags(&self) -> Result> { + async fn list_shards_with_etags(&self) -> Result> { let resp = self .http_client - .get(self.sim_url("/shards_with_tags")) + .get(self.sim_url("/shards_with_etags")) .send() .await .map_err(|e| ClientError::Other(e.to_string()))?; let resp = Self::check_status(resp).await?; - let entries: Vec = resp.json().await.map_err(|e| ClientError::Other(e.to_string()))?; - Ok(entries.into_iter().map(|e| (e.hash, e.tag)).collect()) + let entries: Vec = resp.json().await.map_err(|e| ClientError::Other(e.to_string()))?; + Ok(entries.into_iter().map(|e| (e.hash, e.etag)).collect()) } - async fn delete_shard_if_tag_matches(&self, hash: &MerkleHash, tag: &ObjectTag) -> Result { + async fn delete_shard_if_etag_matches(&self, hash: &MerkleHash, etag: &ObjectETag) -> Result { let hex = HexMerkleHash::from(*hash); - let url = self.sim_url(&format!("/shards/{hex}/tag_delete")); - let body = TagDeleteRequest { tag: *tag }; + let url = self.sim_url(&format!("/shards/{hex}/etag_delete")); + let body = ETagDeleteRequest { etag: *etag }; let resp = self .http_client .post(&url) @@ -632,7 +632,7 @@ impl DeletionControlableClient for SimulationControlClient { .await .map_err(|e| ClientError::Other(e.to_string()))?; let resp = Self::check_status(resp).await?; - let result: TagDeleteResponse = resp.json().await.map_err(|e| ClientError::Other(e.to_string()))?; + let result: ETagDeleteResponse = resp.json().await.map_err(|e| ClientError::Other(e.to_string()))?; Ok(result.deleted) } diff --git a/xet_client/src/cas_client/simulation/local_server/simulation_handlers.rs b/xet_client/src/cas_client/simulation/local_server/simulation_handlers.rs index 8577daeb6..abc3c4916 100644 --- a/xet_client/src/cas_client/simulation/local_server/simulation_handlers.rs +++ b/xet_client/src/cas_client/simulation/local_server/simulation_handlers.rs @@ -9,8 +9,8 @@ use http::header::RANGE; use super::handlers::{FileRangeVariant, ServerState, error_to_response, parse_range_header}; use super::simulation_types::{ - ConfigDelayRangeRequest, ConfigDurationRequest, FetchTermDataRequest, FetchTermDataResponse, FileShardsEntry, - FileSizeResponse, HashWithTag, TagDeleteRequest, TagDeleteResponse, TagSetBody, XorbExistsResponse, + ConfigDelayRangeRequest, ConfigDurationRequest, ETagDeleteRequest, ETagDeleteResponse, FetchTermDataRequest, + FetchTermDataResponse, FileShardsEntry, FileSizeResponse, HashWithETag, TagSetBody, XorbExistsResponse, XorbLengthResponse, XorbRangesRequest, XorbRangesResponse, XorbRawLengthResponse, }; use crate::cas_types::{FileRange, HexMerkleHash}; @@ -36,13 +36,13 @@ pub fn simulation_routes() -> Router { .route("/config/api_delay", post(set_api_delay)) .route("/fetch_term_data", post(fetch_term_data)) // DeletionControlableClient routes - .route("/xorbs_with_tags", get(list_xorbs_and_tags)) - .route("/xorbs/{hash}/tag_delete", post(delete_xorb_if_tag_matches)) + .route("/xorbs_with_etags", get(list_xorbs_and_etags)) + .route("/xorbs/{hash}/etag_delete", post(delete_xorb_if_etag_matches)) .route("/xorbs/{hash}/tag_set", get(get_xorb_tag_set).put(set_xorb_tag_set)) .route("/shards", get(list_shard_entries)) - .route("/shards_with_tags", get(list_shards_with_tags)) + .route("/shards_with_etags", get(list_shards_with_etags)) .route("/shards/{hash}", get(get_shard_bytes).delete(delete_shard_entry)) - .route("/shards/{hash}/tag_delete", post(delete_shard_if_tag_matches)) + .route("/shards/{hash}/etag_delete", post(delete_shard_if_etag_matches)) .route("/shards/{hash}/dedup_entries", delete(remove_shard_dedup_entries)) .route("/file_entries", get(list_file_shard_entries)) .route("/file_entries/{hash}", delete(delete_file_entry)) @@ -204,29 +204,30 @@ async fn delete_xorb(State(state): State, Path(HexMerkleHash(hash)) StatusCode::NO_CONTENT.into_response() } -async fn list_xorbs_and_tags(State(state): State) -> Response { +async fn list_xorbs_and_etags(State(state): State) -> Response { let Some(dc) = &state.deletion_client else { return not_implemented(); }; - match dc.list_xorbs_and_tags().await { + match dc.list_xorbs_and_etags().await { Ok(entries) => { - let response: Vec = entries.into_iter().map(|(hash, tag)| HashWithTag { hash, tag }).collect(); + let response: Vec = + entries.into_iter().map(|(hash, etag)| HashWithETag { hash, etag }).collect(); Json(response).into_response() }, Err(e) => error_to_response(e), } } -async fn delete_xorb_if_tag_matches( +async fn delete_xorb_if_etag_matches( State(state): State, Path(HexMerkleHash(hash)): Path, - Json(body): Json, + Json(body): Json, ) -> Response { let Some(dc) = &state.deletion_client else { return not_implemented(); }; - match dc.delete_xorb_if_tag_matches(&hash, &body.tag).await { - Ok(deleted) => Json(TagDeleteResponse { deleted }).into_response(), + match dc.delete_xorb_if_etag_matches(&hash, &body.etag).await { + Ok(deleted) => Json(ETagDeleteResponse { deleted }).into_response(), Err(e) => error_to_response(e), } } @@ -258,29 +259,30 @@ async fn set_xorb_tag_set( } } -async fn list_shards_with_tags(State(state): State) -> Response { +async fn list_shards_with_etags(State(state): State) -> Response { let Some(dc) = &state.deletion_client else { return not_implemented(); }; - match dc.list_shards_with_tags().await { + match dc.list_shards_with_etags().await { Ok(entries) => { - let response: Vec = entries.into_iter().map(|(hash, tag)| HashWithTag { hash, tag }).collect(); + let response: Vec = + entries.into_iter().map(|(hash, etag)| HashWithETag { hash, etag }).collect(); Json(response).into_response() }, Err(e) => error_to_response(e), } } -async fn delete_shard_if_tag_matches( +async fn delete_shard_if_etag_matches( State(state): State, Path(HexMerkleHash(hash)): Path, - Json(body): Json, + Json(body): Json, ) -> Response { let Some(dc) = &state.deletion_client else { return not_implemented(); }; - match dc.delete_shard_if_tag_matches(&hash, &body.tag).await { - Ok(deleted) => Json(TagDeleteResponse { deleted }).into_response(), + match dc.delete_shard_if_etag_matches(&hash, &body.etag).await { + Ok(deleted) => Json(ETagDeleteResponse { deleted }).into_response(), Err(e) => error_to_response(e), } } diff --git a/xet_client/src/cas_client/simulation/local_server/simulation_types.rs b/xet_client/src/cas_client/simulation/local_server/simulation_types.rs index 9d455e2e4..4a43bacf9 100644 --- a/xet_client/src/cas_client/simulation/local_server/simulation_types.rs +++ b/xet_client/src/cas_client/simulation/local_server/simulation_types.rs @@ -1,7 +1,7 @@ use serde::{Deserialize, Serialize}; use xet_core_structures::merklehash::MerkleHash; -use crate::cas_client::simulation::deletion_controls::{ObjectTag, ObjectTagSet}; +use crate::cas_client::simulation::deletion_controls::{ObjectETag, ObjectTagSet}; use crate::cas_types::XorbReconstructionFetchInfo; #[derive(Debug, Serialize, Deserialize)] @@ -64,18 +64,18 @@ pub struct FetchTermDataResponse { } #[derive(Debug, Serialize, Deserialize)] -pub struct HashWithTag { +pub struct HashWithETag { pub hash: MerkleHash, - pub tag: ObjectTag, + pub etag: ObjectETag, } #[derive(Debug, Serialize, Deserialize)] -pub struct TagDeleteRequest { - pub tag: ObjectTag, +pub struct ETagDeleteRequest { + pub etag: ObjectETag, } #[derive(Debug, Serialize, Deserialize)] -pub struct TagDeleteResponse { +pub struct ETagDeleteResponse { pub deleted: bool, } diff --git a/xet_client/src/cas_client/simulation/memory_client.rs b/xet_client/src/cas_client/simulation/memory_client.rs index 541daa08a..9c2e43924 100644 --- a/xet_client/src/cas_client/simulation/memory_client.rs +++ b/xet_client/src/cas_client/simulation/memory_client.rs @@ -27,7 +27,7 @@ use super::super::interface::{ShardUploadProgressCallback, ShardUploadProgressTy use super::super::progress_tracked_streams::ProgressCallback; use super::client_testing_utils::{FileTermReference, RandomFileContents}; #[cfg(not(target_family = "wasm"))] -use super::deletion_controls::{ObjectTag, ObjectTagSet}; +use super::deletion_controls::{ObjectETag, ObjectTagSet}; use super::direct_access_client::DirectAccessClient; use super::random_xorb::RandomXorb; use super::xorb_utils::{self, REFERENCE_INSTANT, duration_to_expiration_secs_ceil}; @@ -47,7 +47,7 @@ struct MaterializedXorb { /// Storage for a XORB - either fully materialized or generated on-the-fly. /// Each variant carries a monotonic `generation` counter that is bumped on -/// every insert, so the tag changes even when content is identical (matching +/// every insert, so the etag changes even when content is identical (matching /// production ETag semantics). enum XorbStorage { Materialized { entry: MaterializedXorb, generation: u64 }, @@ -64,7 +64,7 @@ pub struct MemoryClient { global_dedup: RwLock>, /// Upload concurrency controller upload_concurrency_controller: Arc, - /// Monotonic counter for xorb upload generations (tag freshness). + /// Monotonic counter for xorb upload generations (etag freshness). xorb_generation: AtomicU64, /// URL expiration in milliseconds url_expiration_ms: AtomicU64, @@ -90,7 +90,7 @@ pub struct MemoryClient { /// retained in `shard` so a re-upload can clear the tag. gc_tagged_shard: RwLock>, /// S3-style tag sets per XORB. Held separately from `xorbs` so writing one - /// cannot perturb the bytes the [`ObjectTag`] is derived from. + /// cannot perturb the bytes the [`ObjectETag`] is derived from. xorb_tag_sets: RwLock>, } @@ -275,7 +275,7 @@ impl MemoryClient { } #[cfg(not(target_family = "wasm"))] - fn object_tag_from_key_and_payload(prefix: &[u8], key: &MerkleHash, payload: &[u8]) -> ObjectTag { + fn object_etag_from_key_and_payload(prefix: &[u8], key: &MerkleHash, payload: &[u8]) -> ObjectETag { let key_bytes: [u8; 32] = (*key).into(); let payload_hash: [u8; 32] = compute_data_hash(payload).into(); let mut entropy = Vec::with_capacity(prefix.len() + key_bytes.len() + payload_hash.len()); @@ -286,18 +286,18 @@ impl MemoryClient { } #[cfg(not(target_family = "wasm"))] - fn xorb_tag(hash: &MerkleHash, storage: &XorbStorage) -> ObjectTag { + fn xorb_etag(hash: &MerkleHash, storage: &XorbStorage) -> ObjectETag { match storage { XorbStorage::Materialized { entry, generation } => { let mut payload = Vec::from(entry.serialized_data.as_ref()); payload.extend_from_slice(&generation.to_le_bytes()); - Self::object_tag_from_key_and_payload(b"xorb", hash, &payload) + Self::object_etag_from_key_and_payload(b"xorb", hash, &payload) }, XorbStorage::Random { xorb, generation } => { let mut entropy = Vec::with_capacity(16); entropy.extend_from_slice(&xorb.num_chunks().to_le_bytes()); entropy.extend_from_slice(&generation.to_le_bytes()); - Self::object_tag_from_key_and_payload(b"xorb", hash, &entropy) + Self::object_etag_from_key_and_payload(b"xorb", hash, &entropy) }, } } @@ -913,8 +913,8 @@ impl Client for MemoryClient { let serialized_data = serialized_xorb_object.serialized_data; // Always overwrite: even if the xorb already exists, we must store it - // with a fresh generation so its tag changes, matching production ETag - // semantics and ensuring delete_xorb_if_tag_matches is safe under + // with a fresh generation so its etag changes, matching production ETag + // semantics and ensuring delete_xorb_if_etag_matches is safe under // concurrent uploads. info!("Storing XORB {hash:?} in memory"); @@ -1210,25 +1210,25 @@ impl super::DeletionControlableClient for MemoryClient { } } - async fn list_xorbs_and_tags(&self) -> Result> { + async fn list_xorbs_and_etags(&self) -> Result> { let tagged = self.gc_tagged_xorbs.read().await; let xorbs = self.xorbs.read().await; Ok(xorbs .iter() .filter(|(hash, _)| !tagged.contains(hash)) - .map(|(hash, storage)| (*hash, Self::xorb_tag(hash, storage))) + .map(|(hash, storage)| (*hash, Self::xorb_etag(hash, storage))) .collect()) } - async fn delete_xorb_if_tag_matches(&self, hash: &MerkleHash, tag: &ObjectTag) -> Result { - let current_tag = { + async fn delete_xorb_if_etag_matches(&self, hash: &MerkleHash, etag: &ObjectETag) -> Result { + let current_etag = { let xorbs = self.xorbs.read().await; let Some(storage) = xorbs.get(hash) else { return Err(ClientError::XORBNotFound(*hash)); }; - Self::xorb_tag(hash, storage) + Self::xorb_etag(hash, storage) }; - if ¤t_tag != tag { + if ¤t_etag != etag { return Ok(false); } if self.lifecycle_tag_deletion_enabled() { @@ -1254,7 +1254,7 @@ impl super::DeletionControlableClient for MemoryClient { Ok(()) } - async fn list_shards_with_tags(&self) -> Result> { + async fn list_shards_with_etags(&self) -> Result> { let shard = self.shard.read().await; let Some((shard_hash, shard_bytes)) = Self::current_shard_hash_and_bytes(&shard)? else { return Ok(Vec::new()); @@ -1262,12 +1262,12 @@ impl super::DeletionControlableClient for MemoryClient { if self.shard_is_tagged(&shard_hash).await { return Ok(Vec::new()); } - let tag = Self::object_tag_from_key_and_payload(b"shard", &shard_hash, shard_bytes.as_ref()); - Ok(vec![(shard_hash, tag)]) + let etag = Self::object_etag_from_key_and_payload(b"shard", &shard_hash, shard_bytes.as_ref()); + Ok(vec![(shard_hash, etag)]) } - async fn delete_shard_if_tag_matches(&self, hash: &MerkleHash, tag: &ObjectTag) -> Result { - let (current_hash, current_tag) = { + async fn delete_shard_if_etag_matches(&self, hash: &MerkleHash, etag: &ObjectETag) -> Result { + let (current_hash, current_etag) = { let shard = self.shard.read().await; let Some((current_hash, shard_bytes)) = Self::current_shard_hash_and_bytes(&shard)? else { return Err(ClientError::Other(format!("Shard not found: {}", hash.hex()))); @@ -1275,10 +1275,10 @@ impl super::DeletionControlableClient for MemoryClient { if ¤t_hash != hash { return Err(ClientError::Other(format!("Shard not found: {}", hash.hex()))); } - let current_tag = Self::object_tag_from_key_and_payload(b"shard", ¤t_hash, shard_bytes.as_ref()); - (current_hash, current_tag) + let current_etag = Self::object_etag_from_key_and_payload(b"shard", ¤t_hash, shard_bytes.as_ref()); + (current_hash, current_etag) }; - if ¤t_tag != tag { + if ¤t_etag != etag { return Ok(false); } if self.lifecycle_tag_deletion_enabled() { @@ -1388,26 +1388,26 @@ mod tests { let client = new_deletion_client(); let file = client.upload_random_file(&[(1, (0, 3))], 2048).await.unwrap(); - let xorbs_and_tags = client.list_xorbs_and_tags().await.unwrap(); - assert!(!xorbs_and_tags.is_empty()); - let (xorb_hash, tag) = xorbs_and_tags[0]; + let xorbs_and_etags = client.list_xorbs_and_etags().await.unwrap(); + assert!(!xorbs_and_etags.is_empty()); + let (xorb_hash, tag) = xorbs_and_etags[0]; let wrong_tag = [0xABu8; 32]; - assert!(!client.delete_xorb_if_tag_matches(&xorb_hash, &wrong_tag).await.unwrap()); + assert!(!client.delete_xorb_if_etag_matches(&xorb_hash, &wrong_tag).await.unwrap()); assert!(client.xorb_exists(&xorb_hash).await.unwrap()); - assert!(client.delete_xorb_if_tag_matches(&xorb_hash, &tag).await.unwrap()); + assert!(client.delete_xorb_if_etag_matches(&xorb_hash, &tag).await.unwrap()); assert!(!client.xorb_exists(&xorb_hash).await.unwrap()); // file deletion is idempotent for parity with the disk-backed behavior. client.delete_file_entry(&file.file_hash).await.unwrap(); client.delete_file_entry(&file.file_hash).await.unwrap(); - let shards_and_tags = client.list_shards_with_tags().await.unwrap(); + let shards_and_tags = client.list_shards_with_etags().await.unwrap(); if !shards_and_tags.is_empty() { - let (shard_hash, shard_tag) = shards_and_tags[0]; - assert!(!client.delete_shard_if_tag_matches(&shard_hash, &wrong_tag).await.unwrap()); - assert!(client.delete_shard_if_tag_matches(&shard_hash, &shard_tag).await.unwrap()); + let (shard_hash, shard_etag) = shards_and_tags[0]; + assert!(!client.delete_shard_if_etag_matches(&shard_hash, &wrong_tag).await.unwrap()); + assert!(client.delete_shard_if_etag_matches(&shard_hash, &shard_etag).await.unwrap()); assert!(client.list_shard_entries().await.unwrap().is_empty()); } } @@ -1638,14 +1638,14 @@ mod tests { } /// Tag sets round-trip, replace wholesale on the next write, and leave the - /// xorb's `ObjectTag` alone — the same contract `LocalClient` provides. + /// xorb's `ObjectETag` alone — the same contract `LocalClient` provides. #[tokio::test] async fn test_xorb_tag_set_round_trip_leaves_object_tag_intact() { let client = new_deletion_client(); let file = client.upload_random_file(&[(1, (0, 2))], 2048).await.unwrap(); let xorb_hash = file.terms[0].xorb_hash; - let before = client.list_xorbs_and_tags().await.unwrap(); + let before = client.list_xorbs_and_etags().await.unwrap(); assert!(client.get_xorb_tag_set(&xorb_hash).await.unwrap().is_empty()); client @@ -1667,7 +1667,7 @@ mod tests { "PutObjectTagging semantics replace the whole set" ); - assert_eq!(client.list_xorbs_and_tags().await.unwrap(), before, "tagging must not move the ObjectTag"); + assert_eq!(client.list_xorbs_and_etags().await.unwrap(), before, "tagging must not move the ObjectETag"); } #[tokio::test] diff --git a/xet_client/src/cas_client/simulation/mod.rs b/xet_client/src/cas_client/simulation/mod.rs index 80523b175..900f6a2b7 100644 --- a/xet_client/src/cas_client/simulation/mod.rs +++ b/xet_client/src/cas_client/simulation/mod.rs @@ -30,7 +30,7 @@ mod deletion_controls; mod local_client; #[cfg(not(target_family = "wasm"))] -pub use deletion_controls::{DeletionControlableClient, ObjectTag, ObjectTagSet}; +pub use deletion_controls::{DeletionControlableClient, ObjectETag, ObjectTagSet}; #[cfg(not(target_family = "wasm"))] pub use local_client::LocalClient; From 7356034ab87ef128a20d7ce778231f85ee07c77e Mon Sep 17 00:00:00 2001 From: Sam Horradarn Date: Fri, 4 Sep 2026 13:42:13 +0200 Subject: [PATCH 3/8] fix(simulation): gate the xorb tag set field out of the wasm build `deletion_controls` is `#[cfg(not(target_family = "wasm"))]`, so `ObjectTagSet` does not exist on wasm, but `MemoryClient::xorb_tag_sets` was declared unconditionally. Native builds passed because the type is in scope there; wasm32 failed with E0425 on the field. Gate the field and its initializer to match the module. Co-Authored-By: Claude Opus 5 --- xet_client/src/cas_client/simulation/memory_client.rs | 2 ++ 1 file changed, 2 insertions(+) diff --git a/xet_client/src/cas_client/simulation/memory_client.rs b/xet_client/src/cas_client/simulation/memory_client.rs index 9c2e43924..a417a3dfe 100644 --- a/xet_client/src/cas_client/simulation/memory_client.rs +++ b/xet_client/src/cas_client/simulation/memory_client.rs @@ -91,6 +91,7 @@ pub struct MemoryClient { gc_tagged_shard: RwLock>, /// S3-style tag sets per XORB. Held separately from `xorbs` so writing one /// cannot perturb the bytes the [`ObjectETag`] is derived from. + #[cfg(not(target_family = "wasm"))] xorb_tag_sets: RwLock>, } @@ -111,6 +112,7 @@ impl MemoryClient { lifecycle_tag_deletion: AtomicBool::new(false), gc_tagged_xorbs: RwLock::new(HashSet::new()), gc_tagged_shard: RwLock::new(None), + #[cfg(not(target_family = "wasm"))] xorb_tag_sets: RwLock::new(MerkleHashMap::new()), }) } From 800366413fa01c79e31249852f4da24f90af7f62 Mon Sep 17 00:00:00 2001 From: Sam Horradarn Date: Fri, 4 Sep 2026 14:26:27 +0200 Subject: [PATCH 4/8] feat(simulation): stamp last-upload on xorb upload, behind a builder flag MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The tag set was only writable out of band, so a simulated upload produced an untagged xorb — a consumer reading `last-upload` saw nothing during an ordinary run and could only be exercised by a test stamping the tag by hand. `LocalTestServerBuilder::with_upload_tagging(true)` now makes `upload_xorb` stamp `last-upload=`, mirroring CAS. It sits beside the existing `gc_tagged_*` clear in both clients, and for the same reason: PutObject replaces an object's whole tag set, so the stamp both records this write and clears what was there. Off by default and opt-in like `with_lifecycle_tag_deletion`, so no existing behaviour changes; a xorb uploaded without it still has no tag set. `LAST_UPLOAD_TAG_KEY` and `last_upload_tag_set_now` are exported so a consumer can read the stamp without restating the key. xet-core reproduces the stamp; what the value means belongs to whoever reads it. Co-Authored-By: Claude Opus 5 --- .../simulation/deletion_controls.rs | 16 ++++++ .../src/cas_client/simulation/local_client.rs | 53 ++++++++++++++++++- .../cas_client/simulation/memory_client.rs | 50 ++++++++++++++++- xet_client/src/cas_client/simulation/mod.rs | 4 +- .../simulation/simulation_server.rs | 16 ++++++ 5 files changed, 136 insertions(+), 3 deletions(-) diff --git a/xet_client/src/cas_client/simulation/deletion_controls.rs b/xet_client/src/cas_client/simulation/deletion_controls.rs index cfe0a6adc..a16dda742 100644 --- a/xet_client/src/cas_client/simulation/deletion_controls.rs +++ b/xet_client/src/cas_client/simulation/deletion_controls.rs @@ -18,6 +18,22 @@ pub type ObjectETag = [u8; 32]; /// without invalidating anything keyed on its content. pub type ObjectTagSet = Vec<(String, String)>; +/// Tag key CAS stamps with the unix seconds of a xorb's most recent write. +/// +/// Named here so the simulation can model that write (see +/// `LocalTestServerBuilder::with_upload_tagging`). What the value *means* is +/// the reader's business — xet-core only reproduces the stamp. +pub const LAST_UPLOAD_TAG_KEY: &str = "last-upload"; + +/// The tag set CAS puts on a xorb it has just written. +pub fn last_upload_tag_set_now() -> ObjectTagSet { + let unix = std::time::SystemTime::now() + .duration_since(std::time::UNIX_EPOCH) + .map(|d| d.as_secs()) + .unwrap_or_default(); + vec![(LAST_UPLOAD_TAG_KEY.to_string(), unix.to_string())] +} + /// Trait for clients that support deletion and integrity operations on shards and file entries. /// /// Implemented by `LocalClient` (disk-backed) and `MemoryClient` (in-memory). diff --git a/xet_client/src/cas_client/simulation/local_client.rs b/xet_client/src/cas_client/simulation/local_client.rs index 319890c97..26036c279 100644 --- a/xet_client/src/cas_client/simulation/local_client.rs +++ b/xet_client/src/cas_client/simulation/local_client.rs @@ -31,7 +31,7 @@ use xet_runtime::core::XetContext; use xet_runtime::fd_diagnostics::{report_fd_count, track_fd_scope}; use xet_runtime::file_utils::SafeFileCreator; -use super::deletion_controls::{ObjectETag, ObjectTagSet}; +use super::deletion_controls::{ObjectETag, ObjectTagSet, last_upload_tag_set_now}; use super::direct_access_client::DirectAccessClient; use super::xorb_utils::{self, REFERENCE_INSTANT, duration_to_expiration_secs_ceil}; use crate::cas_client::Client; @@ -267,6 +267,10 @@ pub struct LocalClient { /// clears the `.gctag` file (matching S3 PutObject overwriting a tagged /// object). Off by default; opt in via [`Self::set_lifecycle_tag_deletion`]. lifecycle_tag_deletion: AtomicBool, + /// When true, `upload_xorb` stamps a `last-upload` tag set, modelling what + /// CAS does on every xorb write. Off by default; opt in via + /// [`Self::set_upload_tagging`]. + upload_tagging: AtomicBool, _tmp_dir: Option, } @@ -343,6 +347,7 @@ impl LocalClient { max_ranges_per_fetch: AtomicUsize::new(usize::MAX), v2_disabled_status: AtomicU16::new(0), lifecycle_tag_deletion: AtomicBool::new(false), + upload_tagging: AtomicBool::new(false), _tmp_dir: tmp_dir, }) } @@ -361,6 +366,15 @@ impl LocalClient { self.lifecycle_tag_deletion.store(on, Ordering::Relaxed); } + /// Toggle `last-upload` stamping on xorb upload (see [`Self::upload_tagging`]). + pub fn set_upload_tagging(&self, on: bool) { + self.upload_tagging.store(on, Ordering::Relaxed); + } + + fn upload_tagging_enabled(&self) -> bool { + self.upload_tagging.load(Ordering::Relaxed) + } + fn lifecycle_tag_deletion_enabled(&self) -> bool { self.lifecycle_tag_deletion.load(Ordering::Relaxed) } @@ -1700,6 +1714,25 @@ impl Client for LocalClient { // readable again. Mirrors S3 PutObject overwriting a tagged object. let _ = std::fs::remove_file(self.gctag_xorb_path(&hash)); + // CAS stamps `last-upload` on every xorb write, and PutObject replaces + // the whole tag set, so the stamp both records this write and clears + // whatever was there. + if self.upload_tagging_enabled() { + let path = self.tag_set_xorb_path(&hash); + match serde_json::to_vec(&last_upload_tag_set_now()) { + Ok(raw) => { + #[cfg(windows)] + if path.exists() { + Self::clear_readonly(&path); + } + if let Err(e) = std::fs::write(&path, raw) { + warn!("failed to stamp last-upload tag at {}: {e}", path.display()); + } + }, + Err(e) => warn!("failed to serialize last-upload tag set: {e}"), + } + } + info!("{file_path:?} successfully written with {bytes_written} bytes."); Ok(bytes_written as u64) @@ -2358,6 +2391,24 @@ mod tests { assert_eq!(tag_before, tag_after, "tagging must not move the ObjectETag"); } + /// With upload tagging on, an uploaded xorb carries `last-upload` and the + /// sidecar still does not show up as a xorb in either listing. + #[tokio::test] + async fn test_upload_tagging_stamps_last_upload() { + let client = LocalClient::temporary(test_context()).await.unwrap(); + client.set_upload_tagging(true); + + let file = client.upload_random_file(&[(1, (0, 2))], 2048).await.unwrap(); + let xorb_hash = file.terms[0].xorb_hash; + + let tags = client.get_xorb_tag_set(&xorb_hash).await.unwrap(); + let (key, value) = tags.first().expect("upload must stamp a tag set"); + assert_eq!(key, super::super::deletion_controls::LAST_UPLOAD_TAG_KEY); + assert!(value.parse::().is_ok(), "value must be unix seconds, got {value:?}"); + + assert_eq!(client.list_xorbs().await.unwrap(), vec![xorb_hash]); + } + /// Tagging an absent xorb is an error rather than creating an orphan sidecar. #[tokio::test] async fn test_xorb_tag_set_requires_the_xorb_to_exist() { diff --git a/xet_client/src/cas_client/simulation/memory_client.rs b/xet_client/src/cas_client/simulation/memory_client.rs index a417a3dfe..5630cc0de 100644 --- a/xet_client/src/cas_client/simulation/memory_client.rs +++ b/xet_client/src/cas_client/simulation/memory_client.rs @@ -27,7 +27,7 @@ use super::super::interface::{ShardUploadProgressCallback, ShardUploadProgressTy use super::super::progress_tracked_streams::ProgressCallback; use super::client_testing_utils::{FileTermReference, RandomFileContents}; #[cfg(not(target_family = "wasm"))] -use super::deletion_controls::{ObjectETag, ObjectTagSet}; +use super::deletion_controls::{ObjectETag, ObjectTagSet, last_upload_tag_set_now}; use super::direct_access_client::DirectAccessClient; use super::random_xorb::RandomXorb; use super::xorb_utils::{self, REFERENCE_INSTANT, duration_to_expiration_secs_ceil}; @@ -93,6 +93,10 @@ pub struct MemoryClient { /// cannot perturb the bytes the [`ObjectETag`] is derived from. #[cfg(not(target_family = "wasm"))] xorb_tag_sets: RwLock>, + /// When true, `upload_xorb` stamps a `last-upload` tag set, modelling what + /// CAS does on every xorb write. Off by default; opt in via + /// [`Self::set_upload_tagging`]. + upload_tagging: AtomicBool, } impl MemoryClient { @@ -114,6 +118,7 @@ impl MemoryClient { gc_tagged_shard: RwLock::new(None), #[cfg(not(target_family = "wasm"))] xorb_tag_sets: RwLock::new(MerkleHashMap::new()), + upload_tagging: AtomicBool::new(false), }) } @@ -122,6 +127,15 @@ impl MemoryClient { self.lifecycle_tag_deletion.store(on, Ordering::Relaxed); } + /// Toggle `last-upload` stamping on xorb upload (see [`Self::upload_tagging`]). + pub fn set_upload_tagging(&self, on: bool) { + self.upload_tagging.store(on, Ordering::Relaxed); + } + + fn upload_tagging_enabled(&self) -> bool { + self.upload_tagging.load(Ordering::Relaxed) + } + fn lifecycle_tag_deletion_enabled(&self) -> bool { self.lifecycle_tag_deletion.load(Ordering::Relaxed) } @@ -964,6 +978,14 @@ impl Client for MemoryClient { // Mirrors S3 PutObject overwriting a tagged object. self.gc_tagged_xorbs.write().await.remove(&hash); + // CAS stamps `last-upload` on every xorb write, and PutObject replaces + // the whole tag set, so the stamp both records this write and clears + // whatever was there. + #[cfg(not(target_family = "wasm"))] + if self.upload_tagging_enabled() { + self.xorb_tag_sets.write().await.insert(hash, last_upload_tag_set_now()); + } + if let Some(ref cb) = progress_callback { let n = bytes_written as u64; cb(n, n, n); @@ -1672,6 +1694,32 @@ mod tests { assert_eq!(client.list_xorbs_and_etags().await.unwrap(), before, "tagging must not move the ObjectETag"); } + /// With upload tagging on, an uploaded xorb carries `last-upload` without + /// its etag moving; with it off, no tag set appears at all. + #[tokio::test] + async fn test_upload_tagging_stamps_last_upload() { + let client = new_deletion_client(); + client.set_upload_tagging(true); + + let file = client.upload_random_file(&[(1, (0, 2))], 2048).await.unwrap(); + let xorb_hash = file.terms[0].xorb_hash; + + let tags = client.get_xorb_tag_set(&xorb_hash).await.unwrap(); + let (key, value) = tags.first().expect("upload must stamp a tag set"); + assert_eq!(key, super::super::deletion_controls::LAST_UPLOAD_TAG_KEY); + assert!(value.parse::().is_ok(), "value must be unix seconds, got {value:?}"); + } + + #[tokio::test] + async fn test_upload_tagging_off_leaves_no_tag_set() { + let client = new_deletion_client(); + assert!(!client.upload_tagging_enabled()); + + let file = client.upload_random_file(&[(1, (0, 2))], 2048).await.unwrap(); + let xorb_hash = file.terms[0].xorb_hash; + assert!(client.get_xorb_tag_set(&xorb_hash).await.unwrap().is_empty()); + } + #[tokio::test] async fn test_xorb_tag_set_requires_the_xorb_to_exist() { let client = new_deletion_client(); diff --git a/xet_client/src/cas_client/simulation/mod.rs b/xet_client/src/cas_client/simulation/mod.rs index 900f6a2b7..de586c1aa 100644 --- a/xet_client/src/cas_client/simulation/mod.rs +++ b/xet_client/src/cas_client/simulation/mod.rs @@ -30,7 +30,9 @@ mod deletion_controls; mod local_client; #[cfg(not(target_family = "wasm"))] -pub use deletion_controls::{DeletionControlableClient, ObjectETag, ObjectTagSet}; +pub use deletion_controls::{ + DeletionControlableClient, LAST_UPLOAD_TAG_KEY, ObjectETag, ObjectTagSet, last_upload_tag_set_now, +}; #[cfg(not(target_family = "wasm"))] pub use local_client::LocalClient; diff --git a/xet_client/src/cas_client/simulation/simulation_server.rs b/xet_client/src/cas_client/simulation/simulation_server.rs index 44ca9e4bf..099569e99 100644 --- a/xet_client/src/cas_client/simulation/simulation_server.rs +++ b/xet_client/src/cas_client/simulation/simulation_server.rs @@ -84,6 +84,7 @@ pub struct LocalTestServerBuilder { server_latency_profile: Option, network_profile: Option, lifecycle_tag_deletion: bool, + upload_tagging: bool, } #[allow(dead_code)] @@ -100,6 +101,7 @@ impl LocalTestServerBuilder { server_latency_profile: None, network_profile: None, lifecycle_tag_deletion: false, + upload_tagging: false, } } @@ -185,6 +187,17 @@ impl LocalTestServerBuilder { self } + /// When enabled, a xorb upload stamps a `last-upload` tag set carrying the + /// unix seconds of the write, modelling what CAS does on every xorb write. + /// Enable it to exercise a consumer that reads that tag; leave it off and + /// uploaded xorbs carry no tag set at all. + /// + /// Only applies to the `LocalClient` and `MemoryClient` backends. + pub fn with_upload_tagging(mut self, on: bool) -> Self { + self.upload_tagging = on; + self + } + /// Builds and starts the test server. pub async fn start(self) -> LocalTestServer { let ctx = XetContext::default().expect("XetContext::new"); @@ -208,6 +221,7 @@ impl LocalTestServerBuilder { } else if self.in_memory { let mc = MemoryClient::new(ctx.clone()); mc.set_lifecycle_tag_deletion(self.lifecycle_tag_deletion); + mc.set_upload_tagging(self.upload_tagging); let dc: Arc = mc.clone(); (mc, Some(dc)) } else if self.ephemeral_disk { @@ -215,6 +229,7 @@ impl LocalTestServerBuilder { .await .expect("Failed to create LocalClient with temporary directory"); lc.set_lifecycle_tag_deletion(self.lifecycle_tag_deletion); + lc.set_upload_tagging(self.upload_tagging); let dc: Arc = lc.clone(); (lc, Some(dc)) } else { @@ -225,6 +240,7 @@ impl LocalTestServerBuilder { .await .expect("Failed to create LocalClient"); lc.set_lifecycle_tag_deletion(self.lifecycle_tag_deletion); + lc.set_upload_tagging(self.upload_tagging); let dc: Arc = lc.clone(); (lc, Some(dc)) }; From 2858b8be2af26403e73a660073146a3eac859ede Mon Sep 17 00:00:00 2001 From: Sam Horradarn Date: Fri, 4 Sep 2026 14:38:08 +0200 Subject: [PATCH 5/8] refactor(simulation): stamp last-upload on every xorb upload, not behind a flag MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit CAS stamps `last-upload` on every xorb write unconditionally, so gating the simulation's stamp behind a builder flag left the default simulation diverging from production — the thing the simulation exists to reproduce. Modelling it after `with_lifecycle_tag_deletion` was the wrong analogy: that flag chooses between two real S3 deletion behaviours (hard delete vs. the lifecycle tag), whereas `last-upload` has no "off" in production. Drops `with_upload_tagging`, `set_upload_tagging` and the per-client flag. An uploaded xorb now always carries the tag, and the tests that asserted a fresh xorb has an empty tag set assert the round trip instead. Co-Authored-By: Claude Opus 5 --- .../simulation/deletion_controls.rs | 5 +-- .../src/cas_client/simulation/local_client.rs | 42 ++++++------------- .../cas_client/simulation/memory_client.rs | 30 +------------ .../simulation/simulation_server.rs | 16 ------- 4 files changed, 15 insertions(+), 78 deletions(-) diff --git a/xet_client/src/cas_client/simulation/deletion_controls.rs b/xet_client/src/cas_client/simulation/deletion_controls.rs index a16dda742..4655dbabf 100644 --- a/xet_client/src/cas_client/simulation/deletion_controls.rs +++ b/xet_client/src/cas_client/simulation/deletion_controls.rs @@ -20,9 +20,8 @@ pub type ObjectTagSet = Vec<(String, String)>; /// Tag key CAS stamps with the unix seconds of a xorb's most recent write. /// -/// Named here so the simulation can model that write (see -/// `LocalTestServerBuilder::with_upload_tagging`). What the value *means* is -/// the reader's business — xet-core only reproduces the stamp. +/// Every simulated xorb upload stamps this, as CAS does. What the value +/// *means* is the reader's business — xet-core only reproduces the stamp. pub const LAST_UPLOAD_TAG_KEY: &str = "last-upload"; /// The tag set CAS puts on a xorb it has just written. diff --git a/xet_client/src/cas_client/simulation/local_client.rs b/xet_client/src/cas_client/simulation/local_client.rs index 26036c279..46a23d3d3 100644 --- a/xet_client/src/cas_client/simulation/local_client.rs +++ b/xet_client/src/cas_client/simulation/local_client.rs @@ -267,10 +267,6 @@ pub struct LocalClient { /// clears the `.gctag` file (matching S3 PutObject overwriting a tagged /// object). Off by default; opt in via [`Self::set_lifecycle_tag_deletion`]. lifecycle_tag_deletion: AtomicBool, - /// When true, `upload_xorb` stamps a `last-upload` tag set, modelling what - /// CAS does on every xorb write. Off by default; opt in via - /// [`Self::set_upload_tagging`]. - upload_tagging: AtomicBool, _tmp_dir: Option, } @@ -347,7 +343,6 @@ impl LocalClient { max_ranges_per_fetch: AtomicUsize::new(usize::MAX), v2_disabled_status: AtomicU16::new(0), lifecycle_tag_deletion: AtomicBool::new(false), - upload_tagging: AtomicBool::new(false), _tmp_dir: tmp_dir, }) } @@ -366,15 +361,6 @@ impl LocalClient { self.lifecycle_tag_deletion.store(on, Ordering::Relaxed); } - /// Toggle `last-upload` stamping on xorb upload (see [`Self::upload_tagging`]). - pub fn set_upload_tagging(&self, on: bool) { - self.upload_tagging.store(on, Ordering::Relaxed); - } - - fn upload_tagging_enabled(&self) -> bool { - self.upload_tagging.load(Ordering::Relaxed) - } - fn lifecycle_tag_deletion_enabled(&self) -> bool { self.lifecycle_tag_deletion.load(Ordering::Relaxed) } @@ -1717,20 +1703,18 @@ impl Client for LocalClient { // CAS stamps `last-upload` on every xorb write, and PutObject replaces // the whole tag set, so the stamp both records this write and clears // whatever was there. - if self.upload_tagging_enabled() { - let path = self.tag_set_xorb_path(&hash); - match serde_json::to_vec(&last_upload_tag_set_now()) { - Ok(raw) => { - #[cfg(windows)] - if path.exists() { - Self::clear_readonly(&path); - } - if let Err(e) = std::fs::write(&path, raw) { - warn!("failed to stamp last-upload tag at {}: {e}", path.display()); - } - }, - Err(e) => warn!("failed to serialize last-upload tag set: {e}"), - } + let tag_set_path = self.tag_set_xorb_path(&hash); + match serde_json::to_vec(&last_upload_tag_set_now()) { + Ok(raw) => { + #[cfg(windows)] + if tag_set_path.exists() { + Self::clear_readonly(&tag_set_path); + } + if let Err(e) = std::fs::write(&tag_set_path, raw) { + warn!("failed to stamp last-upload tag at {}: {e}", tag_set_path.display()); + } + }, + Err(e) => warn!("failed to serialize last-upload tag set: {e}"), } info!("{file_path:?} successfully written with {bytes_written} bytes."); @@ -2359,7 +2343,6 @@ mod tests { let before = client.list_xorbs_and_etags().await.unwrap(); let (_, tag_before) = before.iter().find(|(h, _)| *h == xorb_hash).unwrap(); - assert!(client.get_xorb_tag_set(&xorb_hash).await.unwrap().is_empty(), "a fresh xorb has no tag set"); client .set_xorb_tag_set(&xorb_hash, vec![("last-upload".to_string(), "1234".to_string())]) @@ -2396,7 +2379,6 @@ mod tests { #[tokio::test] async fn test_upload_tagging_stamps_last_upload() { let client = LocalClient::temporary(test_context()).await.unwrap(); - client.set_upload_tagging(true); let file = client.upload_random_file(&[(1, (0, 2))], 2048).await.unwrap(); let xorb_hash = file.terms[0].xorb_hash; diff --git a/xet_client/src/cas_client/simulation/memory_client.rs b/xet_client/src/cas_client/simulation/memory_client.rs index 5630cc0de..7e1885de2 100644 --- a/xet_client/src/cas_client/simulation/memory_client.rs +++ b/xet_client/src/cas_client/simulation/memory_client.rs @@ -93,10 +93,6 @@ pub struct MemoryClient { /// cannot perturb the bytes the [`ObjectETag`] is derived from. #[cfg(not(target_family = "wasm"))] xorb_tag_sets: RwLock>, - /// When true, `upload_xorb` stamps a `last-upload` tag set, modelling what - /// CAS does on every xorb write. Off by default; opt in via - /// [`Self::set_upload_tagging`]. - upload_tagging: AtomicBool, } impl MemoryClient { @@ -118,7 +114,6 @@ impl MemoryClient { gc_tagged_shard: RwLock::new(None), #[cfg(not(target_family = "wasm"))] xorb_tag_sets: RwLock::new(MerkleHashMap::new()), - upload_tagging: AtomicBool::new(false), }) } @@ -127,15 +122,6 @@ impl MemoryClient { self.lifecycle_tag_deletion.store(on, Ordering::Relaxed); } - /// Toggle `last-upload` stamping on xorb upload (see [`Self::upload_tagging`]). - pub fn set_upload_tagging(&self, on: bool) { - self.upload_tagging.store(on, Ordering::Relaxed); - } - - fn upload_tagging_enabled(&self) -> bool { - self.upload_tagging.load(Ordering::Relaxed) - } - fn lifecycle_tag_deletion_enabled(&self) -> bool { self.lifecycle_tag_deletion.load(Ordering::Relaxed) } @@ -982,9 +968,7 @@ impl Client for MemoryClient { // the whole tag set, so the stamp both records this write and clears // whatever was there. #[cfg(not(target_family = "wasm"))] - if self.upload_tagging_enabled() { - self.xorb_tag_sets.write().await.insert(hash, last_upload_tag_set_now()); - } + self.xorb_tag_sets.write().await.insert(hash, last_upload_tag_set_now()); if let Some(ref cb) = progress_callback { let n = bytes_written as u64; @@ -1670,7 +1654,6 @@ mod tests { let xorb_hash = file.terms[0].xorb_hash; let before = client.list_xorbs_and_etags().await.unwrap(); - assert!(client.get_xorb_tag_set(&xorb_hash).await.unwrap().is_empty()); client .set_xorb_tag_set(&xorb_hash, vec![("last-upload".to_string(), "1234".to_string())]) @@ -1699,7 +1682,6 @@ mod tests { #[tokio::test] async fn test_upload_tagging_stamps_last_upload() { let client = new_deletion_client(); - client.set_upload_tagging(true); let file = client.upload_random_file(&[(1, (0, 2))], 2048).await.unwrap(); let xorb_hash = file.terms[0].xorb_hash; @@ -1710,16 +1692,6 @@ mod tests { assert!(value.parse::().is_ok(), "value must be unix seconds, got {value:?}"); } - #[tokio::test] - async fn test_upload_tagging_off_leaves_no_tag_set() { - let client = new_deletion_client(); - assert!(!client.upload_tagging_enabled()); - - let file = client.upload_random_file(&[(1, (0, 2))], 2048).await.unwrap(); - let xorb_hash = file.terms[0].xorb_hash; - assert!(client.get_xorb_tag_set(&xorb_hash).await.unwrap().is_empty()); - } - #[tokio::test] async fn test_xorb_tag_set_requires_the_xorb_to_exist() { let client = new_deletion_client(); diff --git a/xet_client/src/cas_client/simulation/simulation_server.rs b/xet_client/src/cas_client/simulation/simulation_server.rs index 099569e99..44ca9e4bf 100644 --- a/xet_client/src/cas_client/simulation/simulation_server.rs +++ b/xet_client/src/cas_client/simulation/simulation_server.rs @@ -84,7 +84,6 @@ pub struct LocalTestServerBuilder { server_latency_profile: Option, network_profile: Option, lifecycle_tag_deletion: bool, - upload_tagging: bool, } #[allow(dead_code)] @@ -101,7 +100,6 @@ impl LocalTestServerBuilder { server_latency_profile: None, network_profile: None, lifecycle_tag_deletion: false, - upload_tagging: false, } } @@ -187,17 +185,6 @@ impl LocalTestServerBuilder { self } - /// When enabled, a xorb upload stamps a `last-upload` tag set carrying the - /// unix seconds of the write, modelling what CAS does on every xorb write. - /// Enable it to exercise a consumer that reads that tag; leave it off and - /// uploaded xorbs carry no tag set at all. - /// - /// Only applies to the `LocalClient` and `MemoryClient` backends. - pub fn with_upload_tagging(mut self, on: bool) -> Self { - self.upload_tagging = on; - self - } - /// Builds and starts the test server. pub async fn start(self) -> LocalTestServer { let ctx = XetContext::default().expect("XetContext::new"); @@ -221,7 +208,6 @@ impl LocalTestServerBuilder { } else if self.in_memory { let mc = MemoryClient::new(ctx.clone()); mc.set_lifecycle_tag_deletion(self.lifecycle_tag_deletion); - mc.set_upload_tagging(self.upload_tagging); let dc: Arc = mc.clone(); (mc, Some(dc)) } else if self.ephemeral_disk { @@ -229,7 +215,6 @@ impl LocalTestServerBuilder { .await .expect("Failed to create LocalClient with temporary directory"); lc.set_lifecycle_tag_deletion(self.lifecycle_tag_deletion); - lc.set_upload_tagging(self.upload_tagging); let dc: Arc = lc.clone(); (lc, Some(dc)) } else { @@ -240,7 +225,6 @@ impl LocalTestServerBuilder { .await .expect("Failed to create LocalClient"); lc.set_lifecycle_tag_deletion(self.lifecycle_tag_deletion); - lc.set_upload_tagging(self.upload_tagging); let dc: Arc = lc.clone(); (lc, Some(dc)) }; From 0b2e1ea902106d042b3907236badfcf01e2bb3aa Mon Sep 17 00:00:00 2001 From: Sam Horradarn Date: Fri, 4 Sep 2026 15:52:13 +0200 Subject: [PATCH 6/8] refactor(simulation): derive the etag from content, not from when it was written MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The simulation moved a xorb's etag on every write — MemoryClient mixed in a generation counter, LocalClient hashed the file's mtime and ctime. A comment called that "matching production ETag semantics", and it no longer is: S3's ETag is a function of content, and CAS deliberately stopped rewriting a xorb on re-upload, so a byte-identical re-upload now leaves the ETag alone. Both clients now derive the etag from what is stored. Xorbs and shards are content-addressed, so the key fixes the content and the length distinguishes a differently serialized rewrite of the same key. The generation counter fed nothing else and is gone. This makes the simulation reproduce the hazard rather than hide it: a stale etag snapshot now matches a resurrected object, exactly as in production, and `last-upload` is what tells them apart. `test_list_xorbs_and_etags_timestamp_changes` asserted the old behaviour and is inverted accordingly. It also fixes a latent oddity in the conditional deletes, which computed the etag from a temp path they had just renamed the object onto — under the old derivation that was a different mtime, and so a different etag, from the object the caller had listed. Co-Authored-By: Claude Opus 5 --- .../simulation/deletion_controls.rs | 5 +- .../src/cas_client/simulation/local_client.rs | 56 +++++++++++-------- .../cas_client/simulation/memory_client.rs | 32 ++++------- 3 files changed, 48 insertions(+), 45 deletions(-) diff --git a/xet_client/src/cas_client/simulation/deletion_controls.rs b/xet_client/src/cas_client/simulation/deletion_controls.rs index 4655dbabf..bc8a3ca10 100644 --- a/xet_client/src/cas_client/simulation/deletion_controls.rs +++ b/xet_client/src/cas_client/simulation/deletion_controls.rs @@ -7,8 +7,9 @@ use crate::error::Result; /// An opaque 32-byte etag used for conditional deletion (compare-and-delete), standing in /// for S3's ETag. /// -/// Implementations should derive this from object metadata/content with enough entropy -/// to reduce false matches when objects are rapidly rewritten. +/// Derived from what is stored, not from when it was written: a byte-identical re-upload +/// leaves it unchanged, as S3's content-derived ETag does. A caller that needs to tell a +/// re-upload apart cannot do it from this value — see [`LAST_UPLOAD_TAG_KEY`]. pub type ObjectETag = [u8; 32]; /// S3-style `(key, value)` tag set attached to an object. diff --git a/xet_client/src/cas_client/simulation/local_client.rs b/xet_client/src/cas_client/simulation/local_client.rs index 46a23d3d3..da278d3d2 100644 --- a/xet_client/src/cas_client/simulation/local_client.rs +++ b/xet_client/src/cas_client/simulation/local_client.rs @@ -436,22 +436,22 @@ impl LocalClient { /// /// We hash multiple metadata fields to increase entropy and reduce false /// matches during rapid rewrite/delete races. - fn object_etag_from_path(path: &Path) -> Result { - let meta = std::fs::metadata(path).map_err(ClientError::internal)?; - let modified = meta.modified().map_err(ClientError::internal)?; - let modified_nanos = modified.duration_since(std::time::UNIX_EPOCH).unwrap_or_default().as_nanos(); - let created_nanos = meta - .created() - .ok() - .and_then(|ts| ts.duration_since(std::time::UNIX_EPOCH).ok()) - .map_or(0u128, |d| d.as_nanos()); - - let mut entropy = Vec::with_capacity(16 + 16 + 8 + 1); - entropy.extend_from_slice(&modified_nanos.to_le_bytes()); - entropy.extend_from_slice(&created_nanos.to_le_bytes()); - entropy.extend_from_slice(&meta.len().to_le_bytes()); - entropy.push(u8::from(meta.permissions().readonly())); - + /// The object's etag, derived from its key and stored length rather than + /// from filesystem timestamps. + /// + /// S3's ETag is a function of content, so a byte-identical re-upload leaves + /// it unchanged. Xorbs and shards are content-addressed, so the key already + /// fixes the content and the length is what distinguishes a differently + /// serialized rewrite of the same key. Hashing mtime/ctime instead made + /// every write look like different content — including a re-upload, and + /// including the rename onto a temp path that the conditional deletes do. + fn object_etag_for(prefix: &[u8], key: &MerkleHash, path: &Path) -> Result { + let len = std::fs::metadata(path).map_err(ClientError::internal)?.len(); + let key_bytes: [u8; 32] = (*key).into(); + let mut entropy = Vec::with_capacity(prefix.len() + key_bytes.len() + 8); + entropy.extend_from_slice(prefix); + entropy.extend_from_slice(&key_bytes); + entropy.extend_from_slice(&len.to_le_bytes()); Ok(compute_data_hash(&entropy).into()) } @@ -1017,7 +1017,7 @@ impl super::DeletionControlableClient for LocalClient { if let Some(pos) = name.rfind('.') { let hex = &name[(pos + 1)..]; if let Ok(hash) = MerkleHash::from_hex(hex) { - let etag = Self::object_etag_from_path(&path)?; + let etag = Self::object_etag_for(b"xorb", &hash, &path)?; ret.push((hash, etag)); } } @@ -1036,7 +1036,7 @@ impl super::DeletionControlableClient for LocalClient { return Err(ClientError::XORBNotFound(*hash)); } - let current_etag = match Self::object_etag_from_path(&tmp_path) { + let current_etag = match Self::object_etag_for(b"xorb", hash, &tmp_path) { Ok(t) => t, Err(e) => { Self::restore_from_tmp(&tmp_path, &file_path); @@ -1093,7 +1093,7 @@ impl super::DeletionControlableClient for LocalClient { async fn list_shards_with_etags(&self) -> Result> { let mut ret = Vec::new(); for (hash, path) in self.shard_file_paths()? { - let etag = Self::object_etag_from_path(&path)?; + let etag = Self::object_etag_for(b"shard", &hash, &path)?; ret.push((hash, etag)); } Ok(ret) @@ -1107,7 +1107,7 @@ impl super::DeletionControlableClient for LocalClient { return Err(ClientError::Other(format!("Shard not found: {}", hash.hex()))); } - let current_etag = match Self::object_etag_from_path(&tmp_path) { + let current_etag = match Self::object_etag_for(b"shard", hash, &tmp_path) { Ok(t) => t, Err(e) => { Self::restore_from_tmp(&tmp_path, &path); @@ -2401,8 +2401,15 @@ mod tests { } /// Tests that list_xorbs_and_etags etags change after file re-creation with a timestamp delay. + /// A xorb deleted and re-uploaded with the same content keeps its etag, + /// even across a filesystem timestamp change. + /// + /// This is S3's contract — the ETag is a function of content — and the + /// resurrection hazard it creates is real: a stale etag snapshot still + /// matches the new object. That is what the `last-upload` tag is for; the + /// etag deliberately no longer hides it. #[tokio::test] - async fn test_list_xorbs_and_etags_timestamp_changes() { + async fn test_etag_is_stable_across_an_identical_reupload() { let client = LocalClient::temporary(test_context()).await.unwrap(); let file1 = client.upload_random_file(&[(1, (0, 2))], 2048).await.unwrap(); @@ -2418,11 +2425,16 @@ mod tests { // Re-upload a file that creates a new xorb with the same hash seed. let file2 = client.upload_random_file(&[(1, (0, 2))], 2048).await.unwrap(); let xorb_hash2 = file2.terms[0].xorb_hash; + assert_eq!(xorb_hash, xorb_hash2, "the seed must reproduce the same xorb"); let tags2 = client.list_xorbs_and_etags().await.unwrap(); let (_, tag2) = tags2.iter().find(|(h, _)| *h == xorb_hash2).unwrap(); - assert_ne!(tag1, tag2, "Tags should differ after re-creation with timestamp delay"); + assert_eq!(tag1, tag2, "identical content must keep its etag despite a newer mtime"); + + // The `last-upload` tag is what moved instead. + let tags = client.get_xorb_tag_set(&xorb_hash).await.unwrap(); + assert_eq!(tags.first().map(|(k, _)| k.as_str()), Some(super::super::deletion_controls::LAST_UPLOAD_TAG_KEY)); } // ── Lifecycle-tag deletion mode tests ─────────────────────────────── diff --git a/xet_client/src/cas_client/simulation/memory_client.rs b/xet_client/src/cas_client/simulation/memory_client.rs index 7e1885de2..63ecddc6e 100644 --- a/xet_client/src/cas_client/simulation/memory_client.rs +++ b/xet_client/src/cas_client/simulation/memory_client.rs @@ -50,8 +50,8 @@ struct MaterializedXorb { /// every insert, so the etag changes even when content is identical (matching /// production ETag semantics). enum XorbStorage { - Materialized { entry: MaterializedXorb, generation: u64 }, - Random { xorb: RandomXorb, generation: u64 }, + Materialized { entry: MaterializedXorb }, + Random { xorb: RandomXorb }, } /// In-memory client for testing purposes. Stores all data in memory using hash tables. @@ -65,7 +65,6 @@ pub struct MemoryClient { /// Upload concurrency controller upload_concurrency_controller: Arc, /// Monotonic counter for xorb upload generations (etag freshness). - xorb_generation: AtomicU64, /// URL expiration in milliseconds url_expiration_ms: AtomicU64, /// Global dedup shard expiration in seconds (0 = disabled). @@ -103,7 +102,6 @@ impl MemoryClient { shard: RwLock::new(MDBInMemoryShard::default()), global_dedup: RwLock::new(MerkleHashMap::new()), upload_concurrency_controller: AdaptiveConcurrencyController::new_upload(ctx, "memory_uploads"), - xorb_generation: AtomicU64::new(0), url_expiration_ms: AtomicU64::new(u64::MAX), global_dedup_expiration_secs: AtomicU64::new(0), random_ms_delay_window: (AtomicU64::new(0), AtomicU64::new(0)), @@ -177,8 +175,7 @@ impl MemoryClient { shard.add_xorb_block(cas_info)?; } - let generation = self.xorb_generation.fetch_add(1, Ordering::Relaxed); - self.xorbs.write().await.insert(hash, XorbStorage::Random { xorb, generation }); + self.xorbs.write().await.insert(hash, XorbStorage::Random { xorb }); Ok(hash) } @@ -290,16 +287,11 @@ impl MemoryClient { #[cfg(not(target_family = "wasm"))] fn xorb_etag(hash: &MerkleHash, storage: &XorbStorage) -> ObjectETag { match storage { - XorbStorage::Materialized { entry, generation } => { - let mut payload = Vec::from(entry.serialized_data.as_ref()); - payload.extend_from_slice(&generation.to_le_bytes()); - Self::object_etag_from_key_and_payload(b"xorb", hash, &payload) + XorbStorage::Materialized { entry } => { + Self::object_etag_from_key_and_payload(b"xorb", hash, entry.serialized_data.as_ref()) }, - XorbStorage::Random { xorb, generation } => { - let mut entropy = Vec::with_capacity(16); - entropy.extend_from_slice(&xorb.num_chunks().to_le_bytes()); - entropy.extend_from_slice(&generation.to_le_bytes()); - Self::object_etag_from_key_and_payload(b"xorb", hash, &entropy) + XorbStorage::Random { xorb } => { + Self::object_etag_from_key_and_payload(b"xorb", hash, &xorb.num_chunks().to_le_bytes()) }, } } @@ -914,10 +906,10 @@ impl Client for MemoryClient { let footer_start = serialized_xorb_object.footer_start; let serialized_data = serialized_xorb_object.serialized_data; - // Always overwrite: even if the xorb already exists, we must store it - // with a fresh generation so its etag changes, matching production ETag - // semantics and ensuring delete_xorb_if_etag_matches is safe under - // concurrent uploads. + // Always overwrite, but the etag is derived from the stored bytes alone, + // so a byte-identical re-upload leaves it unchanged — as S3 does, whose + // ETag is a function of content. What tells a re-upload apart is the + // `last-upload` tag stamped below, not the etag. info!("Storing XORB {hash:?} in memory"); @@ -943,7 +935,6 @@ impl Client for MemoryClient { }; let bytes_written = serialized_data.len(); - let generation = self.xorb_generation.fetch_add(1, Ordering::Relaxed); { let mut xorbs = self.xorbs.write().await; @@ -954,7 +945,6 @@ impl Client for MemoryClient { serialized_data: Bytes::from(serialized_data), xorb_object: xorb_obj, }, - generation, }, ); } From 22e67d2ebdc2f2397d1141eb83ae7b23f61b87f3 Mon Sep 17 00:00:00 2001 From: Sam Horradarn Date: Fri, 4 Sep 2026 21:17:29 +0200 Subject: [PATCH 7/8] fix(simulation): drop a xorb's tag set when the xorb goes away MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Bugbot, on #958. `set_xorb_tag_set` wrote an entry that no delete path cleared, so a hard delete left the sidecar orphaned on disk (or the map entry resident), and the tags outlived the object they described. That is neither S3's behaviour — DeleteObject takes an object's tags with it — nor GC's, whose `gc-delete` write replaces the whole set and so drops `last-upload` either way. Both delete paths in both clients now clear it. Also makes `MemoryClient`'s tag-set accessors treat a lifecycle-tagged xorb as gone. Every other read there already does, and `LocalClient` errors on one because the canonical file has been renamed away, so the two backends disagreed on the same call. Note the re-upload half of the report was already covered: `upload_xorb` stamps `last-upload` unconditionally and replaces the whole set, so a re-uploaded xorb could not inherit stale tags. The leak was the orphan itself, which is what the new tests assert — the `MemoryClient` one reads the map directly, because a deleted xorb refuses `get_xorb_tag_set` and a re-upload would mask the leak by overwriting the entry. Co-Authored-By: Claude Opus 5 --- .../src/cas_client/simulation/local_client.rs | 41 ++++++++++ .../cas_client/simulation/memory_client.rs | 80 +++++++++++++++++-- 2 files changed, 115 insertions(+), 6 deletions(-) diff --git a/xet_client/src/cas_client/simulation/local_client.rs b/xet_client/src/cas_client/simulation/local_client.rs index da278d3d2..56f2e3aea 100644 --- a/xet_client/src/cas_client/simulation/local_client.rs +++ b/xet_client/src/cas_client/simulation/local_client.rs @@ -385,6 +385,21 @@ impl LocalClient { PathBuf::from(name) } + /// Drops a xorb's tag-set sidecar. Called from every path that removes the + /// object or condemns it: a hard delete takes the tags with it, and GC's + /// `gc-delete` write replaces the whole set, dropping `last-upload` either + /// way. Leaving the sidecar behind would orphan it on disk, and let a later + /// upload of the same hash inherit tags from the object that used to live + /// there. + fn clear_tag_set_xorb(&self, hash: &MerkleHash) { + let path = self.tag_set_xorb_path(hash); + #[cfg(windows)] + if path.exists() { + Self::clear_readonly(&path); + } + let _ = std::fs::remove_file(path); + } + /// Path used to park a tagged-for-deletion shard: `.mdb.gctag`. fn gctag_shard_path(&self, hash: &MerkleHash) -> PathBuf { let canonical = self.shard_dir.join(shard_file_name(hash)); @@ -1004,6 +1019,7 @@ impl super::DeletionControlableClient for LocalClient { } else { let _ = std::fs::remove_file(file_path); } + self.clear_tag_set_xorb(hash); } async fn list_xorbs_and_etags(&self) -> Result> { @@ -1060,6 +1076,7 @@ impl super::DeletionControlableClient for LocalClient { } else { std::fs::remove_file(&tmp_path)?; } + self.clear_tag_set_xorb(hash); Ok(true) } @@ -2391,6 +2408,30 @@ mod tests { assert_eq!(client.list_xorbs().await.unwrap(), vec![xorb_hash]); } + /// A hard delete removes the sidecar, so it is not orphaned on disk and a + /// later upload of the same hash cannot inherit the old tags. + #[tokio::test] + async fn test_delete_xorb_clears_its_tag_set_sidecar() { + let client = LocalClient::temporary(test_context()).await.unwrap(); + let file = client.upload_random_file(&[(1, (0, 2))], 2048).await.unwrap(); + let xorb_hash = file.terms[0].xorb_hash; + + client + .set_xorb_tag_set(&xorb_hash, vec![("stale".to_string(), "value".to_string())]) + .await + .unwrap(); + let sidecar = client.tag_set_xorb_path(&xorb_hash); + assert!(sidecar.exists()); + + client.delete_xorb(&xorb_hash).await; + assert!(!sidecar.exists(), "the sidecar outlived the xorb"); + + let refile = client.upload_random_file(&[(1, (0, 2))], 2048).await.unwrap(); + assert_eq!(refile.terms[0].xorb_hash, xorb_hash); + let tags = client.get_xorb_tag_set(&xorb_hash).await.unwrap(); + assert!(!tags.iter().any(|(k, _)| k == "stale"), "stale tags survived a delete: {tags:?}"); + } + /// Tagging an absent xorb is an error rather than creating an orphan sidecar. #[tokio::test] async fn test_xorb_tag_set_requires_the_xorb_to_exist() { diff --git a/xet_client/src/cas_client/simulation/memory_client.rs b/xet_client/src/cas_client/simulation/memory_client.rs index 63ecddc6e..5dd1174ce 100644 --- a/xet_client/src/cas_client/simulation/memory_client.rs +++ b/xet_client/src/cas_client/simulation/memory_client.rs @@ -128,6 +128,30 @@ impl MemoryClient { self.gc_tagged_xorbs.read().await.contains(hash) } + /// Drops a xorb's tag set. Called from every path that removes the object + /// or condemns it: a hard delete takes the tags with it, and GC's + /// `gc-delete` write replaces the whole set, dropping `last-upload` either + /// way. Leaving them behind would let a later upload of the same hash + /// inherit tags from the object that used to live there. + #[cfg(not(target_family = "wasm"))] + async fn clear_xorb_tag_set(&self, hash: &MerkleHash) { + self.xorb_tag_sets.write().await.remove(hash); + } + + #[cfg(target_family = "wasm")] + async fn clear_xorb_tag_set(&self, _hash: &MerkleHash) {} + + /// Errors unless the xorb is present and not condemned. A lifecycle-tagged + /// xorb reads as gone everywhere else here, and `LocalClient` errors on it + /// because the canonical file has been renamed away. + #[cfg(not(target_family = "wasm"))] + async fn require_readable_xorb(&self, hash: &MerkleHash) -> Result<()> { + if !self.xorbs.read().await.contains_key(hash) || self.xorb_is_tagged(hash).await { + return Err(ClientError::Other(format!("XORB not found: {}", hash.hex()))); + } + Ok(()) + } + async fn shard_is_tagged(&self, hash: &MerkleHash) -> bool { self.gc_tagged_shard.read().await.is_some_and(|h| &h == hash) } @@ -1206,6 +1230,7 @@ impl super::DeletionControlableClient for MemoryClient { } else { self.xorbs.write().await.remove(hash); } + self.clear_xorb_tag_set(hash).await; } async fn list_xorbs_and_etags(&self) -> Result> { @@ -1234,20 +1259,17 @@ impl super::DeletionControlableClient for MemoryClient { } else { self.xorbs.write().await.remove(hash); } + self.clear_xorb_tag_set(hash).await; Ok(true) } async fn get_xorb_tag_set(&self, hash: &MerkleHash) -> Result { - if !self.xorbs.read().await.contains_key(hash) { - return Err(ClientError::Other(format!("XORB not found: {}", hash.hex()))); - } + self.require_readable_xorb(hash).await?; Ok(self.xorb_tag_sets.read().await.get(hash).cloned().unwrap_or_default()) } async fn set_xorb_tag_set(&self, hash: &MerkleHash, tags: ObjectTagSet) -> Result<()> { - if !self.xorbs.read().await.contains_key(hash) { - return Err(ClientError::Other(format!("XORB not found: {}", hash.hex()))); - } + self.require_readable_xorb(hash).await?; self.xorb_tag_sets.write().await.insert(*hash, tags); Ok(()) } @@ -1682,6 +1704,52 @@ mod tests { assert!(value.parse::().is_ok(), "value must be unix seconds, got {value:?}"); } + /// A hard delete takes the tag set with it, so a later upload of the same + /// hash cannot inherit tags from the object that used to live there. + #[tokio::test] + async fn test_delete_xorb_clears_its_tag_set() { + let client = new_deletion_client(); + let file = client.upload_random_file(&[(1, (0, 2))], 2048).await.unwrap(); + let xorb_hash = file.terms[0].xorb_hash; + + client + .set_xorb_tag_set(&xorb_hash, vec![("stale".to_string(), "value".to_string())]) + .await + .unwrap(); + client.delete_xorb(&xorb_hash).await; + + // Checked against the map directly: `get_xorb_tag_set` refuses a deleted + // xorb, so the leak is invisible through the public surface, and a + // re-upload would mask it by replacing the entry wholesale. + assert!( + !client.xorb_tag_sets.read().await.contains_key(&xorb_hash), + "the tag set outlived the xorb it belonged to" + ); + + // And a re-upload of the same content carries only its own fresh stamp. + let refile = client.upload_random_file(&[(1, (0, 2))], 2048).await.unwrap(); + assert_eq!(refile.terms[0].xorb_hash, xorb_hash); + let tags = client.get_xorb_tag_set(&xorb_hash).await.unwrap(); + assert!(!tags.iter().any(|(k, _)| k == "stale"), "stale tags survived a delete: {tags:?}"); + assert_eq!(tags.len(), 1, "only the fresh last-upload stamp should be present: {tags:?}"); + } + + /// A lifecycle-tagged xorb reads as gone everywhere else here, and + /// `LocalClient` errors on it because the canonical file is renamed away. + /// The tag-set accessors must agree rather than quietly serving it. + #[tokio::test] + async fn test_tag_set_accessors_treat_a_condemned_xorb_as_gone() { + let client = new_deletion_client(); + client.set_lifecycle_tag_deletion(true); + let file = client.upload_random_file(&[(1, (0, 2))], 2048).await.unwrap(); + let xorb_hash = file.terms[0].xorb_hash; + + client.delete_xorb(&xorb_hash).await; + + assert!(client.get_xorb_tag_set(&xorb_hash).await.is_err()); + assert!(client.set_xorb_tag_set(&xorb_hash, vec![]).await.is_err()); + } + #[tokio::test] async fn test_xorb_tag_set_requires_the_xorb_to_exist() { let client = new_deletion_client(); From 36d36d5f7f32b3b9d036aab5a42ca9b3bd574004 Mon Sep 17 00:00:00 2001 From: Sam Horradarn Date: Fri, 11 Sep 2026 21:47:58 +0200 Subject: [PATCH 8/8] fix(simulation): make xorb tag-set writes atomic and delete-safe `set_xorb_tag_set` truncated the sidecar in place, so a concurrent `get_xorb_tag_set` could read a half-written file; it now goes through `SafeFileCreator`, which writes a temp file and renames. In `MemoryClient` the xorb map and the tag set were mutated under separate locks, so a delete could land between an upload's write and its `last-upload` stamp and erase the tag of the xorb just written. Upload and both delete paths now hold the maps across the whole mutation. The conditional delete additionally compared the etag under a lock it dropped before deleting, which let a re-upload be deleted on the strength of an etag it no longer had; that comparison is now inside the same section. Locks are taken `gc_tagged_xorbs` before `xorbs` to match every existing reader, so writers cannot deadlock against one holding both. --- .../src/cas_client/simulation/local_client.rs | 6 +- .../cas_client/simulation/memory_client.rs | 66 ++++++++++--------- 2 files changed, 41 insertions(+), 31 deletions(-) diff --git a/xet_client/src/cas_client/simulation/local_client.rs b/xet_client/src/cas_client/simulation/local_client.rs index 56f2e3aea..67cf34f99 100644 --- a/xet_client/src/cas_client/simulation/local_client.rs +++ b/xet_client/src/cas_client/simulation/local_client.rs @@ -1103,7 +1103,11 @@ impl super::DeletionControlableClient for LocalClient { if path.exists() { Self::clear_readonly(&path); } - std::fs::write(&path, raw)?; + // Temp file plus rename: `get_xorb_tag_set` reads this path without coordination, + // so a truncate-in-place write would let it observe a half-written tag set. + let mut file = SafeFileCreator::replace_existing(&path)?; + file.write_all(&raw)?; + file.close()?; Ok(()) } diff --git a/xet_client/src/cas_client/simulation/memory_client.rs b/xet_client/src/cas_client/simulation/memory_client.rs index 5dd1174ce..0c07a0985 100644 --- a/xet_client/src/cas_client/simulation/memory_client.rs +++ b/xet_client/src/cas_client/simulation/memory_client.rs @@ -128,19 +128,6 @@ impl MemoryClient { self.gc_tagged_xorbs.read().await.contains(hash) } - /// Drops a xorb's tag set. Called from every path that removes the object - /// or condemns it: a hard delete takes the tags with it, and GC's - /// `gc-delete` write replaces the whole set, dropping `last-upload` either - /// way. Leaving them behind would let a later upload of the same hash - /// inherit tags from the object that used to live there. - #[cfg(not(target_family = "wasm"))] - async fn clear_xorb_tag_set(&self, hash: &MerkleHash) { - self.xorb_tag_sets.write().await.remove(hash); - } - - #[cfg(target_family = "wasm")] - async fn clear_xorb_tag_set(&self, _hash: &MerkleHash) {} - /// Errors unless the xorb is present and not condemned. A lifecycle-tagged /// xorb reads as gone everywhere else here, and `LocalClient` errors on it /// because the canonical file has been renamed away. @@ -961,7 +948,15 @@ impl Client for MemoryClient { let bytes_written = serialized_data.len(); { + // The write, the un-tag and the `last-upload` stamp are one critical section: + // a delete interleaving between them would erase the tag of the xorb this call + // just wrote. Locks are taken tagged-before-xorbs everywhere, so readers that + // hold both cannot deadlock against this. + let mut tagged = self.gc_tagged_xorbs.write().await; let mut xorbs = self.xorbs.write().await; + #[cfg(not(target_family = "wasm"))] + let mut tag_sets = self.xorb_tag_sets.write().await; + xorbs.insert( hash, XorbStorage::Materialized { @@ -971,18 +966,12 @@ impl Client for MemoryClient { }, }, ); - } - // A re-upload of the same xorb hash supersedes any prior - // lifecycle-tagged copy: clear the tag so the xorb is readable again. - // Mirrors S3 PutObject overwriting a tagged object. - self.gc_tagged_xorbs.write().await.remove(&hash); + tagged.remove(&hash); - // CAS stamps `last-upload` on every xorb write, and PutObject replaces - // the whole tag set, so the stamp both records this write and clears - // whatever was there. - #[cfg(not(target_family = "wasm"))] - self.xorb_tag_sets.write().await.insert(hash, last_upload_tag_set_now()); + #[cfg(not(target_family = "wasm"))] + tag_sets.insert(hash, last_upload_tag_set_now()); + } if let Some(ref cb) = progress_callback { let n = bytes_written as u64; @@ -1225,12 +1214,20 @@ impl super::DeletionControlableClient for MemoryClient { } async fn delete_xorb(&self, hash: &MerkleHash) { + // Removal and tag clear under one critical section, so a concurrent upload cannot + // land between them and have its fresh `last-upload` stamp erased by this delete. + let mut tagged = self.gc_tagged_xorbs.write().await; + let mut xorbs = self.xorbs.write().await; + #[cfg(not(target_family = "wasm"))] + let mut tag_sets = self.xorb_tag_sets.write().await; + if self.lifecycle_tag_deletion_enabled() { - self.gc_tagged_xorbs.write().await.insert(*hash); + tagged.insert(*hash); } else { - self.xorbs.write().await.remove(hash); + xorbs.remove(hash); } - self.clear_xorb_tag_set(hash).await; + #[cfg(not(target_family = "wasm"))] + tag_sets.remove(hash); } async fn list_xorbs_and_etags(&self) -> Result> { @@ -1244,8 +1241,15 @@ impl super::DeletionControlableClient for MemoryClient { } async fn delete_xorb_if_etag_matches(&self, hash: &MerkleHash, etag: &ObjectETag) -> Result { + // The etag comparison and the delete share one critical section. Dropping the lock + // between them would let a re-upload slip in and be deleted on the strength of the + // etag it no longer has, which is the very thing this guard exists to prevent. + let mut tagged = self.gc_tagged_xorbs.write().await; + let mut xorbs = self.xorbs.write().await; + #[cfg(not(target_family = "wasm"))] + let mut tag_sets = self.xorb_tag_sets.write().await; + let current_etag = { - let xorbs = self.xorbs.read().await; let Some(storage) = xorbs.get(hash) else { return Err(ClientError::XORBNotFound(*hash)); }; @@ -1254,12 +1258,14 @@ impl super::DeletionControlableClient for MemoryClient { if ¤t_etag != etag { return Ok(false); } + if self.lifecycle_tag_deletion_enabled() { - self.gc_tagged_xorbs.write().await.insert(*hash); + tagged.insert(*hash); } else { - self.xorbs.write().await.remove(hash); + xorbs.remove(hash); } - self.clear_xorb_tag_set(hash).await; + #[cfg(not(target_family = "wasm"))] + tag_sets.remove(hash); Ok(true) }