From ac9bfc48b0c711a6395b8b20641cd7754dcbd329 Mon Sep 17 00:00:00 2001 From: Nils Pukropp Date: Sun, 19 Jul 2026 14:27:34 +0200 Subject: [PATCH] fixed orphaned docs --- Cargo.lock | 5 +- Cargo.toml | 2 +- crates/proton-drive-rs/Cargo.toml | 2 +- crates/proton-drive-rs/src/client.rs | 8 +- crates/proton-sdk/Cargo.toml | 1 + crates/proton-sdk/src/cache.rs | 241 ++++++++++++++++++++++-- crates/proton-sdk/src/crypto/content.rs | 18 +- crates/proton-sdk/src/http.rs | 12 +- 8 files changed, 248 insertions(+), 41 deletions(-) diff --git a/Cargo.lock b/Cargo.lock index e7b3c27..fca19cf 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -2036,7 +2036,7 @@ dependencies = [ [[package]] name = "proton-drive-rs" -version = "0.1.8" +version = "0.1.9" dependencies = [ "async-trait", "base64", @@ -2061,7 +2061,7 @@ dependencies = [ [[package]] name = "proton-sdk" -version = "0.1.8" +version = "0.1.9" dependencies = [ "aes-gcm 0.11.0", "async-trait", @@ -2074,6 +2074,7 @@ dependencies = [ "hex", "hkdf 0.13.0", "hmac 0.13.0", + "lru", "num-bigint", "num-traits", "pgp", diff --git a/Cargo.toml b/Cargo.toml index 111fb37..476c8b5 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -3,7 +3,7 @@ resolver = "2" members = ["crates/proton-sdk", "crates/proton-drive-rs"] [workspace.package] -version = "0.1.8" +version = "0.1.9" edition = "2024" license = "MIT" repository = "https://github.com/narrrl/proton-sdk-rs" diff --git a/crates/proton-drive-rs/Cargo.toml b/crates/proton-drive-rs/Cargo.toml index 8e5762f..a1510d0 100644 --- a/crates/proton-drive-rs/Cargo.toml +++ b/crates/proton-drive-rs/Cargo.toml @@ -14,7 +14,7 @@ readme = "../../README.md" documentation = "https://docs.rs/proton-drive-rs" [dependencies] -proton-sdk = { path = "../proton-sdk", version = "0.1.6" } +proton-sdk = { path = "../proton-sdk", version = "0.1.9" } tokio.workspace = true reqwest.workspace = true serde.workspace = true diff --git a/crates/proton-drive-rs/src/client.rs b/crates/proton-drive-rs/src/client.rs index 3df063d..c466778 100644 --- a/crates/proton-drive-rs/src/client.rs +++ b/crates/proton-drive-rs/src/client.rs @@ -5070,12 +5070,6 @@ fn read_full_block(reader: &mut R, buf: &mut [u8]) -> Result { Ok(filled) } -/// Group node uids by volume, preserving order, so each volume's links are -/// batched into a single request family (C# groups by `VolumeId`). -/// Whether a `v2/shares/my-files` error means the account has no My Files volume -/// yet (and so one must be created). C# catches [`ResponseCode::DoesNotExist`]; -/// in practice a fresh account's lookup also surfaces as a bare HTTP 404, so -/// both are treated as "missing". /// Whether a listed revision is one a history view should show. /// /// Only active (1) and superseded (2) qualify — TS @@ -5241,6 +5235,8 @@ fn join_node_path<'a>(names: impl Iterator) -> String { path } +/// Group node uids by volume, preserving order, so each volume's links are +/// batched into a single request family (C# groups by `VolumeId`). fn group_by_volume(uids: &[NodeUid]) -> Vec<(VolumeId, Vec)> { let mut groups: Vec<(VolumeId, Vec)> = Vec::new(); for uid in uids { diff --git a/crates/proton-sdk/Cargo.toml b/crates/proton-sdk/Cargo.toml index ce25e54..c8f53f7 100644 --- a/crates/proton-sdk/Cargo.toml +++ b/crates/proton-sdk/Cargo.toml @@ -14,6 +14,7 @@ readme = "../../README.md" documentation = "https://docs.rs/proton-sdk" [dependencies] +lru.workspace = true tokio.workspace = true reqwest.workspace = true serde.workspace = true diff --git a/crates/proton-sdk/src/cache.rs b/crates/proton-sdk/src/cache.rs index 6ced21f..58b7d16 100644 --- a/crates/proton-sdk/src/cache.rs +++ b/crates/proton-sdk/src/cache.rs @@ -12,7 +12,10 @@ //! layers. use std::collections::{HashMap, HashSet}; -use std::sync::{Arc, Mutex}; +use std::num::NonZeroUsize; +use std::sync::{Arc, Mutex, PoisonError}; + +use lru::LruCache; use aes_gcm::aead::{Aead, KeyInit, Payload}; use aes_gcm::{Aes256Gcm, Nonce}; @@ -56,31 +59,113 @@ pub async fn set_untagged(repo: &dyn CacheRepository, key: &str, value: &str) -> repo.set(key, value, &[]).await } +/// Default entry ceiling for [`InMemoryCacheRepository`]. +/// +/// Generous: eviction here is always safe — a miss re-fetches — so the cost of +/// being too small is a network round trip, while the cost of being unbounded +/// is unbounded. Sized to cover a working set of a few thousand nodes (a deep +/// browse, a sync pass) with room to spare, at a few MiB of JSON. +pub const DEFAULT_CACHE_CAPACITY: usize = 8192; + /// Thread-safe in-memory [`CacheRepository`]. Mirrors C# -/// `InMemoryCacheRepository`. -#[derive(Default)] +/// `InMemoryCacheRepository`, with an LRU bound the C# version does not have. +/// +/// The bound matters because the SDK's consumers are long-running: the entity +/// cache is written through on every node built, so an unbounded map retains +/// every node the process has ever seen. Entries are evicted least-recently-used +/// once [`DEFAULT_CACHE_CAPACITY`] (or the capacity passed to +/// [`with_capacity`](Self::with_capacity)) is reached. pub struct InMemoryCacheRepository { state: Mutex, } -#[derive(Default)] +impl Default for InMemoryCacheRepository { + fn default() -> Self { + Self::with_capacity(DEFAULT_CACHE_CAPACITY) + } +} + struct InMemoryState { - entries: HashMap, + /// Bounded, so this cache cannot become a leak. Every removal from here + /// must go through [`InMemoryCacheRepository::clear_tags_for_key`] — an + /// evicted key left behind in `tag_to_keys` would leak exactly what the + /// bound is meant to prevent, and would make `get_by_tags` name keys that + /// are no longer held. + entries: LruCache, key_to_tags: HashMap>, tag_to_keys: HashMap>, } impl InMemoryCacheRepository { - /// Create an empty in-memory cache. + /// Create an empty in-memory cache holding up to [`DEFAULT_CACHE_CAPACITY`] + /// entries. pub fn new() -> Self { Self::default() } + /// Create an empty in-memory cache holding up to `capacity` entries. A + /// `capacity` of zero is raised to one: a cache that can hold nothing would + /// turn every read into a fetch while still paying the bookkeeping. + pub fn with_capacity(capacity: usize) -> Self { + let capacity = NonZeroUsize::new(capacity).unwrap_or(NonZeroUsize::MIN); + Self { + state: Mutex::new(InMemoryState { + entries: LruCache::new(capacity), + key_to_tags: HashMap::new(), + tag_to_keys: HashMap::new(), + }), + } + } + /// Wrap a new in-memory cache in an [`Arc`] behind the trait object. pub fn shared() -> Arc { Arc::new(Self::new()) } + /// Wrap a new capacity-bounded in-memory cache behind the trait object. + pub fn shared_with_capacity(capacity: usize) -> Arc { + Arc::new(Self::with_capacity(capacity)) + } + + /// Number of entries currently held. Diagnostic — and what the bound test + /// asserts on. + pub fn len(&self) -> usize { + self.state().entries.len() + } + + /// Whether the cache holds nothing. + pub fn is_empty(&self) -> bool { + self.len() == 0 + } + + /// Number of keys the tag index knows about. Should track [`len`](Self::len); + /// a divergence is the leak this bound exists to prevent. + pub fn tag_index_len(&self) -> usize { + self.state().key_to_tags.len() + } + + /// Take the state lock, tolerating poisoning. + /// + /// Every access goes through here, and none of them use `.lock().unwrap()`. + /// That is the point: a `std::sync::Mutex` poisons when a thread panics + /// while holding it, and thereafter *every* acquisition panics — so one + /// panic anywhere near this cache would break it for the process lifetime. + /// + /// `pdfs-fuse`'s worker pool deliberately catches a panicking job and keeps + /// the worker alive, which assumes shared state survives a panic. Poisoning + /// would silently break that assumption across the crate boundary, turning a + /// recoverable EIO into a permanently broken client. + /// + /// Recovering the guard is sound here because the invariant this lock + /// protects — that `entries` and the two tag indexes agree — is restored by + /// the next write, and the worst case of reading a half-updated index is a + /// spurious cache miss. A miss re-fetches. There is nothing to corrupt. + /// + /// See `a_panic_holding_the_lock_does_not_break_the_cache`. + fn state(&self) -> std::sync::MutexGuard<'_, InMemoryState> { + self.state.lock().unwrap_or_else(PoisonError::into_inner) + } + fn clear_tags_for_key(state: &mut InMemoryState, key: &str) { if let Some(tags) = state.key_to_tags.remove(key) { for tag in tags { @@ -98,9 +183,16 @@ impl InMemoryCacheRepository { #[async_trait] impl CacheRepository for InMemoryCacheRepository { async fn set(&self, key: &str, value: &str, tags: &[String]) -> Result<()> { - let mut state = self.state.lock().unwrap(); + let mut state = self.state(); Self::clear_tags_for_key(&mut state, key); - state.entries.insert(key.to_owned(), value.to_owned()); + // `push` returns the entry evicted to make room, if any. Its tags must + // be dropped here — this is the only place the LRU discards a key on its + // own, so it is the only place the two indexes could drift apart. + if let Some((evicted, _)) = state.entries.push(key.to_owned(), value.to_owned()) + && evicted != *key + { + Self::clear_tags_for_key(&mut state, &evicted); + } let tag_set: HashSet = tags.iter().cloned().collect(); for tag in &tag_set { state @@ -114,33 +206,38 @@ impl CacheRepository for InMemoryCacheRepository { } async fn get(&self, key: &str) -> Result> { - let state = self.state.lock().unwrap(); + // `LruCache::get` promotes the entry to most-recently-used, which is why + // this takes the lock mutably: a read is use, and a cache that evicted + // the entry it is being asked for most often would be worse than none. + let mut state = self.state(); Ok(state.entries.get(key).cloned()) } async fn remove(&self, key: &str) -> Result<()> { - let mut state = self.state.lock().unwrap(); - state.entries.remove(key); + let mut state = self.state(); + state.entries.pop(key); Self::clear_tags_for_key(&mut state, key); Ok(()) } async fn remove_by_tag(&self, tag: &str) -> Result<()> { - let mut state = self.state.lock().unwrap(); + let mut state = self.state(); let keys: Vec = state .tag_to_keys .get(tag) .map(|keys| keys.iter().cloned().collect()) .unwrap_or_default(); for key in keys { - state.entries.remove(&key); + // The entry may already have been evicted; `pop` on an absent key is + // a no-op, so invalidation and eviction do not fight. + state.entries.pop(&key); Self::clear_tags_for_key(&mut state, &key); } Ok(()) } async fn clear(&self) -> Result<()> { - let mut state = self.state.lock().unwrap(); + let mut state = self.state(); state.entries.clear(); state.key_to_tags.clear(); state.tag_to_keys.clear(); @@ -151,7 +248,7 @@ impl CacheRepository for InMemoryCacheRepository { if tags.is_empty() { return Ok(Vec::new()); } - let state = self.state.lock().unwrap(); + let state = self.state(); let mut candidates: Option> = None; for tag in tags { match state.tag_to_keys.get(tag) { @@ -168,9 +265,12 @@ impl CacheRepository for InMemoryCacheRepository { } } let candidates = candidates.unwrap_or_default(); + // `peek` rather than `get`: a tag sweep is bookkeeping, not use, and + // promoting every tagged entry would let one bulk query reorder the + // whole cache. Ok(candidates .into_iter() - .filter_map(|key| state.entries.get(&key).map(|v| (key.clone(), v.clone()))) + .filter_map(|key| state.entries.peek(&key).map(|v| (key.clone(), v.clone()))) .collect()) } } @@ -388,6 +488,115 @@ mod tests { assert!(cache.get_by_tags(&tags(&["x"])).await.unwrap().is_empty()); } + /// **C8.** The entity cache is a cache, not a ledger. Every node the daemon + /// touches is written through to it as a JSON string, so an unbounded map + /// means a full tree walk of a large account is retained for the process + /// lifetime — the daemon is long-running, and nothing ever removes an entry + /// that is merely old. + /// + /// The tag indexes have to be bounded with it: an evicted entry whose key + /// stayed in `tag_to_keys` would leak just as surely, and would make + /// `get_by_tags` report keys that are no longer there. + #[tokio::test] + async fn in_memory_is_bounded() { + let cache = InMemoryCacheRepository::with_capacity(64); + for i in 0..10_000 { + cache + .set(&format!("node:{i}"), "{}", &tags(&["volume:1"])) + .await + .unwrap(); + } + assert_eq!( + cache.len(), + 64, + "entries bounded by the configured capacity" + ); + assert_eq!( + cache.tag_index_len(), + 64, + "the tag index is bounded with the entries, not left to grow" + ); + // The most recent writes survived; the oldest were evicted. + assert!(cache.get("node:9999").await.unwrap().is_some()); + assert!(cache.get("node:0").await.unwrap().is_none()); + // And a tag query reports exactly what is still held. + assert_eq!( + cache.get_by_tags(&tags(&["volume:1"])).await.unwrap().len(), + 64 + ); + } + + /// Eviction must not fight invalidation: removing a tag whose keys have + /// already been evicted is a no-op, not an error. The daemon's event loop + /// calls `remove_by_tag` on every server event. + #[tokio::test] + async fn evicted_entries_can_still_be_invalidated() { + let cache = InMemoryCacheRepository::with_capacity(4); + for i in 0..64 { + cache + .set(&format!("k{i}"), "v", &tags(&["share:1"])) + .await + .unwrap(); + } + cache.remove_by_tag("share:1").await.unwrap(); + assert_eq!(cache.len(), 0); + assert_eq!(cache.tag_index_len(), 0); + // Invalidating again, and invalidating a tag never seen, are both fine. + cache.remove_by_tag("share:1").await.unwrap(); + cache.remove_by_tag("share:never").await.unwrap(); + } + + /// A read counts as use. Without this the bound would be a FIFO: the node + /// being walked repeatedly would be evicted while a one-off lookup stayed. + #[tokio::test] + async fn reads_keep_an_entry_alive() { + let cache = InMemoryCacheRepository::with_capacity(2); + cache.set("a", "1", &[]).await.unwrap(); + cache.set("b", "2", &[]).await.unwrap(); + assert!(cache.get("a").await.unwrap().is_some()); // `b` is now the LRU + cache.set("c", "3", &[]).await.unwrap(); + assert!(cache.get("a").await.unwrap().is_some(), "recently read"); + assert!( + cache.get("b").await.unwrap().is_none(), + "least recently used" + ); + assert!(cache.get("c").await.unwrap().is_some(), "just written"); + } + + /// **The D9 reproduce.** A panic while the cache lock is held poisons it, + /// and every later acquisition panics for the process lifetime. + /// + /// This matters because `pdfs-fuse`'s worker pool `catch_unwind`s a + /// panicking job and keeps the worker alive — a design that assumes shared + /// state survives a panic. A poisoned lock turns one recoverable EIO into a + /// permanently broken client, which is exactly what the rescue exists to + /// prevent. + #[test] + fn a_panic_holding_the_lock_does_not_break_the_cache() { + let cache = Arc::new(InMemoryCacheRepository::with_capacity(8)); + + // Panic with the guard alive, as a worker would if it unwound inside a + // critical section. + let poisoner = { + let cache = cache.clone(); + std::thread::spawn(move || { + let _guard = cache.state.lock(); + panic!("worker panicked mid-cache-update"); + }) + }; + assert!(poisoner.join().is_err(), "the thread really did panic"); + + // The cache must still work. + assert_eq!(cache.len(), 0); + let rt = tokio::runtime::Builder::new_current_thread() + .build() + .unwrap(); + rt.block_on(async { + cache.set("k", "v", &[]).await.unwrap(); + assert_eq!(cache.get("k").await.unwrap().as_deref(), Some("v")); + }); + } + #[tokio::test] async fn encrypted_round_trips_and_hides_plaintext() { let inner = InMemoryCacheRepository::shared(); diff --git a/crates/proton-sdk/src/crypto/content.rs b/crates/proton-sdk/src/crypto/content.rs index 83d6fc8..bd182b3 100644 --- a/crates/proton-sdk/src/crypto/content.rs +++ b/crates/proton-sdk/src/crypto/content.rs @@ -484,15 +484,6 @@ impl PrivateKey { } impl PrivateKey { - /// Re-encrypt the key packet of `armored_message` — a PGP message currently - /// addressed to this key — to `destination`, preserving the encrypted data - /// packet (and thus the plaintext and any *detached* signature over it). - /// - /// This is the session-key rewrap a node **move** performs on a node - /// passphrase: the secret stays identical, only its recipient changes (old - /// parent → new parent), so the locked node key still unlocks and the node's - /// `NodePassphraseSignature` need not be reissued. Mirrors C# move's - /// `destinationKey.EncryptSessionKey(currentKey.DecryptSessionKey(passphrase))`. /// Recover the session key of an armored PGP message addressed to this key. /// /// De-armors the message and decrypts its leading PKESK to recover the @@ -510,6 +501,15 @@ impl PrivateKey { self.decrypt_content_key(&raw) } + /// Re-encrypt the key packet of `armored_message` — a PGP message currently + /// addressed to this key — to `destination`, preserving the encrypted data + /// packet (and thus the plaintext and any *detached* signature over it). + /// + /// This is the session-key rewrap a node **move** performs on a node + /// passphrase: the secret stays identical, only its recipient changes (old + /// parent → new parent), so the locked node key still unlocks and the node's + /// `NodePassphraseSignature` need not be reissued. Mirrors C# move's + /// `destinationKey.EncryptSessionKey(currentKey.DecryptSessionKey(passphrase))`. pub fn rewrap_message_to( &self, armored_message: &str, diff --git a/crates/proton-sdk/src/http.rs b/crates/proton-sdk/src/http.rs index 6fc379f..3a2bc78 100644 --- a/crates/proton-sdk/src/http.rs +++ b/crates/proton-sdk/src/http.rs @@ -439,12 +439,7 @@ struct SessionRefreshResponse { refresh_token: String, } -/// `POST {path}` without a session: no `x-pm-uid` and no bearer token. -/// -/// Used by the SRP login flow (`auth/v4/info`, `auth/v4`), which runs before a -/// session exists. Mirrors the C# SDK's `BeginAsync`, which issues these calls -/// on a session-less `HttpClient`. -/// Issue a GET with no session, for endpoints reachable before authentication. +/// `GET {path}` without a session: no `x-pm-uid` and no bearer token. /// /// The public-link flow needs this: `drive/urls/{token}/info` opens the SRP /// handshake and is by definition callable by a visitor who has no Proton @@ -476,6 +471,11 @@ pub async fn get_unauthenticated( parse_response(response).await } +/// `POST {path}` without a session: no `x-pm-uid` and no bearer token. +/// +/// Used by the SRP login flow (`auth/v4/info`, `auth/v4`), which runs before a +/// session exists. Mirrors the C# SDK's `BeginAsync`, which issues these calls +/// on a session-less `HttpClient`. pub async fn post_unauthenticated( config: &ProtonClientConfiguration, path: &str,