diff --git a/extensions/s3/API.md b/extensions/s3/API.md index abecf8d5..05c213f9 100644 --- a/extensions/s3/API.md +++ b/extensions/s3/API.md @@ -1,5 +1,9 @@ # Prolly S3 API guide +Stable-release compatibility, upgrade/downgrade recovery, production cache, +OpenTelemetry, and provider/cardinality support rules are defined in +[`GA-CONTRACT.md`](GA-CONTRACT.md). + The application-facing type is `prolly_s3_client::Client`. This guide describes the public client surface in version 0.1.0 and separates ordinary application operations from administrative maintenance. diff --git a/extensions/s3/CACHE-AND-SCALE-DESIGN.md b/extensions/s3/CACHE-AND-SCALE-DESIGN.md index 34624971..70b0a1ed 100644 --- a/extensions/s3/CACHE-AND-SCALE-DESIGN.md +++ b/extensions/s3/CACHE-AND-SCALE-DESIGN.md @@ -24,8 +24,10 @@ content-hash verified before use. The client supports: - bounded in-process memory caching for hot nodes and pack locations; -- optional Foyer memory/disk caching behind the `foyer-cache` feature; +- Foyer memory/disk caching enabled by the default client feature and required + by `ProductionCacheProfile`; - safe cache persistence across restarts; +- byte-bounded root/upper-level pinning and cardinality-derived sizing; - provider reads as the authoritative fallback. Cache corruption becomes a miss or a validation error; it cannot silently @@ -53,9 +55,10 @@ A process should: 1. open the repository and validate provider attestation; 2. load the branch ref and index heads; -3. open the persistent Foyer cache, if configured; +3. open the persistent Foyer cache (required by the production profile); 4. catch indexes up to the current journal generation; -5. optionally prewarm upper tree levels for known hot branches. +5. prewarm and pin a bounded number of upper tree levels for the attached + branch, subject to the configured startup timeout. Persisting immutable nodes removes the 421+ request cold-traversal pattern seen in early prototypes. A fully cold cache still performs bounded tree-depth and @@ -118,10 +121,11 @@ block-index and entry overhead. Prewarming is advisory and cancellable. A reader remains correct if prewarming never runs or the cache directory is deleted. -## Remaining production gaps +## Remaining qualification gaps - production garbage collection for unreachable immutable data; - published AWS qualification at customer-specific scale and traffic; -- automatic cache sizing from observed working set; +- feedback-driven cache resizing from observed working set (the initial size is + derived from expected cardinality); - operational SLOs for index lag and rebuild completion; - cross-region and disaster-recovery workflows. diff --git a/extensions/s3/Cargo.lock b/extensions/s3/Cargo.lock index e743c91f..b96ed05a 100644 --- a/extensions/s3/Cargo.lock +++ b/extensions/s3/Cargo.lock @@ -1801,6 +1801,15 @@ version = "0.2.1" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "7c87def4c32ab89d880effc9e097653c8da5d6ef28e6b539d313baaacfbafcbe" +[[package]] +name = "opentelemetry" +version = "0.32.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "b0142c63252a9e054e68a4c61a5778f7b14f576274d593f8ce883d191a099682" +dependencies = [ + "js-sys", +] + [[package]] name = "outref" version = "0.5.2" @@ -1985,6 +1994,7 @@ dependencies = [ "hmac 0.12.1", "http-body 1.1.0", "md-5 0.10.6", + "opentelemetry", "prolly-s3-core", "serde", "serde_json", diff --git a/extensions/s3/GA-CONTRACT.md b/extensions/s3/GA-CONTRACT.md new file mode 100644 index 00000000..abdf9c7d --- /dev/null +++ b/extensions/s3/GA-CONTRACT.md @@ -0,0 +1,137 @@ +# Prolly S3 general-availability contract + +Status: proposed contract for the first stable release. The `0.1.x` crates are +still a preview and do not yet carry the GA compatibility promise below. + +## Storage boundary + +One logical live file is one complete immutable provider object. Prolly S3 +does not pack payloads, split payloads into chunks, or own multipart-upload +state. Prolly metadata nodes may share a range-addressable commit object; +logical payloads never do. + +## Repository-format compatibility + +The create-once `format/repository.cbor` marker is the repository's canonical +format identity. Its tree descriptor, canonical limits, idempotency retention, +and provider version-limit profile are immutable for the lifetime of the +repository prefix. + +The first stable release will apply these rules: + +- patch and minor client upgrades must open and write every repository created + by an earlier release in the same major series; +- new readers may supply documented defaults for fields absent from older + objects, while canonical decoding rejects unknown fields, object magic, and + version values rather than silently changing their meaning; +- a writer must fail closed with `RepositoryFormatConflict` or + `UnsupportedRepositoryFormat` before publishing when it cannot reproduce the + stored canonical format exactly; +- caches, journal-derived indexes, rightmost-path hints, and prewarm state are + advisory and may be deleted/rebuilt by any compatible release; +- commit IDs, object-version IDs, payload bindings, authority epochs, branch + generations, and published state roots are durable protocol state and may + not be rewritten during an upgrade; +- a future incompatible format uses a new major protocol generation and a new + repository prefix. It is transferred through the verified history-transfer + API, never migrated in place. + +The repository core already enforces exact create-once canonical settings on +open. Commit metadata and node packs use explicit versioned wire magic and +content hashes; unsupported or corrupt encodings fail before becoming visible. + +## Upgrade, downgrade, and recovery guarantees + +Before an upgrade: + +1. complete or cancel active merges, restores, transfers, fsck, and GC jobs; +2. retain a provider-versioned backup or verified replica and the provider + attestation; +3. record every branch/tag/pin target and run shallow fsck; +4. open the repository read-only with the candidate release and rebuild + advisory indexes/cache from an empty local directory; +5. run the provider- and cardinality-specific qualification gate before + enabling writers. + +A downgrade is supported only when the older binary declares support for the +exact stored repository format and every required feature. Canonical derived +indexes written by a newer binary can intentionally fail closed in an older +binary; downgrade preparation must reset those advisory index heads with the +newer release before the older release rebuilds them. Otherwise restore the +pre-upgrade provider snapshot or transfer verified history into a new prefix. +Downgrade code must never delete newer immutable objects or rewrite the format +marker. If a release does not publish and test that reset procedure, downgrade +from that release is explicitly unsupported. + +Recovery guarantees are based on immutable closure and fenced mutable +controls: branch publication is CAS-protected, retry outcomes are reconciled, +authority takeover fences old epochs, commit sessions and maintenance cursors +are restartable, and exact physical versions are deleted only after the GC +safety protocol. Recovery does not guarantee retention of unreachable payloads +outside the configured GC policy. + +## Production cache profile + +`ProductionCacheProfile` is the supported production path. It requires a +single-owner persistent Foyer directory, chooses memory/disk/location bounds +from expected repository cardinality, enables bounded sibling prefetch, and +prewarms and pins root/upper-level metadata nodes. New repositories initialized +with the profile use encoded-byte-bounded metadata nodes; existing repositories +retain their create-once tree format and still use exact node-pack range reads. + +The profile is a latency requirement, never a correctness dependency. Removing +the cache must leave every repository operation correct, although cold SLOs may +fail until prewarming completes. + +## OpenTelemetry and reference alerts + +Build `prolly-s3-client` with `opentelemetry` and attach an +`OpenTelemetryClientMetrics` sink through `ClientBuilder::telemetry`. The +application owns the `MeterProvider`, exporter, resource attributes, sampling, +and shutdown. The client exports bounded-dimension metrics for: + +- cache hits/misses, admissions, errors, corruptions, and singleflight waits; +- requested, provider-fetched, and cache-avoided metadata bytes; +- predictive prefetch batches/nodes; +- S3 operations and transferred bytes; +- total open, index catch-up, and prewarm duration/failure. + +Recommended initial alerts, tuned after a representative baseline: + +| Condition | Initial threshold | Window | +| --- | ---: | ---: | +| Metadata cache hit ratio | below 90% after warmup | 15 minutes | +| Provider-fetched / requested metadata bytes | above 4x | 10 minutes | +| Cache admission rejects | above 1% of node requests | 10 minutes | +| Cache corruption | any sustained nonzero value | 5 minutes | +| Startup prewarm timeout/failure | any occurrence | immediate | +| Branch-index lag | above 100 generations or not ready | 5 minutes | +| S3 429/503 wire attempts | above provider-specific error budget | 5 minutes | +| GC/fsck checkpoint progress | no progress during an active job | 15 minutes | + +SDK operation counters do not include SDK-internal retries. Attach +`S3WireAttemptInterceptor` and provider-side request metrics for retry/error +alerts. + +## Explicit support envelope + +`SupportedEnvelope::for_deployment` exposes the same policy to applications. +The current evidence supports only a controlled RustFS/local pilot through +100K objects after its release gates pass. AWS always requires workload-specific +qualification. Repositories above 100K require cardinality-matched maintenance +and performance qualification, and one-million-object production support is +not claimed while the published cold and ingestion gates remain below target. + +Promotion requires all of the following at the intended provider, region, +cardinality, key distribution, concurrency, retention, and cache size: + +- provider capability and lifecycle/Object Lock/replication attestation; +- cold, prewarmed, steady-state, and cache-loss read/list SLOs; +- ingest throughput and request/byte amplification budgets; +- branch, arbitrary-snapshot diff, and merge SLOs; +- authority expiry/takeover and process-loss fault injection; +- fsck, journal-driven GC, restart, backup, and restore drills; +- operator dashboards, alerts, runbooks, IAM/KMS review, and cost approval. + +RustFS conformance proves protocol behavior, not AWS latency, throttling, cost, +or operational readiness. diff --git a/extensions/s3/OPERATIONS.md b/extensions/s3/OPERATIONS.md index 37220f1a..9aa00e43 100644 --- a/extensions/s3/OPERATIONS.md +++ b/extensions/s3/OPERATIONS.md @@ -1,5 +1,8 @@ # Prolly S3 operations +The stable upgrade/recovery contract, exported OpenTelemetry instruments, and +reference production alerts are defined in [GA-CONTRACT.md](GA-CONTRACT.md). + ## Provisioning 1. Create a dedicated S3 or S3-compatible bucket. diff --git a/extensions/s3/README.md b/extensions/s3/README.md index 708f6dec..cd1b05a7 100644 --- a/extensions/s3/README.md +++ b/extensions/s3/README.md @@ -1,5 +1,10 @@ # Prolly S3 +The proposed stable compatibility, cache, telemetry, recovery, and provider +support promises are defined in [GA-CONTRACT.md](GA-CONTRACT.md). The current +`0.1.x` crates remain suitable for controlled pilots after the qualification +gates in that contract pass; they are not yet a universal GA claim. + Prolly S3 adds repository history to a versioned S3 bucket. It stores each file as one immutable, content-addressed S3 object and stores directory state, commits, branches, tags, and indexes as Prolly trees. diff --git a/extensions/s3/client/Cargo.toml b/extensions/s3/client/Cargo.toml index efb040d7..82e6b95f 100644 --- a/extensions/s3/client/Cargo.toml +++ b/extensions/s3/client/Cargo.toml @@ -8,8 +8,9 @@ license = "MIT OR Apache-2.0" repository = "https://github.com/crabbuild/prolly" [features] -default = [] +default = ["foyer-cache"] foyer-cache = ["dep:foyer"] +opentelemetry = ["dep:opentelemetry"] [dependencies] async-trait = "0.1" @@ -24,6 +25,7 @@ hmac = "0.12" http-body = "1" md-5 = "0.10" hex = "0.4" +opentelemetry = { version = "=0.32.0", default-features = false, features = ["metrics"], optional = true } prolly-s3-core = { path = "../core", version = "0.1.0" } serde = { version = "1.0", features = ["derive"] } sha2 = "0.10" diff --git a/extensions/s3/client/README.md b/extensions/s3/client/README.md index ca26e9f1..db752667 100644 --- a/extensions/s3/client/README.md +++ b/extensions/s3/client/README.md @@ -520,8 +520,44 @@ are discovered and journaled by the same protocol. ## Cache immutable nodes -The in-memory cache is enabled through repository limits. For a persistent -Foyer cache, enable the `foyer-cache` feature: +The `foyer-cache` feature is enabled by default. Production deployments should +use the cardinality-aware profile so persistent storage, metadata-node bounds, +startup prewarming, and upper-level pinning are configured together: + +```rust +use prolly_s3_client::{Client, ProductionCacheProfile}; + +let profile = ProductionCacheProfile::new( + "./prolly-node-cache", + 1_000_000, // expected live logical objects +); +let client = Client::builder() + // supply the required AWS client, provider identity, signer, bucket, etc. + .production_cache_profile(profile) + .open() + .await?; + +let startup = client.startup_metrics(); +let performance = client.performance_snapshot(); +println!( + "startup={}ms hit_ratio={:.3} metadata_amplification={:.3}x", + startup.total_open_millis, + performance.cache.hit_ratio(), + performance.metadata_download_amplification(), +); +``` + +`CacheSizingRecommendation::for_object_count` exposes the selected memory, +disk, location, and prewarm bounds without opening a client. Supplying the +profile always opens persistent Foyer storage; `--no-default-features` callers +must enable `foyer-cache` explicitly. + +After stopping request traffic and dropping other client clones, call +`client.close_production_cache().await?` to flush and close the cache opened by +the production profile. This also stops the shared authority, branch-index, +and telemetry maintenance tasks for that client. + +For custom deployments, construct the cache directly: ```rust use std::path::PathBuf; @@ -559,8 +595,34 @@ Use `prewarm_node_cache(snapshot)` during startup to traverse both state trees. Use `prewarm_node_cache_levels(snapshot, levels)` when startup should load only the roots and shared upper paths instead of scanning every leaf. Use `node_cache_snapshot()` before and after to observe hits, misses, -insertions, corruptions, coalesced waits, ranged fetches, fetched and avoided -bytes, and admission rejections. +insertions, corruptions, coalesced waits, ranged fetches, requested/fetched/ +avoided bytes, byte amplification, predictive prefetches, pinned nodes, and +admission rejections. +Take `performance_snapshot()` before and after a metadata-only operation and +use `delta_since` plus `metadata_download_amplification()` to include commit, +index, and control-object response bytes in the measured amplification. + +## OpenTelemetry + +Enable the `opentelemetry` feature and supply the application-owned meter. The +client records deltas on a bounded maintenance interval; the embedding service +continues to own the SDK, exporter, resource attributes, and shutdown: + +```rust +use std::{sync::Arc, time::Duration}; +use opentelemetry::global; +use prolly_s3_client::{Client, OpenTelemetryClientMetrics}; + +let telemetry = OpenTelemetryClientMetrics::new(global::meter("prolly-s3")); +let client = Client::builder() + // supply the remaining required settings + .telemetry(telemetry, Duration::from_secs(15)) + .open() + .await?; +``` + +Metric names and initial alert thresholds are defined in +[`GA-CONTRACT.md`](../GA-CONTRACT.md). ## Performance model diff --git a/extensions/s3/client/examples/rustfs_small_files_benchmark.rs b/extensions/s3/client/examples/rustfs_small_files_benchmark.rs index b2a97896..66e3ba69 100644 --- a/extensions/s3/client/examples/rustfs_small_files_benchmark.rs +++ b/extensions/s3/client/examples/rustfs_small_files_benchmark.rs @@ -734,7 +734,7 @@ async fn main() -> BenchResult { let page = client.advance_fsck(&cursor, 10_000).await?; pages += 1; cursor = page.cursor; - if pages % 10 == 0 { + if pages.is_multiple_of(10) { println!( "FSCK_PROGRESS mode={fsck_mode} pages={pages} phase={:?} commits={} current_objects={} logical_versions={}", cursor.phase, @@ -793,7 +793,7 @@ async fn main() -> BenchResult { }; pages += 1; cursor = page.cursor; - if pages % 100 == 0 { + if pages.is_multiple_of(100) { println!( "GC_PROGRESS pages={pages} phase={:?} commits={} nodes={} logical_versions={} candidates={} deleted_versions={}", cursor.phase, diff --git a/extensions/s3/client/src/aws_object.rs b/extensions/s3/client/src/aws_object.rs index b305666b..31bb3a52 100644 --- a/extensions/s3/client/src/aws_object.rs +++ b/extensions/s3/client/src/aws_object.rs @@ -49,6 +49,26 @@ impl S3OperationMetrics { + self.delete_object + self.delete_objects } + + pub fn delta_since(self, earlier: Self) -> Self { + Self { + get_object: self.get_object.saturating_sub(earlier.get_object), + head_object: self.head_object.saturating_sub(earlier.head_object), + put_object: self.put_object.saturating_sub(earlier.put_object), + list_objects_v2: self.list_objects_v2.saturating_sub(earlier.list_objects_v2), + list_object_versions: self + .list_object_versions + .saturating_sub(earlier.list_object_versions), + delete_object: self.delete_object.saturating_sub(earlier.delete_object), + delete_objects: self.delete_objects.saturating_sub(earlier.delete_objects), + uploaded_body_bytes: self + .uploaded_body_bytes + .saturating_sub(earlier.uploaded_body_bytes), + downloaded_body_bytes: self + .downloaded_body_bytes + .saturating_sub(earlier.downloaded_body_bytes), + } + } } #[derive(Default)] diff --git a/extensions/s3/client/src/cache.rs b/extensions/s3/client/src/cache.rs index 73505537..9006be56 100644 --- a/extensions/s3/client/src/cache.rs +++ b/extensions/s3/client/src/cache.rs @@ -1,4 +1,8 @@ -use std::{path::PathBuf, sync::Arc}; +use std::{ + collections::{BTreeMap, VecDeque}, + path::PathBuf, + sync::{Arc, Mutex}, +}; use foyer::{ BlockEngineConfig, DeviceBuilder, FsDeviceBuilder, HybridCache, HybridCachePolicy, @@ -69,6 +73,15 @@ impl FoyerNodeCacheConfig { pub struct FoyerNodeCache { cache: HybridCache, Vec>, max_entry_size_bytes: usize, + pinned_capacity_bytes: usize, + pinned: Mutex, +} + +#[derive(Default)] +struct PinnedFoyerState { + entries: BTreeMap, Arc<[u8]>>, + order: VecDeque>, + bytes: usize, } impl FoyerNodeCache { @@ -118,6 +131,8 @@ impl FoyerNodeCache { Ok(Arc::new(Self { cache, max_entry_size_bytes, + pinned_capacity_bytes: config.memory_capacity_bytes / 4, + pinned: Mutex::new(PinnedFoyerState::default()), })) } @@ -142,11 +157,29 @@ impl NodeCache for FoyerNodeCache { value_len <= self.max_entry_size_bytes } + fn pinned_usage(&self) -> Option<(usize, usize)> { + let state = self + .pinned + .lock() + .unwrap_or_else(std::sync::PoisonError::into_inner); + Some((state.entries.len(), state.bytes)) + } + async fn get( &self, key: &NodeCacheKey, ) -> std::result::Result>, NodeCacheError> { let encoded = key.encode().to_vec(); + if let Some(value) = self + .pinned + .lock() + .unwrap_or_else(std::sync::PoisonError::into_inner) + .entries + .get(&encoded) + .cloned() + { + return Ok(Some(value.as_ref().to_vec())); + } self.cache .get(encoded.as_slice()) .await @@ -177,8 +210,55 @@ impl NodeCache for FoyerNodeCache { Ok(()) } + async fn pin( + &self, + key: NodeCacheKey, + value: Vec, + ) -> std::result::Result<(), NodeCacheError> { + if !self.admits(&key, value.len()) { + return Ok(()); + } + self.insert(key.clone(), value.clone()).await?; + if value.len() > self.pinned_capacity_bytes { + return Err(NodeCacheError::new( + "node exceeds the bounded Foyer pinned-memory tier", + )); + } + let encoded = key.encode().to_vec(); + let mut state = self + .pinned + .lock() + .unwrap_or_else(std::sync::PoisonError::into_inner); + if let Some(previous) = state.entries.remove(&encoded) { + state.bytes = state.bytes.saturating_sub(previous.len()); + } + state.order.retain(|candidate| candidate != &encoded); + state.bytes = state.bytes.saturating_add(value.len()); + state.entries.insert(encoded.clone(), Arc::from(value)); + state.order.push_back(encoded); + while state.bytes > self.pinned_capacity_bytes { + let Some(evicted) = state.order.pop_front() else { + break; + }; + if let Some(value) = state.entries.remove(&evicted) { + state.bytes = state.bytes.saturating_sub(value.len()); + } + } + Ok(()) + } + async fn remove(&self, key: &NodeCacheKey) -> std::result::Result<(), NodeCacheError> { let encoded = key.encode().to_vec(); + { + let mut state = self + .pinned + .lock() + .unwrap_or_else(std::sync::PoisonError::into_inner); + if let Some(value) = state.entries.remove(&encoded) { + state.bytes = state.bytes.saturating_sub(value.len()); + } + state.order.retain(|candidate| candidate != &encoded); + } // Foyer 0.22's delete tombstone is not recovered after every clean // reopen. Persist an adapter-level tombstone so a corrupt entry cannot // reappear after restart. The storage engine bounds these markers by @@ -354,4 +434,19 @@ mod tests { ); reopened_after_reinsert.close().await.unwrap(); } + + #[tokio::test] + async fn pinned_nodes_use_the_bounded_memory_tier() { + let directory = tempfile::tempdir().unwrap(); + let cache = FoyerNodeCache::open(config(directory.path().to_path_buf())) + .await + .unwrap(); + cache.pin(key(), vec![7; 1024]).await.unwrap(); + assert_eq!(cache.pinned_usage(), Some((1, 1024))); + assert_eq!(cache.get(&key()).await.unwrap(), Some(vec![7; 1024])); + cache.remove(&key()).await.unwrap(); + assert_eq!(cache.pinned_usage(), Some((0, 0))); + assert_eq!(cache.get(&key()).await.unwrap(), None); + cache.close().await.unwrap(); + } } diff --git a/extensions/s3/client/src/client.rs b/extensions/s3/client/src/client.rs index 46480258..2fb447e3 100644 --- a/extensions/s3/client/src/client.rs +++ b/extensions/s3/client/src/client.rs @@ -23,9 +23,9 @@ use prolly_s3_core::{ OperationIndexRebuildCursor, OperationIndexRebuildStep, ProviderAttestation, ProviderPerKeyVersionLimit, ProviderProfileId, PublicationJournalCursor, PublicationJournalPage, RefCatalogCursor, RefCatalogRepairPage, RefKind, RefMoveReceipt, - RepairCursor, RepairPage, Repository, RepositoryOptions, RestoreCursor, RestorePage, Result, - RetentionPin, RetentionPinPage, StagedMutation, Tag, TagCatalogPage, TraversalBudget, - TreeFormat, VersionSummary, + RepairCursor, RepairPage, Repository, RepositoryOptions, ResolvedSnapshot, RestoreCursor, + RestorePage, Result, RetentionPin, RetentionPinPage, StagedMutation, Tag, TagCatalogPage, + TraversalBudget, TreeFormat, VersionSummary, }; use sha2::{Digest as _, Sha256}; use tokio::{ @@ -35,8 +35,9 @@ use tokio::{ use crate::{ ensure_attestation_current, load_valid_attestation, qualify_and_store, - validate_provider_bucket, AttestationSigner, AwsS3ObjectPlane, ProviderIdentity, - ProviderQualificationOptions, S3OperationMetrics, + validate_provider_bucket, AttestationSigner, AwsS3ObjectPlane, ClientPerformanceSnapshot, + ClientStartupMetrics, ClientTelemetry, ClientTelemetryContext, ClientTelemetryInterval, + ProductionCacheProfile, ProviderIdentity, ProviderQualificationOptions, S3OperationMetrics, }; /// Application-facing repository client. @@ -52,8 +53,12 @@ pub struct Client { branch: String, checked_out: CheckedOutRef, provider_attestation: ProviderAttestation, + startup_metrics: ClientStartupMetrics, shard_authority_maintenance: Arc>>, _branch_index_maintenance: Arc>>, + _telemetry_maintenance: Arc>>, + #[cfg(feature = "foyer-cache")] + production_node_cache: Option>, } /// Durable handoff describing one complete provider object. @@ -96,6 +101,9 @@ pub struct ClientBuilder { qualification_options: Option, provider_per_key_version_limit: Option, background_index_maintenance: Option, + production_cache_profile: Option, + telemetry: Option>, + telemetry_interval: Option, } #[derive(Clone, Debug, PartialEq, Eq)] @@ -324,6 +332,44 @@ impl Client { self.repository.node_cache_snapshot() } + pub fn startup_metrics(&self) -> ClientStartupMetrics { + self.startup_metrics + } + + pub fn performance_snapshot(&self) -> ClientPerformanceSnapshot { + ClientPerformanceSnapshot { + cache: self.node_cache_snapshot(), + provider: self.s3_operation_metrics(), + } + } + + /// Stop this client's background maintenance, then flush and close the + /// internally owned production cache. Stop serving requests and drop + /// other client clones before calling this during graceful shutdown. + #[cfg(feature = "foyer-cache")] + pub async fn close_production_cache(&self) -> Result<()> { + let authority = self + .shard_authority_maintenance + .lock() + .unwrap_or_else(std::sync::PoisonError::into_inner) + .take(); + let indexes = self + ._branch_index_maintenance + .lock() + .unwrap_or_else(std::sync::PoisonError::into_inner) + .take(); + let telemetry = self + ._telemetry_maintenance + .lock() + .unwrap_or_else(std::sync::PoisonError::into_inner) + .take(); + drop((authority, indexes, telemetry)); + if let Some(cache) = &self.production_node_cache { + cache.close().await?; + } + Ok(()) + } + pub async fn prewarm_node_cache(&self, snapshot: CommitId) -> Result { self.ensure_provider_qualified()?; self.repository @@ -342,6 +388,13 @@ impl Client { .await } + pub async fn resolve_snapshot(&self, snapshot: CommitId) -> Result { + self.ensure_provider_qualified()?; + self.repository + .resolve_snapshot(&self.branch, snapshot) + .await + } + /// Select a branch, tag, or immutable commit, following Git-like ref /// precedence. Branch checkouts are writable; tag and commit checkouts are /// detached and reject mutation APIs with `InvalidRevision`. @@ -2405,6 +2458,22 @@ impl ClientBuilder { self } + /// Apply the production metadata-cache profile. The profile owns a + /// persistent Foyer directory, cardinality-derived bounds, upper-level + /// pinning, and bounded startup prewarming. + pub fn production_cache_profile(mut self, profile: ProductionCacheProfile) -> Self { + self.production_cache_profile = Some(profile); + self + } + + /// Export cache, provider, and cold-start deltas through an application + /// telemetry sink. The OpenTelemetry feature provides a ready-made sink. + pub fn telemetry(mut self, telemetry: Arc, interval: Duration) -> Self { + self.telemetry = Some(telemetry); + self.telemetry_interval = Some(interval); + self + } + pub fn max_cached_node_pack_bytes(mut self, bytes: usize) -> Self { self.max_cached_node_pack_bytes = Some(bytes); self @@ -2461,12 +2530,24 @@ impl ClientBuilder { } async fn finish(self, initialize: bool) -> Result { + let total_started = Instant::now(); if initialize && self.read_only { return Err(invalid( "repository initialization requires a writable client", )); } let background_index_maintenance = self.background_index_maintenance.unwrap_or(true); + let production_cache_profile = self.production_cache_profile.clone(); + let telemetry = self.telemetry.clone(); + let telemetry_interval = self.telemetry_interval.unwrap_or(Duration::from_secs(15)); + if telemetry.is_some() && telemetry_interval.is_zero() { + return Err(invalid("telemetry interval must be nonzero")); + } + if production_cache_profile.is_some() && !background_index_maintenance { + return Err(invalid( + "production cache profile requires branch-index maintenance", + )); + } let aws = self .aws_client .ok_or_else(|| invalid("aws_client is required"))?; @@ -2475,6 +2556,11 @@ impl ClientBuilder { .provider_identity .ok_or_else(|| invalid("provider_identity is required"))?; validate_provider_bucket(&identity, &bucket)?; + let telemetry_provider = match identity.bucket_class() { + prolly_s3_core::BucketClass::GeneralPurpose => "aws-s3", + prolly_s3_core::BucketClass::S3Compatible => "s3-compatible", + } + .to_string(); let signer = self .attestation_signer .ok_or_else(|| invalid("attestation_signer is required"))?; @@ -2525,7 +2611,7 @@ impl ClientBuilder { attestation.body.capabilities.validate_prolly_s3()?; let mut options = RepositoryOptions { - repository_prefix: prefix, + repository_prefix: prefix.clone(), read_only: self.read_only, provider_per_key_version_limit, ..RepositoryOptions::default() @@ -2542,6 +2628,24 @@ impl ClientBuilder { } if let Some(format) = self.state_tree_format { options.state_tree_format = format; + } else if initialize && production_cache_profile.is_some() { + options.state_tree_format = crate::production_metadata_tree_format(); + } else if !initialize { + options.state_tree_format = + Repository::::load_format(plane.clone(), &prefix) + .await? + .state_tree_format; + } + if let Some(profile) = &production_cache_profile { + if profile.directory.as_os_str().is_empty() + || !(1..=64).contains(&profile.startup_prewarm_levels) + || profile.startup_prewarm_timeout.is_zero() + { + return Err(invalid("production cache profile is invalid")); + } + options.max_cached_node_pack_bytes = profile.sizing.max_cached_node_pack_bytes; + options.max_cached_node_locations = profile.sizing.max_cached_node_locations; + options.max_cached_node_bytes = profile.sizing.memory_capacity_bytes; } if let Some(bytes) = self.max_cached_node_pack_bytes { options.max_cached_node_pack_bytes = bytes; @@ -2567,7 +2671,43 @@ impl ClientBuilder { if let Some(events) = self.operation_index_max_unindexed_events { options.operation_index_max_unindexed_events = events; } - options.node_cache = self.node_cache; + #[cfg(feature = "foyer-cache")] + let (node_cache, production_node_cache) = { + let mut node_cache = self.node_cache; + let mut production_node_cache = None; + if let Some(profile) = &production_cache_profile { + if node_cache.is_some() { + return Err(invalid( + "production cache profile cannot be combined with a custom node cache", + )); + } + let cache = crate::FoyerNodeCache::open(crate::FoyerNodeCacheConfig { + directory: profile.directory.clone(), + memory_capacity_bytes: profile.sizing.memory_capacity_bytes, + disk_capacity_bytes: profile.sizing.disk_capacity_bytes, + disk_block_size_bytes: crate::production::cache_block_size_for_tree( + profile.sizing.disk_block_size_bytes, + &options.state_tree_format, + ), + memory_shards: profile.sizing.memory_shards, + }) + .await?; + node_cache = Some(cache.clone()); + production_node_cache = Some(cache); + } + (node_cache, production_node_cache) + }; + #[cfg(not(feature = "foyer-cache"))] + let node_cache = { + if production_cache_profile.is_some() { + return Err(Error::new( + ErrorCode::UnsupportedParameter, + "production cache profile requires the foyer-cache feature", + )); + } + self.node_cache + }; + options.node_cache = node_cache; let branch = options.default_branch.clone(); let repository = if initialize { Repository::initialize(plane.clone(), options).await? @@ -2585,24 +2725,106 @@ impl ClientBuilder { } else { None }; - let client = Client { + let mut client = Client { repository, bucket, branch: branch.clone(), checked_out: CheckedOutRef::Branch(branch), provider_attestation: attestation, + startup_metrics: ClientStartupMetrics::default(), shard_authority_maintenance: Arc::new(Mutex::new(shard_authority_maintenance)), _branch_index_maintenance: Arc::new(Mutex::new(branch_index_maintenance)), + _telemetry_maintenance: Arc::new(Mutex::new(None)), + #[cfg(feature = "foyer-cache")] + production_node_cache, }; + let cache_before = client.node_cache_snapshot(); + let provider_before = S3OperationMetrics::default(); + let index_started = Instant::now(); if background_index_maintenance { client .wait_for_branch_indexes(Duration::from_secs(30)) .await?; } + client.startup_metrics.index_catchup_millis = elapsed_millis(index_started); + if let Some(profile) = &production_cache_profile { + let prewarm_started = Instant::now(); + let snapshot = client.head().await?; + match tokio::time::timeout( + profile.startup_prewarm_timeout, + client.prewarm_node_cache_levels(snapshot, profile.startup_prewarm_levels), + ) + .await + { + Ok(Ok(report)) => client.startup_metrics.prewarm_report = Some(report), + Ok(Err(error)) if profile.require_successful_prewarm => return Err(error), + Ok(Err(_)) => client.startup_metrics.prewarm_failed = true, + Err(_) if profile.require_successful_prewarm => { + return Err(Error::new( + ErrorCode::Timeout, + "production metadata-cache prewarm exceeded its startup timeout", + )) + } + Err(_) => client.startup_metrics.prewarm_timed_out = true, + } + client.startup_metrics.prewarm_millis = elapsed_millis(prewarm_started); + } + client.startup_metrics.total_open_millis = elapsed_millis(total_started); + client.startup_metrics.cache_activity = + client.node_cache_snapshot().delta_since(cache_before); + client.startup_metrics.provider_activity = + client.s3_operation_metrics().delta_since(provider_before); + if let Some(telemetry) = telemetry { + let context = ClientTelemetryContext { + repository_id: client.repository_id().to_string(), + provider: telemetry_provider, + expected_objects: production_cache_profile + .as_ref() + .map(|profile| profile.sizing.expected_objects), + }; + telemetry.record_startup(&context, client.startup_metrics); + let initial_cache = client.repository.node_cache_snapshot(); + let initial_provider = client.repository.plane().metrics(); + let repository = Arc::downgrade(&client.repository); + let task = tokio::spawn(async move { + let mut ticker = tokio::time::interval(telemetry_interval); + ticker.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Skip); + let mut previous_cache = initial_cache; + let mut previous_provider = initial_provider; + // The startup report already covers the initial interval. + ticker.tick().await; + loop { + ticker.tick().await; + let Some(repository) = repository.upgrade() else { + break; + }; + let current_cache = repository.node_cache_snapshot(); + let current_provider = repository.plane().metrics(); + telemetry.record_interval( + &context, + ClientTelemetryInterval { + cache: current_cache.delta_since(previous_cache), + provider: current_provider.delta_since(previous_provider), + }, + ); + previous_cache = current_cache; + previous_provider = current_provider; + } + }); + *client + ._telemetry_maintenance + .lock() + .unwrap_or_else(std::sync::PoisonError::into_inner) = + Some(crate::telemetry::ClientTelemetryMaintenance::new(task)); + } Ok(client) } } +fn elapsed_millis(started: Instant) -> u64 { + u64::try_from(started.elapsed().as_millis()).unwrap_or(u64::MAX) +} + fn invalid(message: impl Into) -> Error { Error::new(ErrorCode::InvalidRequest, message) } diff --git a/extensions/s3/client/src/lib.rs b/extensions/s3/client/src/lib.rs index a33e91ce..6a2d920e 100644 --- a/extensions/s3/client/src/lib.rs +++ b/extensions/s3/client/src/lib.rs @@ -4,14 +4,25 @@ mod aws_object; #[cfg(feature = "foyer-cache")] mod cache; mod client; +mod production; mod provider; +mod telemetry; mod wire_metrics; pub use aws_object::{AwsS3ObjectPlane, S3OperationMetrics}; #[cfg(feature = "foyer-cache")] pub use cache::{FoyerNodeCache, FoyerNodeCacheConfig}; pub use client::*; +pub use production::{ + production_metadata_tree_format, CacheSizingRecommendation, ClientStartupMetrics, + ProductionCacheProfile, ProviderDeployment, SupportStatus, SupportedEnvelope, +}; pub use prolly_s3_core as core; pub use prolly_s3_core::{Error, ErrorCode, Result}; pub use provider::*; +#[cfg(feature = "opentelemetry")] +pub use telemetry::OpenTelemetryClientMetrics; +pub use telemetry::{ + ClientPerformanceSnapshot, ClientTelemetry, ClientTelemetryContext, ClientTelemetryInterval, +}; pub use wire_metrics::*; diff --git a/extensions/s3/client/src/production.rs b/extensions/s3/client/src/production.rs new file mode 100644 index 00000000..73ff8a82 --- /dev/null +++ b/extensions/s3/client/src/production.rs @@ -0,0 +1,290 @@ +use std::{path::PathBuf, time::Duration}; + +use prolly_s3_core::{ + BoundaryInput, BoundaryRule, ChunkMeasure, ChunkingSpec, HashAlgorithm, NodeCachePrewarmReport, + NodeCacheSnapshot, NodeLayoutSpec, TreeFormat, +}; + +use crate::S3OperationMetrics; + +/// Cardinality-derived cache capacities for one repository process. +/// +/// These are starting points, not provider-independent SLO guarantees. Use +/// the exported cache and byte-amplification metrics to tune them against the +/// actual key distribution and read mix. +#[derive(Clone, Copy, Debug, PartialEq, Eq)] +pub struct CacheSizingRecommendation { + pub expected_objects: u64, + pub memory_capacity_bytes: usize, + pub disk_capacity_bytes: usize, + pub disk_block_size_bytes: usize, + pub memory_shards: usize, + pub max_cached_node_pack_bytes: usize, + pub max_cached_node_locations: usize, + pub startup_prewarm_levels: usize, +} + +impl CacheSizingRecommendation { + pub fn for_object_count(expected_objects: u64) -> Self { + const MIB: usize = 1024 * 1024; + const GIB: usize = 1024 * MIB; + let expected_objects = expected_objects.max(1); + match expected_objects { + 1..=100_000 => Self { + expected_objects, + memory_capacity_bytes: 128 * MIB, + disk_capacity_bytes: 2 * GIB, + disk_block_size_bytes: MIB, + memory_shards: 4, + max_cached_node_pack_bytes: 64 * MIB, + max_cached_node_locations: 131_072, + startup_prewarm_levels: 2, + }, + 100_001..=500_000 => Self { + expected_objects, + memory_capacity_bytes: 256 * MIB, + disk_capacity_bytes: 8 * GIB, + disk_block_size_bytes: MIB, + memory_shards: 8, + max_cached_node_pack_bytes: 128 * MIB, + max_cached_node_locations: 262_144, + startup_prewarm_levels: 3, + }, + 500_001..=1_000_000 => Self { + expected_objects, + memory_capacity_bytes: 512 * MIB, + disk_capacity_bytes: 16 * GIB, + disk_block_size_bytes: MIB, + memory_shards: 16, + max_cached_node_pack_bytes: 256 * MIB, + max_cached_node_locations: 524_288, + startup_prewarm_levels: 3, + }, + _ => { + let millions = expected_objects.saturating_add(999_999) / 1_000_000; + let scale = usize::try_from(millions).unwrap_or(usize::MAX); + Self { + expected_objects, + memory_capacity_bytes: (512 * MIB) + .saturating_mul(scale) + .clamp(512 * MIB, 4 * GIB), + disk_capacity_bytes: (16 * GIB) + .saturating_mul(scale) + .clamp(16 * GIB, 128 * GIB), + disk_block_size_bytes: MIB, + memory_shards: 16, + max_cached_node_pack_bytes: (256 * MIB) + .saturating_mul(scale) + .clamp(256 * MIB, 2 * GIB), + max_cached_node_locations: 524_288usize + .saturating_mul(scale) + .clamp(524_288, 4_194_304), + startup_prewarm_levels: 4, + } + } + } + } +} + +/// Persistent-cache and bounded cold-start policy for a production client. +/// +/// Applying this profile always opens a hybrid memory/disk node cache. It is +/// intentionally explicit because the application must choose a directory +/// with one filesystem owner and an appropriate storage budget. +#[derive(Clone, Debug, PartialEq, Eq)] +pub struct ProductionCacheProfile { + pub directory: PathBuf, + pub sizing: CacheSizingRecommendation, + pub startup_prewarm_levels: usize, + pub startup_prewarm_timeout: Duration, + pub require_successful_prewarm: bool, +} + +impl ProductionCacheProfile { + pub fn new(directory: impl Into, expected_objects: u64) -> Self { + let sizing = CacheSizingRecommendation::for_object_count(expected_objects); + Self { + directory: directory.into(), + startup_prewarm_levels: sizing.startup_prewarm_levels, + sizing, + startup_prewarm_timeout: Duration::from_secs(30), + require_successful_prewarm: true, + } + } + + pub fn startup_prewarm(mut self, levels: usize, timeout: Duration) -> Self { + self.startup_prewarm_levels = levels; + self.startup_prewarm_timeout = timeout; + self + } + + pub fn require_successful_prewarm(mut self, required: bool) -> Self { + self.require_successful_prewarm = required; + self + } +} + +/// Bounded startup evidence exported by each client opened with a production +/// cache profile. +#[derive(Clone, Copy, Debug, Default, PartialEq, Eq)] +pub struct ClientStartupMetrics { + pub total_open_millis: u64, + pub index_catchup_millis: u64, + pub prewarm_millis: u64, + pub prewarm_timed_out: bool, + pub prewarm_failed: bool, + pub prewarm_report: Option, + pub cache_activity: NodeCacheSnapshot, + pub provider_activity: S3OperationMetrics, +} + +#[derive(Clone, Copy, Debug, PartialEq, Eq)] +pub enum ProviderDeployment { + AwsGeneralPurpose, + RustFs, + OtherS3Compatible, +} + +#[derive(Clone, Copy, Debug, PartialEq, Eq)] +pub enum SupportStatus { + /// Suitable for a controlled pilot when the listed release gates pass. + ControlledPilot, + /// The architecture supports the size, but provider-specific evidence is + /// required before production promotion. + QualificationRequired, + /// Current measured performance does not support a production claim. + PerformanceGateFailed, +} + +#[derive(Clone, Debug, PartialEq, Eq)] +pub struct SupportedEnvelope { + pub provider: ProviderDeployment, + pub object_count: u64, + pub status: SupportStatus, + pub persistent_cache_required: bool, + pub aws_qualification_required: bool, + pub maintenance_scale_gate_required: bool, + pub required_release_gates: Vec<&'static str>, +} + +impl SupportedEnvelope { + /// Return the explicit support posture represented by the published 2026 + /// qualification evidence. This deliberately does not turn architectural + /// capacity into an unmeasured production claim. + pub fn for_deployment(provider: ProviderDeployment, object_count: u64) -> Self { + let object_count = object_count.max(1); + let aws_qualification_required = provider == ProviderDeployment::AwsGeneralPurpose; + let maintenance_scale_gate_required = object_count > 100_000; + let status = if object_count >= 1_000_000 { + SupportStatus::PerformanceGateFailed + } else if aws_qualification_required || object_count > 100_000 { + SupportStatus::QualificationRequired + } else { + SupportStatus::ControlledPilot + }; + let mut required_release_gates = vec![ + "provider capability attestation", + "cold and warm read/list SLOs", + "authority takeover and restart fault injection", + "fsck and GC recovery drill", + ]; + if aws_qualification_required { + required_release_gates.push("AWS lifecycle, replication, throttling, and cost matrix"); + } + if maintenance_scale_gate_required { + required_release_gates.push("cardinality-matched fsck and GC timing"); + } + Self { + provider, + object_count, + status, + persistent_cache_required: object_count >= 100_000, + aws_qualification_required, + maintenance_scale_gate_required, + required_release_gates, + } + } +} + +/// Metadata-tree geometry for newly initialized production repositories. +/// +/// Logical payloads remain one complete immutable object. This only bounds +/// canonical Prolly metadata nodes so exact node-pack range reads remain +/// sub-megabyte even for wide logical records. +pub fn production_metadata_tree_format() -> TreeFormat { + TreeFormat { + chunking: ChunkingSpec { + measure: ChunkMeasure::EncodedBytes, + input: BoundaryInput::Key, + hash: HashAlgorithm::XxHash64, + rule: BoundaryRule::HashThreshold { factor: 32 * 1024 }, + min: 8 * 1024, + target: 32 * 1024, + max: 64 * 1024, + hash_seed: 0x243f_6a88_85a3_08d3, + level_salt: true, + hard_max_node_bytes: 256 * 1024, + }, + node_layout: NodeLayoutSpec::PrefixCompressed, + value_encoding: prolly_s3_core::Encoding::Raw, + } +} + +#[cfg(any(feature = "foyer-cache", test))] +pub(crate) fn cache_block_size_for_tree(recommended: usize, format: &TreeFormat) -> usize { + let hard_max = usize::try_from(format.chunking.hard_max_node_bytes).unwrap_or(usize::MAX); + // Leave room for Foyer's aligned block index and entry/key envelopes. + hard_max + .saturating_add(8 * 1024) + .checked_next_power_of_two() + .unwrap_or(usize::MAX) + .max(recommended) +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn recommendations_scale_monotonically_through_one_million_objects() { + let small = CacheSizingRecommendation::for_object_count(100_000); + let medium = CacheSizingRecommendation::for_object_count(500_000); + let large = CacheSizingRecommendation::for_object_count(1_000_000); + assert!(small.memory_capacity_bytes < medium.memory_capacity_bytes); + assert!(medium.memory_capacity_bytes < large.memory_capacity_bytes); + assert!(small.disk_capacity_bytes < medium.disk_capacity_bytes); + assert!(medium.disk_capacity_bytes < large.disk_capacity_bytes); + assert!(small.max_cached_node_locations < large.max_cached_node_locations); + } + + #[test] + fn production_metadata_nodes_are_bounded_for_range_reads() { + let format = production_metadata_tree_format(); + format.validate().unwrap(); + assert_eq!(format.chunking.measure, ChunkMeasure::EncodedBytes); + assert_eq!(format.chunking.hard_max_node_bytes, 256 * 1024); + assert!(format.chunking.max <= format.chunking.hard_max_node_bytes); + } + + #[test] + fn legacy_wide_tree_formats_receive_a_large_enough_foyer_block() { + let format = TreeFormat::default(); + assert_eq!( + cache_block_size_for_tree(1024 * 1024, &format), + 32 * 1024 * 1024 + ); + } + + #[test] + fn support_envelope_never_promotes_unqualified_aws_or_million_plus_scale() { + assert_eq!( + SupportedEnvelope::for_deployment(ProviderDeployment::AwsGeneralPurpose, 100_000) + .status, + SupportStatus::QualificationRequired + ); + assert_eq!( + SupportedEnvelope::for_deployment(ProviderDeployment::RustFs, 1_000_001).status, + SupportStatus::PerformanceGateFailed + ); + } +} diff --git a/extensions/s3/client/src/telemetry.rs b/extensions/s3/client/src/telemetry.rs new file mode 100644 index 00000000..daa90c67 --- /dev/null +++ b/extensions/s3/client/src/telemetry.rs @@ -0,0 +1,278 @@ +#[cfg(feature = "opentelemetry")] +use std::sync::Arc; + +use prolly_s3_core::NodeCacheSnapshot; + +use crate::{ClientStartupMetrics, S3OperationMetrics}; + +#[derive(Clone, Debug, PartialEq, Eq)] +pub struct ClientTelemetryContext { + pub repository_id: String, + pub provider: String, + pub expected_objects: Option, +} + +#[derive(Clone, Copy, Debug, Default, PartialEq, Eq)] +pub struct ClientTelemetryInterval { + pub cache: NodeCacheSnapshot, + pub provider: S3OperationMetrics, +} + +#[derive(Clone, Copy, Debug, Default, PartialEq, Eq)] +pub struct ClientPerformanceSnapshot { + pub cache: NodeCacheSnapshot, + pub provider: S3OperationMetrics, +} + +impl ClientPerformanceSnapshot { + pub fn delta_since(self, earlier: Self) -> Self { + Self { + cache: self.cache.delta_since(earlier.cache), + provider: self.provider.delta_since(earlier.provider), + } + } + + /// Provider response bytes per canonical metadata-node byte returned. + /// Measure around metadata-only operations (list/diff/merge planning) so + /// logical payload downloads do not enter the numerator. + pub fn metadata_download_amplification(self) -> f64 { + if self.cache.requested_bytes == 0 { + 0.0 + } else { + self.provider.downloaded_body_bytes as f64 / self.cache.requested_bytes as f64 + } + } +} + +/// Application-owned telemetry target. Implementations must return quickly; +/// collection runs on a lightweight periodic maintenance task. +pub trait ClientTelemetry: Send + Sync + 'static { + fn record_startup(&self, context: &ClientTelemetryContext, startup: ClientStartupMetrics); + fn record_interval(&self, context: &ClientTelemetryContext, interval: ClientTelemetryInterval); +} + +pub(crate) struct ClientTelemetryMaintenance { + task: tokio::task::JoinHandle<()>, +} + +impl ClientTelemetryMaintenance { + pub(crate) fn new(task: tokio::task::JoinHandle<()>) -> Self { + Self { task } + } +} + +impl Drop for ClientTelemetryMaintenance { + fn drop(&mut self) { + self.task.abort(); + } +} + +#[cfg(feature = "opentelemetry")] +pub struct OpenTelemetryClientMetrics { + cache_lookups: opentelemetry::metrics::Counter, + metadata_bytes: opentelemetry::metrics::Counter, + cache_events: opentelemetry::metrics::Counter, + provider_operations: opentelemetry::metrics::Counter, + provider_bytes: opentelemetry::metrics::Counter, + startup_duration: opentelemetry::metrics::Histogram, +} + +#[cfg(feature = "opentelemetry")] +impl OpenTelemetryClientMetrics { + pub fn new(meter: opentelemetry::metrics::Meter) -> Arc { + Arc::new(Self { + cache_lookups: meter + .u64_counter("prolly.s3.metadata.cache.lookups") + .build(), + metadata_bytes: meter.u64_counter("prolly.s3.metadata.bytes").build(), + cache_events: meter.u64_counter("prolly.s3.metadata.cache.events").build(), + provider_operations: meter.u64_counter("prolly.s3.provider.operations").build(), + provider_bytes: meter.u64_counter("prolly.s3.provider.bytes").build(), + startup_duration: meter + .u64_histogram("prolly.s3.client.startup.duration") + .build(), + }) + } + + fn attributes( + context: &ClientTelemetryContext, + name: &'static str, + value: &'static str, + ) -> Vec { + let mut attributes = vec![ + opentelemetry::KeyValue::new("prolly.repository.id", context.repository_id.clone()), + opentelemetry::KeyValue::new("prolly.provider", context.provider.clone()), + opentelemetry::KeyValue::new(name, value), + ]; + if let Some(objects) = context.expected_objects { + attributes.push(opentelemetry::KeyValue::new( + "prolly.repository.expected_objects", + i64::try_from(objects).unwrap_or(i64::MAX), + )); + } + attributes + } + + fn add_event(&self, context: &ClientTelemetryContext, event: &'static str, value: u64) { + if value > 0 { + self.cache_events.add( + value, + &Self::attributes(context, "prolly.cache.event", event), + ); + } + } +} + +#[cfg(feature = "opentelemetry")] +impl ClientTelemetry for OpenTelemetryClientMetrics { + fn record_startup(&self, context: &ClientTelemetryContext, startup: ClientStartupMetrics) { + for (phase, duration) in [ + ("total", startup.total_open_millis), + ("index_catchup", startup.index_catchup_millis), + ("prewarm", startup.prewarm_millis), + ] { + self.startup_duration.record( + duration, + &Self::attributes(context, "prolly.startup.phase", phase), + ); + } + self.record_interval( + context, + ClientTelemetryInterval { + cache: startup.cache_activity, + provider: startup.provider_activity, + }, + ); + self.add_event( + context, + "prewarm_timeout", + u64::from(startup.prewarm_timed_out), + ); + self.add_event( + context, + "prewarm_failure", + u64::from(startup.prewarm_failed), + ); + } + + fn record_interval(&self, context: &ClientTelemetryContext, interval: ClientTelemetryInterval) { + for (outcome, value) in [ + ("hit", interval.cache.hits), + ("miss", interval.cache.misses), + ] { + if value > 0 { + self.cache_lookups.add( + value, + &Self::attributes(context, "prolly.cache.outcome", outcome), + ); + } + } + for (kind, value) in [ + ("requested", interval.cache.requested_bytes), + ("provider_fetched", interval.cache.fetched_bytes), + ("cache_avoided", interval.cache.avoided_bytes), + ("pinned", interval.cache.pinned_bytes), + ] { + if value > 0 { + self.metadata_bytes.add( + value, + &Self::attributes(context, "prolly.metadata.bytes.kind", kind), + ); + } + } + for (event, value) in [ + ("admission_rejection", interval.cache.admission_rejections), + ("cache_error", interval.cache.errors), + ("cache_corruption", interval.cache.corruptions), + ("coalesced_wait", interval.cache.coalesced_waits), + ("prefetch_batch", interval.cache.prefetch_batches), + ("prefetched_node", interval.cache.prefetched_nodes), + ("pinned_node", interval.cache.pinned_nodes), + ] { + self.add_event(context, event, value); + } + for (operation, value) in [ + ("get_object", interval.provider.get_object), + ("head_object", interval.provider.head_object), + ("put_object", interval.provider.put_object), + ("list_objects_v2", interval.provider.list_objects_v2), + ( + "list_object_versions", + interval.provider.list_object_versions, + ), + ("delete_object", interval.provider.delete_object), + ("delete_objects", interval.provider.delete_objects), + ] { + if value > 0 { + self.provider_operations.add( + value, + &Self::attributes(context, "prolly.provider.operation", operation), + ); + } + } + for (direction, value) in [ + ("uploaded", interval.provider.uploaded_body_bytes), + ("downloaded", interval.provider.downloaded_body_bytes), + ] { + if value > 0 { + self.provider_bytes.add( + value, + &Self::attributes(context, "prolly.provider.direction", direction), + ); + } + } + } +} + +#[cfg(all(test, feature = "opentelemetry"))] +mod tests { + use super::*; + + #[test] + fn opentelemetry_sink_accepts_startup_and_interval_metrics() { + let sink = OpenTelemetryClientMetrics::new(opentelemetry::global::meter("test-prolly-s3")); + let context = ClientTelemetryContext { + repository_id: "pr_test".to_string(), + provider: "s3-compatible".to_string(), + expected_objects: Some(100_000), + }; + sink.record_startup( + &context, + ClientStartupMetrics { + total_open_millis: 20, + index_catchup_millis: 5, + prewarm_millis: 10, + cache_activity: NodeCacheSnapshot { + hits: 2, + misses: 1, + requested_bytes: 300, + fetched_bytes: 100, + avoided_bytes: 200, + ..NodeCacheSnapshot::default() + }, + provider_activity: S3OperationMetrics { + get_object: 1, + downloaded_body_bytes: 100, + ..S3OperationMetrics::default() + }, + ..ClientStartupMetrics::default() + }, + ); + sink.record_interval( + &context, + ClientTelemetryInterval { + cache: NodeCacheSnapshot { + admission_rejections: 1, + prefetch_batches: 1, + prefetched_nodes: 4, + ..NodeCacheSnapshot::default() + }, + provider: S3OperationMetrics { + get_object: 2, + downloaded_body_bytes: 128, + ..S3OperationMetrics::default() + }, + }, + ); + } +} diff --git a/extensions/s3/client/tests/rustfs_repository.rs b/extensions/s3/client/tests/rustfs_repository.rs index fd385753..d1a2e6e4 100644 --- a/extensions/s3/client/tests/rustfs_repository.rs +++ b/extensions/s3/client/tests/rustfs_repository.rs @@ -101,6 +101,72 @@ async fn rustfs_client() -> (aws_sdk_s3::Client, String) { (client, bucket) } +#[cfg(feature = "foyer-cache")] +#[tokio::test(flavor = "multi_thread", worker_threads = 4)] +async fn rustfs_production_cache_profile_reopens_prewarms_and_closes_cleanly() { + if !rustfs_enabled() { + eprintln!("set PROLLY_S3_RUSTFS=1 to run RustFS integration tests"); + return; + } + + let (aws, bucket) = rustfs_client().await; + let repository_prefix = unique_name("production-cache-profile"); + let cache_directory = tempfile::tempdir().unwrap(); + let mut profile = prolly_s3_client::ProductionCacheProfile::new(cache_directory.path(), 10_000) + .startup_prewarm(2, Duration::from_secs(10)); + profile.sizing.memory_capacity_bytes = 16 * 1024 * 1024; + profile.sizing.disk_capacity_bytes = 128 * 1024 * 1024; + profile.sizing.max_cached_node_pack_bytes = 16 * 1024 * 1024; + profile.sizing.max_cached_node_locations = 4_096; + let writer = Client::builder() + .aws_client(aws.clone()) + .bucket(&bucket) + .repository_prefix(&repository_prefix) + .writer("rustfs-production-cache-writer") + .provider_identity(provider_identity()) + .attestation_signer(attestation_signer()) + .provider_per_key_version_limit(ProviderPerKeyVersionLimit::Finite(10_000)) + .production_cache_profile(profile.clone()) + .initialize() + .await + .unwrap(); + writer + .put_object("docs/cached.txt", b"persistent metadata".to_vec()) + .await + .unwrap(); + writer.advance_branch_indexes().await.unwrap(); + writer.close_production_cache().await.unwrap(); + drop(writer); + + let reader = Client::builder() + .aws_client(aws) + .bucket(&bucket) + .repository_prefix(&repository_prefix) + .read_only(true) + .provider_identity(provider_identity()) + .attestation_signer(attestation_signer()) + .provider_per_key_version_limit(ProviderPerKeyVersionLimit::Finite(10_000)) + .production_cache_profile(profile) + .open() + .await + .unwrap(); + let startup = reader.startup_metrics(); + let prewarm = startup.prewarm_report.expect("production prewarm report"); + assert!(prewarm.object_nodes > 0); + assert!(prewarm.version_nodes > 0); + assert!(prewarm.after.pinned_nodes > 0); + assert_eq!( + reader + .get_object("docs/cached.txt") + .await + .unwrap() + .unwrap() + .bytes, + b"persistent metadata" + ); + reader.close_production_cache().await.unwrap(); +} + #[tokio::test(flavor = "multi_thread", worker_threads = 4)] async fn rustfs_client_uses_immutable_payloads_and_fences_takeover() { if !rustfs_enabled() { diff --git a/extensions/s3/core/src/cache.rs b/extensions/s3/core/src/cache.rs index dfe9a6b8..d88e63c2 100644 --- a/extensions/s3/core/src/cache.rs +++ b/extensions/s3/core/src/cache.rs @@ -1,5 +1,5 @@ use std::{ - collections::{BTreeMap, VecDeque}, + collections::{BTreeMap, BTreeSet, VecDeque}, sync::{Arc, Mutex}, }; @@ -57,6 +57,13 @@ pub trait NodeCache: Send + Sync + 'static { true } + /// Current node/byte usage of a distinct pinned tier, when the cache can + /// report it exactly. `None` asks the repository to use bounded fallback + /// accounting for successful pin requests. + fn pinned_usage(&self) -> Option<(usize, usize)> { + None + } + async fn get(&self, key: &NodeCacheKey) -> std::result::Result>, NodeCacheError>; @@ -66,12 +73,27 @@ pub trait NodeCache: Send + Sync + 'static { value: Vec, ) -> std::result::Result<(), NodeCacheError>; + /// Retain a verified upper-level node in the fastest available cache tier. + /// + /// Pinning is advisory and byte bounded. Implementations that do not have + /// a distinct pinned tier may treat this as an ordinary insertion. Cache + /// loss or pin rejection can affect latency but never repository + /// correctness. + async fn pin( + &self, + key: NodeCacheKey, + value: Vec, + ) -> std::result::Result<(), NodeCacheError> { + self.insert(key, value).await + } + async fn remove(&self, key: &NodeCacheKey) -> std::result::Result<(), NodeCacheError>; } struct MemoryNodeCacheState { entries: BTreeMap>, order: VecDeque, + pinned: BTreeSet, bytes: usize, } @@ -90,6 +112,7 @@ impl MemoryNodeCache { state: Mutex::new(MemoryNodeCacheState { entries: BTreeMap::new(), order: VecDeque::new(), + pinned: BTreeSet::new(), bytes: 0, }), } @@ -113,6 +136,64 @@ impl MemoryNodeCache { pub fn is_empty(&self) -> bool { self.len() == 0 } + + pub fn pinned_len(&self) -> usize { + self.state + .lock() + .unwrap_or_else(std::sync::PoisonError::into_inner) + .pinned + .len() + } + + fn insert_locked( + &self, + state: &mut MemoryNodeCacheState, + key: NodeCacheKey, + value: Vec, + pin: bool, + ) -> bool { + if !self.admits(&key, value.len()) { + return false; + } + let was_pinned = state.pinned.contains(&key); + if let Some(previous) = state.entries.remove(&key) { + state.bytes = state.bytes.saturating_sub(previous.len()); + } + state.order.retain(|candidate| candidate != &key); + state.bytes = state.bytes.saturating_add(value.len()); + state.entries.insert(key.clone(), Arc::from(value)); + state.order.push_back(key.clone()); + if pin || was_pinned { + state.pinned.insert(key.clone()); + } + + let mut inspected = 0usize; + while state.bytes > self.max_bytes && inspected < state.order.len() { + let Some(candidate) = state.order.pop_front() else { + break; + }; + if state.pinned.contains(&candidate) { + state.order.push_back(candidate); + inspected = inspected.saturating_add(1); + continue; + } + if let Some(removed) = state.entries.remove(&candidate) { + state.bytes = state.bytes.saturating_sub(removed.len()); + } + inspected = 0; + } + + // A pin set larger than the configured capacity is rejected rather + // than allowing an advisory tier to become unbounded. + if state.bytes > self.max_bytes { + state.pinned.remove(&key); + if let Some(removed) = state.entries.remove(&key) { + state.bytes = state.bytes.saturating_sub(removed.len()); + } + state.order.retain(|candidate| candidate != &key); + } + state.entries.contains_key(&key) && (!pin || state.pinned.contains(&key)) + } } #[async_trait::async_trait] @@ -121,6 +202,22 @@ impl NodeCache for MemoryNodeCache { self.max_bytes > 0 && value_len <= self.max_bytes } + fn pinned_usage(&self) -> Option<(usize, usize)> { + let state = self + .state + .lock() + .unwrap_or_else(std::sync::PoisonError::into_inner); + Some(( + state.pinned.len(), + state + .pinned + .iter() + .filter_map(|key| state.entries.get(key)) + .map(|value| value.len()) + .sum(), + )) + } + async fn get( &self, key: &NodeCacheKey, @@ -149,24 +246,28 @@ impl NodeCache for MemoryNodeCache { .state .lock() .unwrap_or_else(std::sync::PoisonError::into_inner); - if let Some(previous) = state.entries.remove(&key) { - state.bytes = state.bytes.saturating_sub(previous.len()); - } - state.order.retain(|candidate| candidate != &key); - state.bytes = state.bytes.saturating_add(value.len()); - state.entries.insert(key.clone(), Arc::from(value)); - state.order.push_back(key); - while state.bytes > self.max_bytes { - let Some(evicted) = state.order.pop_front() else { - break; - }; - if let Some(value) = state.entries.remove(&evicted) { - state.bytes = state.bytes.saturating_sub(value.len()); - } - } + self.insert_locked(&mut state, key, value, false); Ok(()) } + async fn pin( + &self, + key: NodeCacheKey, + value: Vec, + ) -> std::result::Result<(), NodeCacheError> { + let mut state = self + .state + .lock() + .unwrap_or_else(std::sync::PoisonError::into_inner); + if self.insert_locked(&mut state, key, value, true) { + Ok(()) + } else { + Err(NodeCacheError::new( + "pinned memory-cache capacity is exhausted", + )) + } + } + async fn remove(&self, key: &NodeCacheKey) -> std::result::Result<(), NodeCacheError> { let mut state = self .state @@ -175,6 +276,7 @@ impl NodeCache for MemoryNodeCache { if let Some(value) = state.entries.remove(key) { state.bytes = state.bytes.saturating_sub(value.len()); } + state.pinned.remove(key); state.order.retain(|candidate| candidate != key); Ok(()) } @@ -212,6 +314,21 @@ mod tests { assert!(cache.is_empty()); } + #[tokio::test] + async fn pinned_nodes_survive_lru_pressure_within_the_byte_bound() { + let cache = MemoryNodeCache::new(6); + cache.pin(key(1), vec![1; 3]).await.unwrap(); + cache.insert(key(2), vec![2; 3]).await.unwrap(); + cache.insert(key(3), vec![3; 3]).await.unwrap(); + + assert_eq!(cache.get(&key(1)).await.unwrap(), Some(vec![1; 3])); + assert_eq!(cache.get(&key(2)).await.unwrap(), None); + assert_eq!(cache.get(&key(3)).await.unwrap(), Some(vec![3; 3])); + assert_eq!(cache.pinned_len(), 1); + assert_eq!(cache.pinned_usage(), Some((1, 3))); + assert_eq!(cache.resident_bytes(), 6); + } + #[test] fn encoded_key_is_namespaced_and_fixed_width() { let first = key(1).encode(); diff --git a/extensions/s3/core/src/journal_indexes.rs b/extensions/s3/core/src/journal_indexes.rs index 327320b6..6b62d71b 100644 --- a/extensions/s3/core/src/journal_indexes.rs +++ b/extensions/s3/core/src/journal_indexes.rs @@ -1082,6 +1082,10 @@ impl JournalDerivedIndexes

{ generation: commit.generation, parents: commit.parents, first_parent_jumps: jumps, + snapshot: Some(crate::JournalSnapshotMetadata { + state: commit.state, + delta: commit.delta, + }), }) } diff --git a/extensions/s3/core/src/lib.rs b/extensions/s3/core/src/lib.rs index 03b74b80..ec9d4df4 100644 --- a/extensions/s3/core/src/lib.rs +++ b/extensions/s3/core/src/lib.rs @@ -57,8 +57,8 @@ pub use operation_index::{ }; pub use payload::ImmutablePayloadStore; pub use prolly::{ - BoundaryInput, BoundaryRule, ChunkMeasure, ChunkingSpec, Cid, HashAlgorithm, NodeLayoutSpec, - TreeFormat, + BoundaryInput, BoundaryRule, ChunkMeasure, ChunkingSpec, Cid, Encoding, HashAlgorithm, + NodeLayoutSpec, TreeFormat, }; pub use publication::{ AppliedBranchBarrier, CommitPublication, LoadedRef, PublicationJournalCursor, @@ -76,8 +76,8 @@ pub use repository::{ FsckCleanupPhase, FsckCursor, FsckPage, FsckPhase, FsckReport, HistoryCursor, ListObjectsPage, NodeCachePrewarmReport, ObjectData, ObjectDiff, ObjectDiffCursor, ObjectDiffPage, ObjectRangeData, ObjectSummary, RefCatalogRepairPage, RefMoveReceipt, RepairCursor, RepairPage, - RepairPhase, RepairReport, Repository, RepositoryOptions, RestoreCursor, RestorePage, - RetentionPin, RetentionPinPage, ShardAuthorityMaintenance, Tag, TagCatalogPage, + RepairPhase, RepairReport, Repository, RepositoryOptions, ResolvedSnapshot, RestoreCursor, + RestorePage, RetentionPin, RetentionPinPage, ShardAuthorityMaintenance, Tag, TagCatalogPage, TraversalBudget, VersionSummary, }; pub use runtime::*; diff --git a/extensions/s3/core/src/model.rs b/extensions/s3/core/src/model.rs index 5ed0fc3e..d44d8d6f 100644 --- a/extensions/s3/core/src/model.rs +++ b/extensions/s3/core/src/model.rs @@ -674,6 +674,17 @@ pub struct JournalCommitGraphEntry { pub generation: CommitGeneration, pub parents: Vec, pub first_parent_jumps: Vec, + /// Snapshot roots and exact logical delta resolved while the publication + /// is indexed. Older index entries omit this field and fall back to the + /// immutable commit envelope. + #[serde(default, skip_serializing_if = "Option::is_none")] + pub snapshot: Option, +} + +#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)] +pub struct JournalSnapshotMetadata { + pub state: BucketState, + pub delta: BucketDelta, } /// One atomic checkpoint for all branch-local journal-derived indexes. diff --git a/extensions/s3/core/src/repository.rs b/extensions/s3/core/src/repository.rs index bf8c133f..366ac1ff 100644 --- a/extensions/s3/core/src/repository.rs +++ b/extensions/s3/core/src/repository.rs @@ -5,7 +5,7 @@ use std::{ atomic::{AtomicBool, AtomicU64, Ordering}, Arc, Mutex, OnceLock, RwLock, Weak, }, - time::Duration, + time::{Duration, Instant}, }; use futures_util::{stream, StreamExt}; @@ -151,6 +151,15 @@ struct ListObjectsCursor { traversal: prolly::RangeCursor, } +struct PutObjectRequest { + key: Vec, + bytes: Vec, + headers: ObjectHeaders, + user_metadata: BTreeMap, + operation: OperationId, + reconcile_before_publication: bool, +} + #[derive(Clone, Debug, PartialEq, Eq)] pub struct ListObjectsPage { pub snapshot: CommitId, @@ -504,10 +513,27 @@ pub struct NodeCachePrewarmReport { pub snapshot: CommitId, pub object_nodes: usize, pub version_nodes: usize, + pub pinned_nodes: usize, + pub elapsed_millis: u64, pub before: crate::NodeCacheSnapshot, pub after: crate::NodeCacheSnapshot, } +/// Branch-indexed immutable snapshot roots used by listing, diff, and merge +/// planning without downloading the containing commit's node-pack body. +#[derive(Clone, Debug, PartialEq, Eq)] +pub struct ResolvedSnapshot { + pub repository: crate::RepositoryId, + pub branch: String, + pub commit: CommitId, + pub generation: CommitGeneration, + pub parents: Vec, + pub state: BucketState, + pub delta: BucketDelta, + /// True when the branch-local journal index supplied the roots directly. + pub indexed: bool, +} + #[derive(Clone, Debug, PartialEq, Eq)] pub struct BranchHead { pub name: String, @@ -860,6 +886,26 @@ pub struct Repository { } impl Repository

{ + /// Load and decode the create-once repository format marker without + /// acquiring writer authority. Clients use this to discover persisted + /// tree geometry before opening an existing repository. + pub async fn load_format(plane: Arc

, repository_prefix: &str) -> Result { + let stored = plane + .get(GetRequest { + path: format_path(repository_prefix)?, + range: None, + physical_version: None, + }) + .await? + .ok_or_else(|| { + Error::new( + ErrorCode::RepositoryNotInitialized, + "repository format marker does not exist", + ) + })?; + decode_canonical(&stored.bytes) + } + pub async fn initialize(plane: Arc

, options: RepositoryOptions) -> Result { validate_options(&options)?; if options.read_only { @@ -992,20 +1038,7 @@ impl Repository

{ pub async fn open(plane: Arc

, options: RepositoryOptions) -> Result { validate_options(&options)?; - let stored = plane - .get(GetRequest { - path: format_path(&options.repository_prefix)?, - range: None, - physical_version: None, - }) - .await? - .ok_or_else(|| { - Error::new( - ErrorCode::RepositoryNotInitialized, - "repository format marker does not exist", - ) - })?; - let format: RepositoryFormat = decode_canonical(&stored.bytes)?; + let format = Self::load_format(plane.clone(), &options.repository_prefix).await?; validate_format_compatibility(&format, &options)?; let repository = Self::from_format(plane, options, format)?; repository.restore_gc_state().await?; @@ -1207,11 +1240,12 @@ impl Repository

{ branch: &str, snapshot: CommitId, ) -> Result { + let started = Instant::now(); validate_branch(branch)?; self.locator.register(branch)?; self.require_branch_indexes_ready(branch).await?; let before = self.node_store.node_cache_snapshot(); - let commit = self.load_commit_object(snapshot).await?.commit; + let commit = self.resolve_snapshot(branch, snapshot).await?; let engine = self.engine(self.node_store.clone()); let object_stats = engine .collect_stats(&self.tree_from_root(&commit.state.objects)?) @@ -1223,6 +1257,8 @@ impl Repository

{ snapshot, object_nodes: object_stats.num_nodes, version_nodes: version_stats.num_nodes, + pinned_nodes: 0, + elapsed_millis: started.elapsed().as_millis().min(u128::from(u64::MAX)) as u64, before, after: self.node_store.node_cache_snapshot(), }) @@ -1237,6 +1273,7 @@ impl Repository

{ snapshot: CommitId, levels: usize, ) -> Result { + let started = Instant::now(); if !(1..=64).contains(&levels) { return Err(Error::new( ErrorCode::InvalidLimit, @@ -1247,19 +1284,26 @@ impl Repository

{ self.locator.register(branch)?; self.require_branch_indexes_ready(branch).await?; let before = self.node_store.node_cache_snapshot(); - let commit = self.load_commit_object(snapshot).await?.commit; + let commit = self.resolve_snapshot(branch, snapshot).await?; let object_nodes = self .prewarm_root_levels(&commit.state.objects, levels) .await?; let version_nodes = self .prewarm_root_levels(&commit.state.versions, levels) .await?; + let after = self.node_store.node_cache_snapshot(); Ok(NodeCachePrewarmReport { snapshot, object_nodes, version_nodes, + // Pinning is deliberately best effort. Report only nodes the + // configured cache actually retained, not merely nodes visited + // during the upper-level traversal. + pinned_nodes: usize::try_from(after.pinned_nodes.saturating_sub(before.pinned_nodes)) + .unwrap_or(usize::MAX), + elapsed_millis: started.elapsed().as_millis().min(u128::from(u64::MAX)) as u64, before, - after: self.node_store.node_cache_snapshot(), + after, }) } @@ -1283,14 +1327,15 @@ impl Repository

{ format!("prewarm node could not be decoded: {error}"), ) })?; - Ok::<_, Error>(node) + Ok::<_, Error>((cid, bytes, node)) })) .buffered(32) .collect::>() .await; let mut next = Vec::new(); for node in loaded { - let node = node?; + let (cid, bytes, node) = node?; + self.node_store.pin_node(cid, bytes).await?; loaded_nodes = loaded_nodes.saturating_add(1); if !node.leaf { for child in node.vals { @@ -1313,6 +1358,48 @@ impl Repository

{ Ok(self.publisher.load(branch).await?.value.target) } + /// Resolve immutable state roots from the branch-local journal index. + /// Legacy index entries fall back to the bounded commit-metadata range + /// read and remain fully compatible. + pub async fn resolve_snapshot( + &self, + branch: &str, + snapshot: CommitId, + ) -> Result { + validate_branch(branch)?; + self.locator.register(branch)?; + self.require_branch_indexes_ready(branch).await?; + if let Some(entry) = self + .journal_indexes + .commit_graph_entry(branch, snapshot) + .await? + { + if let Some(indexed) = entry.snapshot { + return Ok(ResolvedSnapshot { + repository: self.format.repository_id, + branch: branch.to_string(), + commit: snapshot, + generation: entry.generation, + parents: entry.parents, + state: indexed.state, + delta: indexed.delta, + indexed: true, + }); + } + } + let commit = self.load_commit_metadata(snapshot).await?; + Ok(ResolvedSnapshot { + repository: self.format.repository_id, + branch: branch.to_string(), + commit: snapshot, + generation: commit.generation, + parents: commit.parents.clone(), + state: commit.state.clone(), + delta: commit.delta.clone(), + indexed: false, + }) + } + pub async fn create_branch(&self, name: &str, from: CommitId) -> Result { self.create_branch_from(&self.options.default_branch, name, from) .await @@ -2419,8 +2506,18 @@ impl Repository

{ // retry of an earlier publication. Avoid walking the unindexed journal // to prove absence on every hot-branch write; caller-stable operation // IDs still take the full reconciliation path below. - self.put_object_inner(branch, key, bytes, headers, user_metadata, operation, false) - .await + self.put_object_inner( + branch, + PutObjectRequest { + key, + bytes, + headers, + user_metadata, + operation, + reconcile_before_publication: false, + }, + ) + .await } async fn validate_commit_session(&self, session: &CommitSessionManifest) -> Result<()> { @@ -2499,20 +2596,33 @@ impl Repository

{ user_metadata: BTreeMap, operation: OperationId, ) -> Result { - self.put_object_inner(branch, key, bytes, headers, user_metadata, operation, true) - .await + self.put_object_inner( + branch, + PutObjectRequest { + key, + bytes, + headers, + user_metadata, + operation, + reconcile_before_publication: true, + }, + ) + .await } async fn put_object_inner( &self, branch: &str, - key: Vec, - bytes: Vec, - headers: ObjectHeaders, - user_metadata: BTreeMap, - operation: OperationId, - reconcile_before_publication: bool, + request: PutObjectRequest, ) -> Result { + let PutObjectRequest { + key, + bytes, + headers, + user_metadata, + operation, + reconcile_before_publication, + } = request; if !self.writable.load(Ordering::Acquire) { return Err(Error::new( ErrorCode::PreconditionFailed, @@ -3187,7 +3297,7 @@ impl Repository

{ }; self.locator.register(branch)?; self.require_branch_indexes_ready(branch).await?; - let commit = self.load_commit_metadata(cursor.snapshot).await?; + let commit = self.resolve_snapshot(branch, cursor.snapshot).await?; let objects = self.tree_from_root(&commit.state.objects)?; let engine = self.engine(self.node_store.clone()); let mut page = engine @@ -5193,8 +5303,8 @@ impl Repository

{ validate_branch(branch)?; self.locator.register(branch)?; self.require_branch_indexes_ready(branch).await?; - let from_commit = self.load_commit_metadata(from).await?; - let to_commit = self.load_commit_metadata(to).await?; + let from_commit = self.resolve_snapshot(branch, from).await?; + let to_commit = self.resolve_snapshot(branch, to).await?; if to_commit.parents.first() == Some(&from) && to_commit.delta.changes_root.is_none() { return to_commit .delta @@ -5244,8 +5354,8 @@ impl Repository

{ } self.locator.register(branch)?; self.require_branch_indexes_ready(branch).await?; - let from_commit = self.load_commit_metadata(from).await?; - let to_commit = self.load_commit_metadata(to).await?; + let from_commit = self.resolve_snapshot(branch, from).await?; + let to_commit = self.resolve_snapshot(branch, to).await?; // Direct children carry an exact logical transition stream, inline or // in an immutable delta tree. Page that stream instead of comparing // snapshot structure; tree boundary churn is irrelevant to this path. @@ -5260,8 +5370,9 @@ impl Repository

{ None => None, Some(ObjectDiffTraversal::Structural(_)) => unreachable!("excluded above"), }; - let (transitions, next_after) = - self.commit_delta_page(&to_commit, after, limit).await?; + let (transitions, next_after) = self + .commit_delta_page(&to_commit.delta, after, limit) + .await?; let changes = transitions .iter() .map(object_diff_from_transition) @@ -5939,8 +6050,10 @@ impl Repository

{ )); } let permit = self.active_permit(&cursor.target_branch, now).await?; - let ours = self.load_commit_metadata(cursor.ours).await?; - let theirs = self.load_commit_metadata(cursor.theirs).await?; + let (ours, theirs) = futures_util::try_join!( + self.merge_graph_entry(&cursor.target_branch, &cursor.source_branch, cursor.ours,), + self.merge_graph_entry(&cursor.target_branch, &cursor.source_branch, cursor.theirs,), + )?; let generation = CommitGeneration( ours.generation .0 @@ -8705,6 +8818,49 @@ impl Repository

{ generation: commit_object.generation, parents: commit_object.parents.clone(), first_parent_jumps: Vec::new(), + snapshot: Some(crate::JournalSnapshotMetadata { + state: commit_object.state.clone(), + delta: commit_object.delta.clone(), + }), + }) + } + + async fn resolve_merge_snapshot( + &self, + target_branch: &str, + source_branch: &str, + commit: CommitId, + ) -> Result { + for branch in [target_branch, source_branch] { + if let Some(entry) = self + .journal_indexes + .commit_graph_entry(branch, commit) + .await? + { + if let Some(snapshot) = entry.snapshot { + return Ok(ResolvedSnapshot { + repository: self.format.repository_id, + branch: branch.to_string(), + commit, + generation: entry.generation, + parents: entry.parents, + state: snapshot.state, + delta: snapshot.delta, + indexed: true, + }); + } + } + } + let commit_object = self.load_commit_metadata(commit).await?; + Ok(ResolvedSnapshot { + repository: self.format.repository_id, + branch: target_branch.to_string(), + commit, + generation: commit_object.generation, + parents: commit_object.parents.clone(), + state: commit_object.state.clone(), + delta: commit_object.delta.clone(), + indexed: false, }) } @@ -8958,9 +9114,15 @@ impl Repository

{ "merge plan has no selected base", ) })?; - let base_commit = self.load_commit_metadata(base).await?; - let ours_commit = self.load_commit_metadata(cursor.ours).await?; - let theirs_commit = self.load_commit_metadata(cursor.theirs).await?; + let (base_commit, ours_commit, theirs_commit) = futures_util::try_join!( + self.resolve_merge_snapshot(&cursor.target_branch, &cursor.source_branch, base), + self.resolve_merge_snapshot(&cursor.target_branch, &cursor.source_branch, cursor.ours,), + self.resolve_merge_snapshot( + &cursor.target_branch, + &cursor.source_branch, + cursor.theirs, + ), + )?; let base_tree = self.tree_from_root(&base_commit.state.objects)?; let ours_tree = self.tree_from_root(&ours_commit.state.objects)?; let theirs_tree = self.tree_from_root(&theirs_commit.state.objects)?; @@ -9209,8 +9371,8 @@ impl Repository

{ async fn direct_parent_diffs( &self, parent: CommitId, - parent_commit: &BucketCommit, - child_commit: &BucketCommit, + parent_commit: &ResolvedSnapshot, + child_commit: &ResolvedSnapshot, ) -> Result>> { if child_commit.parents.first() != Some(&parent) || child_commit.delta.changes_root.is_some() @@ -9279,12 +9441,14 @@ impl Repository

{ async fn direct_parent_diff_page( &self, - parent_commit: &BucketCommit, - child_commit: &BucketCommit, + parent_commit: &ResolvedSnapshot, + child_commit: &ResolvedSnapshot, after: Option<&[u8]>, limit: usize, ) -> Result<(Vec, Option>)> { - let (transitions, next_after) = self.commit_delta_page(child_commit, after, limit).await?; + let (transitions, next_after) = self + .commit_delta_page(&child_commit.delta, after, limit) + .await?; if transitions.is_empty() { return Ok((Vec::new(), next_after)); } @@ -9340,11 +9504,11 @@ impl Repository

{ async fn commit_delta_page( &self, - commit: &BucketCommit, + delta: &BucketDelta, after: Option<&[u8]>, limit: usize, ) -> Result<(Vec, Option>)> { - if let Some(root) = &commit.delta.changes_root { + if let Some(root) = &delta.changes_root { let tree = self.tree_from_root(root)?; let engine = self.engine(self.node_store.clone()); let mut entries = match after { @@ -9381,14 +9545,13 @@ impl Repository

{ } let start = after.map_or(0, |after| { - commit - .delta + delta .changes .partition_point(|transition| transition.key.as_slice() <= after) }); - let end = start.saturating_add(limit).min(commit.delta.changes.len()); - let transitions = commit.delta.changes[start..end].to_vec(); - let next = (end < commit.delta.changes.len()).then(|| { + let end = start.saturating_add(limit).min(delta.changes.len()); + let transitions = delta.changes[start..end].to_vec(); + let next = (end < delta.changes.len()).then(|| { transitions .last() .expect("a truncated inline delta page has a last transition") @@ -9403,8 +9566,14 @@ impl Repository

{ cursor: &mut MergeCursor, max_steps: usize, ) -> Result { - let ours_commit = self.load_commit_metadata(cursor.ours).await?; - let theirs_commit = self.load_commit_metadata(cursor.theirs).await?; + let (ours_commit, theirs_commit) = futures_util::try_join!( + self.resolve_merge_snapshot(&cursor.target_branch, &cursor.source_branch, cursor.ours,), + self.resolve_merge_snapshot( + &cursor.target_branch, + &cursor.source_branch, + cursor.theirs, + ), + )?; let base = cursor.selected_base.ok_or_else(|| { Error::new( ErrorCode::InternalInvariant, @@ -9505,11 +9674,13 @@ impl Repository

{ /// structural union fallback. async fn direct_child_version_mutations_page( &self, - child_commit: &BucketCommit, + child_commit: &ResolvedSnapshot, after: Option<&[u8]>, limit: usize, ) -> Result<(Vec, Option>)> { - let (transitions, next_after) = self.commit_delta_page(child_commit, after, limit).await?; + let (transitions, next_after) = self + .commit_delta_page(&child_commit.delta, after, limit) + .await?; let keys = transitions .iter() .map(|transition| transition.key.as_slice()) @@ -9553,7 +9724,7 @@ impl Repository

{ Ok((mutations, next_after)) } - async fn commit_delta_contains_delete(&self, commit: &BucketCommit) -> Result { + async fn commit_delta_contains_delete(&self, commit: &ResolvedSnapshot) -> Result { if commit.delta.changes_root.is_none() { return Ok(commit .delta @@ -9564,7 +9735,7 @@ impl Repository

{ let mut after = None; loop { let (page, next) = self - .commit_delta_page(commit, after.as_deref(), 1_000) + .commit_delta_page(&commit.delta, after.as_deref(), 1_000) .await?; if page.iter().any(|transition| transition.delete_marker) { return Ok(true); @@ -9623,8 +9794,10 @@ impl Repository

{ cursor.phase = MergePhase::ReadyToPublish; return Ok(0); } - let ours = self.load_commit_metadata(cursor.ours).await?; - let theirs = self.load_commit_metadata(cursor.theirs).await?; + let (ours, theirs) = futures_util::try_join!( + self.merge_graph_entry(&cursor.target_branch, &cursor.source_branch, cursor.ours,), + self.merge_graph_entry(&cursor.target_branch, &cursor.source_branch, cursor.theirs,), + )?; let generation = CommitGeneration( ours.generation .0 @@ -9802,7 +9975,9 @@ impl Repository

{ )); } if cursor.selected_base == Some(cursor.ours) { - let theirs = self.load_commit_metadata(cursor.theirs).await?; + let theirs = self + .resolve_merge_snapshot(&cursor.target_branch, &cursor.source_branch, cursor.theirs) + .await?; if theirs.parents.first() == Some(&cursor.ours) && theirs.delta.changes_root.is_some() && cursor.built_changes == cursor.planned_changes @@ -9821,7 +9996,7 @@ impl Repository

{ }) .transpose()?; let (transitions, next_after) = - self.commit_delta_page(&theirs, after, limit).await?; + self.commit_delta_page(&theirs.delta, after, limit).await?; let changes = transitions .into_iter() .map(|transition| MergeChange { diff --git a/extensions/s3/core/src/store.rs b/extensions/s3/core/src/store.rs index 5fa5bf0c..8c8b9ef5 100644 --- a/extensions/s3/core/src/store.rs +++ b/extensions/s3/core/src/store.rs @@ -1,12 +1,12 @@ use std::{ - collections::{BTreeMap, VecDeque}, + collections::{BTreeMap, BTreeSet, VecDeque}, sync::{ atomic::{AtomicU64, Ordering}, Arc, Mutex, RwLock, Weak, }, }; -use futures_util::StreamExt; +use futures_util::{stream, StreamExt}; use prolly::{AsyncStore, BatchOp, Cid}; use crate::{ @@ -48,6 +48,13 @@ struct PackedNodeState { fetched_bytes: AtomicU64, avoided_bytes: AtomicU64, admission_rejections: AtomicU64, + node_requests: AtomicU64, + requested_bytes: AtomicU64, + prefetch_batches: AtomicU64, + prefetched_nodes: AtomicU64, + pinned: Mutex>, + pinned_bytes: AtomicU64, + max_tracked_pins: usize, locator: RwLock>>, } @@ -65,6 +72,13 @@ struct DirectNodeState { fetched_bytes: AtomicU64, avoided_bytes: AtomicU64, admission_rejections: AtomicU64, + node_requests: AtomicU64, + requested_bytes: AtomicU64, + prefetch_batches: AtomicU64, + prefetched_nodes: AtomicU64, + pinned: Mutex>, + pinned_bytes: AtomicU64, + max_tracked_pins: usize, } struct BoundedNodeLocations { @@ -144,6 +158,17 @@ pub struct NodeCacheSnapshot { pub fetched_bytes: u64, pub avoided_bytes: u64, pub admission_rejections: u64, + /// Number of immutable node values returned to tree engines. + pub node_requests: u64, + /// Canonical node bytes requested by tree engines, independent of tier. + pub requested_bytes: u64, + /// Predictive multi-node prefetch batches issued by tree traversal. + pub prefetch_batches: u64, + /// Nodes requested by predictive prefetch batches. + pub prefetched_nodes: u64, + /// Upper-level nodes retained in the advisory pinned tier. + pub pinned_nodes: u64, + pub pinned_bytes: u64, } impl NodeCacheSnapshot { @@ -161,6 +186,60 @@ impl NodeCacheSnapshot { admission_rejections: self .admission_rejections .saturating_add(other.admission_rejections), + node_requests: self.node_requests.saturating_add(other.node_requests), + requested_bytes: self.requested_bytes.saturating_add(other.requested_bytes), + prefetch_batches: self.prefetch_batches.saturating_add(other.prefetch_batches), + prefetched_nodes: self.prefetched_nodes.saturating_add(other.prefetched_nodes), + pinned_nodes: self.pinned_nodes.saturating_add(other.pinned_nodes), + pinned_bytes: self.pinned_bytes.saturating_add(other.pinned_bytes), + } + } + + /// Saturating interval metrics suitable for request and startup reports. + pub fn delta_since(self, earlier: Self) -> Self { + Self { + hits: self.hits.saturating_sub(earlier.hits), + misses: self.misses.saturating_sub(earlier.misses), + insertions: self.insertions.saturating_sub(earlier.insertions), + errors: self.errors.saturating_sub(earlier.errors), + corruptions: self.corruptions.saturating_sub(earlier.corruptions), + coalesced_waits: self.coalesced_waits.saturating_sub(earlier.coalesced_waits), + ranged_fetches: self.ranged_fetches.saturating_sub(earlier.ranged_fetches), + fetched_bytes: self.fetched_bytes.saturating_sub(earlier.fetched_bytes), + avoided_bytes: self.avoided_bytes.saturating_sub(earlier.avoided_bytes), + admission_rejections: self + .admission_rejections + .saturating_sub(earlier.admission_rejections), + node_requests: self.node_requests.saturating_sub(earlier.node_requests), + requested_bytes: self.requested_bytes.saturating_sub(earlier.requested_bytes), + prefetch_batches: self + .prefetch_batches + .saturating_sub(earlier.prefetch_batches), + prefetched_nodes: self + .prefetched_nodes + .saturating_sub(earlier.prefetched_nodes), + pinned_nodes: self.pinned_nodes.saturating_sub(earlier.pinned_nodes), + pinned_bytes: self.pinned_bytes.saturating_sub(earlier.pinned_bytes), + } + } + + pub fn hit_ratio(self) -> f64 { + let lookups = self.hits.saturating_add(self.misses); + if lookups == 0 { + 0.0 + } else { + self.hits as f64 / lookups as f64 + } + } + + /// Provider node-body bytes fetched per canonical node byte returned. + /// Values below one indicate cache reuse. Client adapters combine this + /// with all provider response bytes for end-to-end metadata amplification. + pub fn byte_amplification(self) -> f64 { + if self.requested_bytes == 0 { + 0.0 + } else { + self.fetched_bytes as f64 / self.requested_bytes as f64 } } } @@ -291,6 +370,13 @@ impl

ProllyObjectStore

{ fetched_bytes: AtomicU64::new(0), avoided_bytes: AtomicU64::new(0), admission_rejections: AtomicU64::new(0), + node_requests: AtomicU64::new(0), + requested_bytes: AtomicU64::new(0), + prefetch_batches: AtomicU64::new(0), + prefetched_nodes: AtomicU64::new(0), + pinned: Mutex::new(BTreeSet::new()), + pinned_bytes: AtomicU64::new(0), + max_tracked_pins: max_cached_locations, locator: RwLock::new(None), })), packed_pending: Some(Arc::new(RwLock::new(BTreeMap::new()))), @@ -328,6 +414,13 @@ impl

ProllyObjectStore

{ fetched_bytes: AtomicU64::new(0), avoided_bytes: AtomicU64::new(0), admission_rejections: AtomicU64::new(0), + node_requests: AtomicU64::new(0), + requested_bytes: AtomicU64::new(0), + prefetch_batches: AtomicU64::new(0), + prefetched_nodes: AtomicU64::new(0), + pinned: Mutex::new(BTreeSet::new()), + pinned_bytes: AtomicU64::new(0), + max_tracked_pins: 65_536, })), write_direct: true, } @@ -415,6 +508,23 @@ impl ProllyObjectStore

{ pub fn node_cache_snapshot(&self) -> NodeCacheSnapshot { if let Some(state) = &self.packed { + let exact_pinned = state + .node_cache + .as_ref() + .and_then(|cache| cache.pinned_usage()); + let (pinned_nodes, pinned_bytes) = exact_pinned.map_or_else( + || { + ( + state + .pinned + .lock() + .unwrap_or_else(std::sync::PoisonError::into_inner) + .len() as u64, + state.pinned_bytes.load(Ordering::Relaxed), + ) + }, + |(nodes, bytes)| (nodes as u64, bytes as u64), + ); return NodeCacheSnapshot { hits: state.cache_hits.load(Ordering::Relaxed), misses: state.cache_misses.load(Ordering::Relaxed), @@ -426,6 +536,12 @@ impl ProllyObjectStore

{ fetched_bytes: state.fetched_bytes.load(Ordering::Relaxed), avoided_bytes: state.avoided_bytes.load(Ordering::Relaxed), admission_rejections: state.admission_rejections.load(Ordering::Relaxed), + node_requests: state.node_requests.load(Ordering::Relaxed), + requested_bytes: state.requested_bytes.load(Ordering::Relaxed), + prefetch_batches: state.prefetch_batches.load(Ordering::Relaxed), + prefetched_nodes: state.prefetched_nodes.load(Ordering::Relaxed), + pinned_nodes, + pinned_bytes, }; } if let Some(state) = &self.direct { @@ -440,6 +556,16 @@ impl ProllyObjectStore

{ fetched_bytes: state.fetched_bytes.load(Ordering::Relaxed), avoided_bytes: state.avoided_bytes.load(Ordering::Relaxed), admission_rejections: state.admission_rejections.load(Ordering::Relaxed), + node_requests: state.node_requests.load(Ordering::Relaxed), + requested_bytes: state.requested_bytes.load(Ordering::Relaxed), + prefetch_batches: state.prefetch_batches.load(Ordering::Relaxed), + prefetched_nodes: state.prefetched_nodes.load(Ordering::Relaxed), + pinned_nodes: state + .pinned + .lock() + .unwrap_or_else(std::sync::PoisonError::into_inner) + .len() as u64, + pinned_bytes: state.pinned_bytes.load(Ordering::Relaxed), }; } NodeCacheSnapshot::default() @@ -1043,16 +1169,115 @@ impl ProllyObjectStore

{ } Ok(Some(object.bytes)) } + + fn record_returned_node(&self, bytes: usize) { + let bytes = bytes as u64; + if let Some(state) = &self.packed { + state.node_requests.fetch_add(1, Ordering::Relaxed); + state.requested_bytes.fetch_add(bytes, Ordering::Relaxed); + } else if let Some(state) = &self.direct { + state.node_requests.fetch_add(1, Ordering::Relaxed); + state.requested_bytes.fetch_add(bytes, Ordering::Relaxed); + } + } + + fn record_prefetch(&self, nodes: usize) { + let Some(nodes) = u64::try_from(nodes).ok() else { + return; + }; + if let Some(state) = &self.packed { + state.prefetch_batches.fetch_add(1, Ordering::Relaxed); + state.prefetched_nodes.fetch_add(nodes, Ordering::Relaxed); + } else if let Some(state) = &self.direct { + state.prefetch_batches.fetch_add(1, Ordering::Relaxed); + state.prefetched_nodes.fetch_add(nodes, Ordering::Relaxed); + } + } + + /// Promote a verified root or upper-level node into the cache's advisory + /// pinned tier. The provider remains authoritative if the tier is lost. + pub(crate) async fn pin_node(&self, cid: Cid, bytes: Vec) -> Result<()> { + if sha256(&bytes).as_slice() != cid.as_bytes() { + return Err(Error::new( + ErrorCode::CorruptNode, + "cannot pin a node that fails CID verification", + )); + } + let (cache, key) = if let Some(state) = &self.packed { + let Some(cache) = state.node_cache.as_ref() else { + return Ok(()); + }; + let Some(key) = self.node_cache_key(cid.clone()) else { + return Ok(()); + }; + (cache.clone(), key) + } else if let Some(state) = &self.direct { + let Some(key) = self.direct_cache_key(cid.clone()) else { + return Ok(()); + }; + (state.node_cache.clone(), key) + } else { + return Ok(()); + }; + if !cache.admits(&key, bytes.len()) { + if let Some(state) = &self.packed { + state.admission_rejections.fetch_add(1, Ordering::Relaxed); + } else if let Some(state) = &self.direct { + state.admission_rejections.fetch_add(1, Ordering::Relaxed); + } + return Ok(()); + } + if cache.pin(key, bytes.clone()).await.is_err() { + if let Some(state) = &self.packed { + state.cache_errors.fetch_add(1, Ordering::Relaxed); + } else if let Some(state) = &self.direct { + state.cache_errors.fetch_add(1, Ordering::Relaxed); + } + return Ok(()); + } + let (pinned, pinned_bytes, max_tracked_pins, exact_usage) = + if let Some(state) = &self.packed { + ( + &state.pinned, + &state.pinned_bytes, + state.max_tracked_pins, + cache.pinned_usage().is_some(), + ) + } else { + let state = self.direct.as_ref().expect("direct state selected above"); + ( + &state.pinned, + &state.pinned_bytes, + state.max_tracked_pins, + false, + ) + }; + if exact_usage { + return Ok(()); + } + let mut pinned = pinned + .lock() + .unwrap_or_else(std::sync::PoisonError::into_inner); + if !pinned.contains(&cid) && pinned.len() < max_tracked_pins && pinned.insert(cid) { + pinned_bytes.fetch_add(bytes.len() as u64, Ordering::Relaxed); + } + Ok(()) + } } impl AsyncStore for ProllyObjectStore

{ type Error = Error; async fn get(&self, key: &[u8]) -> Result>> { - if self.packed.is_some() { - return self.get_packed(key).await; + let value = if self.packed.is_some() { + self.get_packed(key).await? + } else { + self.get_direct(key).await? + }; + if let Some(bytes) = &value { + self.record_returned_node(bytes.len()); } - self.get_direct(key).await + Ok(value) } async fn put(&self, key: &[u8], value: &[u8]) -> Result<()> { @@ -1156,4 +1381,24 @@ impl AsyncStore for ProllyObjectStore

{ fn read_parallelism(&self) -> usize { 16 } + + fn prefers_batch_reads(&self) -> bool { + true + } + + async fn batch_get_ordered_unique(&self, keys: &[&[u8]]) -> Result>>> { + if keys.len() > 1 { + self.record_prefetch(keys.len()); + } + let keys = keys.iter().map(|key| key.to_vec()).collect::>(); + stream::iter( + keys.into_iter() + .map(|key| async move { self.get(&key).await }), + ) + .buffered(self.read_parallelism()) + .collect::>() + .await + .into_iter() + .collect() + } } diff --git a/extensions/s3/core/tests/journal_indexes.rs b/extensions/s3/core/tests/journal_indexes.rs index a9a5d243..30a5cb54 100644 --- a/extensions/s3/core/tests/journal_indexes.rs +++ b/extensions/s3/core/tests/journal_indexes.rs @@ -2,12 +2,63 @@ use std::{collections::BTreeMap, sync::Arc, time::Duration}; use prolly::TreeFormat; use prolly_s3_core::{ - AuthorityScope, AuthorityStamp, BucketCommit, BucketDelta, BucketState, CommitGeneration, - CommitId, CommitPublication, JournalDerivedIndexes, MemoryObjectPlane, NodePack, NodePackEntry, - OperationId, RepositoryId, RootManifest, ShardWriterAuthority, ShardedBranchPublisher, - TreeFormatDigest, + decode_canonical, encode_canonical, AuthorityScope, AuthorityStamp, BucketCommit, BucketDelta, + BucketState, CommitGeneration, CommitId, CommitPublication, JournalCommitGraphEntry, + JournalDerivedIndexes, MemoryObjectPlane, NodePack, NodePackEntry, OperationId, RepositoryId, + RootManifest, ShardWriterAuthority, ShardedBranchPublisher, TreeFormatDigest, }; +#[derive(Debug, serde::Serialize, serde::Deserialize)] +struct LegacyJournalCommitGraphEntry { + commit: CommitId, + generation: CommitGeneration, + parents: Vec, + first_parent_jumps: Vec, +} + +#[test] +fn graph_index_decodes_entries_written_before_snapshot_roots_were_added() { + let commit = CommitId::from_hash([0x11; 32]); + let encoded = encode_canonical(&LegacyJournalCommitGraphEntry { + commit, + generation: CommitGeneration(7), + parents: Vec::new(), + first_parent_jumps: Vec::new(), + }) + .unwrap(); + let decoded: JournalCommitGraphEntry = decode_canonical(&encoded).unwrap(); + assert_eq!(decoded.commit, commit); + assert!(decoded.snapshot.is_none()); + + let reencoded = encode_canonical(&JournalCommitGraphEntry { + commit, + generation: CommitGeneration(8), + parents: Vec::new(), + first_parent_jumps: Vec::new(), + snapshot: Some(prolly_s3_core::JournalSnapshotMetadata { + state: BucketState { + objects: RootManifest { + root: None, + format_digest: TreeFormatDigest::from_hash([0x22; 32]), + }, + versions: RootManifest { + root: None, + format_digest: TreeFormatDigest::from_hash([0x22; 32]), + }, + }, + delta: BucketDelta { + input_digest: [0; 32], + changes: Vec::new(), + changes_root: None, + change_count: 0, + }, + }), + }) + .unwrap(); + let error = decode_canonical::(&reencoded).unwrap_err(); + assert_eq!(error.code, prolly_s3_core::ErrorCode::CorruptCommit); +} + fn operation(value: u128) -> OperationId { OperationId(uuid::Uuid::from_u128(value)) } diff --git a/extensions/s3/core/tests/repository.rs b/extensions/s3/core/tests/repository.rs index d7ffd35a..b37be03b 100644 --- a/extensions/s3/core/tests/repository.rs +++ b/extensions/s3/core/tests/repository.rs @@ -2,9 +2,10 @@ use std::{collections::BTreeMap, io::Write as _, sync::Arc}; use md5::{Digest as _, Md5}; use prolly_s3_core::{ - decode_canonical, encode_canonical, FixedClock, GetRequest, ListRequest, + decode_canonical, encode_canonical, BoundaryRule, FixedClock, GetRequest, ListRequest, LogicalObjectVersionKind, MemoryNodeCache, MemoryObjectPlane, ObjectHeaders, ObjectPath, ObjectPlane, ProviderPerKeyVersionLimit, Repository, RepositoryOptions, SequenceIdSource, + TreeFormat, }; use sha2::Sha256; @@ -45,6 +46,72 @@ async fn repository_cache_snapshot_includes_ref_catalog_reads_after_reopen() { ); } +#[tokio::test] +async fn listing_predictively_prefetches_adjacent_metadata_nodes() { + let plane = Arc::new(MemoryObjectPlane::new(true)); + let mut format = TreeFormat::default(); + format.chunking.min = 2; + format.chunking.target = 4; + format.chunking.max = 8; + format.chunking.rule = BoundaryRule::HashThreshold { factor: 4 }; + format.chunking.hard_max_node_bytes = 64 * 1024; + let options = RepositoryOptions { + repository_prefix: ".tests/list-prefetch".to_string(), + writer: "prefetch-writer".to_string(), + state_tree_format: format, + provider_per_key_version_limit: ProviderPerKeyVersionLimit::Finite(10_000), + ..RepositoryOptions::default() + }; + let repository = Repository::initialize(plane.clone(), options.clone()) + .await + .unwrap(); + let session = repository + .begin_commit_session("main", "prefetch fixture", 60_000) + .await + .unwrap(); + let inputs = (0..128) + .map(|index| { + ( + format!("prefetch/{index:04}").into_bytes(), + vec![index as u8], + ObjectHeaders::default(), + BTreeMap::new(), + ) + }) + .collect(); + let mutations = repository + .stage_commit_session_put_batch(&session, inputs, 16) + .await + .unwrap(); + repository + .publish_commit_session(session, mutations) + .await + .unwrap(); + repository.advance_branch_indexes("main").await.unwrap(); + drop(repository); + + let reader = Repository::open( + plane, + RepositoryOptions { + read_only: true, + node_cache: Some(Arc::new(MemoryNodeCache::new(64 * 1024 * 1024))), + ..options + }, + ) + .await + .unwrap(); + let before = reader.node_cache_snapshot(); + let page = reader + .list_objects_page("main", b"prefetch/", None, 100) + .await + .unwrap(); + let activity = reader.node_cache_snapshot().delta_since(before); + assert_eq!(page.objects.len(), 100); + assert!(page.continuation.is_some()); + assert!(activity.prefetch_batches > 0); + assert!(activity.prefetched_nodes > activity.prefetch_batches); +} + #[tokio::test] async fn repository_put_read_replay_and_reopen_use_only_authority() { let plane = Arc::new(MemoryObjectPlane::new(true)); @@ -93,6 +160,13 @@ async fn repository_put_read_replay_and_reopen_use_only_authority() { .unwrap(); assert_eq!(upper.object_nodes, 1); assert_eq!(upper.version_nodes, 1); + assert_eq!(upper.pinned_nodes, 2); + assert!(repository.node_cache_snapshot().pinned_nodes >= 2); + repository.advance_branch_indexes("main").await.unwrap(); + let resolved = repository.resolve_snapshot("main", first.id).await.unwrap(); + assert!(resolved.indexed); + assert_eq!(resolved.commit, first.id); + assert_eq!(resolved.parents, vec![first.parents[0]]); assert_eq!( repository .prewarm_node_cache_levels("main", first.id, 0)