diff --git a/Cargo.lock b/Cargo.lock index f4ff7de..e7b3c27 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -88,6 +88,12 @@ dependencies = [ "aes 0.8.4", ] +[[package]] +name = "allocator-api2" +version = "0.2.21" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "683d7910e743518b0e34f1186f92494becacb047c7b6bf616c96772180fef923" + [[package]] name = "android_system_properties" version = "0.1.5" @@ -918,6 +924,12 @@ dependencies = [ "zeroize", ] +[[package]] +name = "equivalent" +version = "1.0.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "877a4ace8713b0bcf2a4e7eec82529c029f1d0619886d18145fea96c3ffe5c0f" + [[package]] name = "ff" version = "0.13.1" @@ -958,6 +970,12 @@ version = "1.0.7" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "3f9eec918d3f24069decb9af1554cad7c880e2da24a9afd88aca000531ab82c1" +[[package]] +name = "foldhash" +version = "0.1.5" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "d9c4f5dac5e15c24eb999c26181a6ca40b39fe946cbe4c263c7209467bc83af2" + [[package]] name = "form_urlencoded" version = "1.2.2" @@ -1147,6 +1165,17 @@ dependencies = [ "subtle", ] +[[package]] +name = "hashbrown" +version = "0.15.5" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "9229cfe53dfd69f0609a49f65461bd93001ea1ef889cd5529dd176593f5338a1" +dependencies = [ + "allocator-api2", + "equivalent", + "foldhash", +] + [[package]] name = "heck" version = "0.5.0" @@ -1600,6 +1629,15 @@ version = "0.4.33" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "0ceec5bc11778974d1bcb055b18002eba7f4b3518b6a0081b3af5f21666da9ad" +[[package]] +name = "lru" +version = "0.12.5" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "234cf4f4a04dc1f57e24b96cc0cd600cf2af460d4161ac5ecdd0af8e1f3b2a38" +dependencies = [ + "hashbrown", +] + [[package]] name = "lru-slab" version = "0.1.2" @@ -2007,6 +2045,7 @@ dependencies = [ "getrandom 0.4.3", "hex", "hmac 0.13.0", + "lru", "pgp", "proton-sdk", "reqwest", diff --git a/Cargo.toml b/Cargo.toml index 6c8f0c2..111fb37 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -22,6 +22,7 @@ serde_repr = "0.1" thiserror = "2" async-trait = "0.1" futures = "0.3" +lru = "0.12" bytes = "1" base64 = "0.22" hex = "0.4" diff --git a/crates/proton-drive-rs/Cargo.toml b/crates/proton-drive-rs/Cargo.toml index 4a152ae..8e5762f 100644 --- a/crates/proton-drive-rs/Cargo.toml +++ b/crates/proton-drive-rs/Cargo.toml @@ -23,6 +23,7 @@ serde_repr.workspace = true thiserror.workspace = true async-trait.workspace = true futures.workspace = true +lru.workspace = true bytes.workspace = true base64.workspace = true getrandom.workspace = true diff --git a/crates/proton-drive-rs/src/client.rs b/crates/proton-drive-rs/src/client.rs index 79758f9..ce7eb28 100644 --- a/crates/proton-drive-rs/src/client.rs +++ b/crates/proton-drive-rs/src/client.rs @@ -2,15 +2,18 @@ use std::collections::HashMap; use std::io::{Cursor, Read}; +use std::num::NonZeroUsize; use std::sync::Arc; use base64::Engine; use base64::engine::general_purpose::STANDARD as BASE64; use futures::stream::{self, StreamExt, TryStreamExt}; +use lru::LruCache; use sha2::{Digest, Sha256}; use tokio::sync::{Mutex, Semaphore}; use proton_sdk::account::{AccountClient, KeySalt}; +use proton_sdk::api::ResponseCode; use proton_sdk::cache::{CacheRepository, InMemoryCacheRepository}; use proton_sdk::crypto::PrivateKey; use proton_sdk::crypto::{ @@ -53,10 +56,10 @@ use crate::dtos::{ LinkDetailsRequest, LinkDetailsResponse, LinkDto, LinkType, ModulusResponse, MoveLinkRequest, MoveMultipleLinksItem, MoveMultipleLinksRequest, MultipleLinksRequest, MyFilesShareResponse, NodeNameAvailabilityRequest, NodeNameAvailabilityResponse, PhotosAttributesDto, - RenameLinkRequest, RevisionCreationRequest, RevisionCreationResponse, RevisionDto, - RevisionResponse, RevisionUpdateRequest, ShareInvitationDto, ShareInvitationsResponse, - ShareMembersResponse, ShareResponse, ShareTargetType, ShareUrlDto, ShareUrlsResponse, - SharedByMeResponse, SharedWithMeResponse, ThumbnailBlockListRequest, + RenameLinkRequest, RevisionConflict, RevisionCreationRequest, RevisionCreationResponse, + RevisionDto, RevisionResponse, RevisionUpdateRequest, ShareInvitationDto, + ShareInvitationsResponse, ShareMembersResponse, ShareResponse, ShareTargetType, ShareUrlDto, + ShareUrlsResponse, SharedByMeResponse, SharedWithMeResponse, ThumbnailBlockListRequest, ThumbnailBlockListResponse, ThumbnailCreationRequest, ThumbnailDto, TimelinePhotoListResponse, UpdatePermissionsRequest, VolumeCreationRequest, VolumeEventDto, VolumeEventListResponse, VolumeTrashResponse, @@ -169,7 +172,13 @@ struct Bootstrap { photos: Mutex<()>, } -#[derive(Default)] +/// Upper bound on decrypted folder node keys held in memory. Reached only by a +/// daemon that has navigated into more than this many distinct folders; eviction +/// is always safe (a missing key is re-derived from the parent chain or, for a +/// root, the share key), so this caps memory rather than changing behavior. See +/// SDK plan #9. +const FOLDER_KEY_CACHE_CAP: usize = 512; + struct DriveCache { main_volume_id: Option, my_files_share: Option, @@ -183,13 +192,34 @@ struct DriveCache { /// `v2/shares/photos` on every timeline page; `None` means "not yet checked". photos_volume_exists: Option, /// Decrypted node key per folder, used as the parent key for its children. - folder_keys: HashMap, + /// Bounded LRU ([`FOLDER_KEY_CACHE_CAP`]): a long-running daemon would + /// otherwise accumulate one entry per folder ever visited and never release + /// them. Entries are also dropped on the matching remote event + /// ([`ProtonDriveClient::invalidate_caches_for_event`]). + folder_keys: LruCache, /// The membership share behind each node another user shares with us, keyed /// by the shared node itself (which is that share's root link). `None` until /// [`ProtonDriveClient::shared_with_me_shares`] pages `v2/sharedwithme`. shared_with_me_shares: Option>, } +impl Default for DriveCache { + fn default() -> Self { + Self { + main_volume_id: None, + my_files_share: None, + my_files_root: None, + photos_share: None, + photos_root: None, + photos_volume_exists: None, + folder_keys: LruCache::new( + NonZeroUsize::new(FOLDER_KEY_CACHE_CAP).expect("cap is non-zero"), + ), + shared_with_me_shares: None, + } + } +} + /// An open revision draft ready to receive content blocks: the target revision /// plus the keys and signing identity needed to encrypt, sign and seal it. /// Produced by [`ProtonDriveClient::create_file_draft`] (new file) or @@ -360,8 +390,22 @@ impl ProtonDriveClient { } /// Fetch a single node's decrypted metadata, or `None` if it does not exist. + /// + /// Read-through: a node already in the entity cache is returned without the + /// link-details round-trip or the S2K decryption the cold path pays (SDK plan + /// #7). The cache is kept honest by event-driven invalidation + /// ([`invalidate_caches_for_event`](Self::invalidate_caches_for_event) drops + /// a node on its remote change), so a hit is as fresh as the last event poll + /// — the same staleness window a consumer's own tree has. Callers that must + /// bypass the cache to force a refresh use + /// [`enumerate_nodes`](Self::enumerate_nodes), which always hits the network. pub async fn get_node(&self, uid: &NodeUid) -> Result> { let mut timer = self.telemetry.start("get_node"); + if let Some(info) = self.entities.try_get_node(uid).await? { + timer.attr("cache", "hit"); + timer.success(); + return Ok(Some(info.node)); + } let response = self .get_link_details(&uid.volume_id, std::slice::from_ref(&uid.link_id)) .await?; @@ -615,6 +659,37 @@ impl ProtonDriveClient { Ok(events) } + /// Drop any cached state that a remote event may have invalidated. + /// + /// The in-memory [`DriveCache::folder_keys`] and the persistable + /// [`DriveEntityCache`] are populated on read and otherwise never expire, so + /// a node moved, renamed or re-keyed by another client would be served from a + /// stale entry forever (the leak/staleness called out as SDK plan #9). An + /// event consumer — e.g. [`EventManager`](crate::EventManager) — calls this + /// per event so those caches converge on the server. + /// + /// - `NodeUpdated` / `NodeDeleted`: forget the node's own folder key (its + /// passphrase may have been re-encrypted) and its entity-cache entry. The + /// parent's key is untouched — adding or removing a child does not re-key + /// the parent. + /// - `ContinuityLost` / `ScopeAccessLost`: the cursor gap means we cannot know + /// what changed, so drop every folder key and clear the entity cache. + /// - cursor-only / shared-with-me events: nothing cached to invalidate. + pub async fn invalidate_caches_for_event(&self, event: &DriveEvent) -> Result<()> { + match event { + DriveEvent::NodeUpdated { node_uid, .. } | DriveEvent::NodeDeleted { node_uid, .. } => { + self.cache.lock().await.folder_keys.pop(node_uid); + self.entities.remove_node(node_uid).await?; + } + DriveEvent::ContinuityLost { .. } | DriveEvent::ScopeAccessLost { .. } => { + self.cache.lock().await.folder_keys.clear(); + self.entities.clear().await?; + } + DriveEvent::CursorAdvanced { .. } | DriveEvent::SharedWithMeUpdated { .. } => {} + } + Ok(()) + } + /// Download and decrypt a file's active revision, returning its plaintext. pub async fn download_file(&self, uid: &NodeUid) -> Result> { let mut buf = Vec::new(); @@ -2855,7 +2930,7 @@ impl ProtonDriveClient { .lock() .await .folder_keys - .insert(uid.clone(), decrypted.node_key); + .put(uid.clone(), decrypted.node_key); Ok(decrypted.name) } @@ -2995,16 +3070,66 @@ impl ProtonDriveClient { let (address_id, email, signing_key) = self.membership_address().await?; // Create the revision draft, superseding the active revision. + let client_uid = self.http.session_id().to_string(); let request = RevisionCreationRequest { current_revision_id: active_revision_id, - client_uid: Some(self.http.session_id().to_string()), + client_uid: Some(client_uid.clone()), intended_upload_size, }; let path = format!( "v2/volumes/{volume_id}/files/{}/revisions", file_uid.link_id ); - let created: RevisionCreationResponse = self.http.post(&path, &request).await?; + + // An upload interrupted mid-flight (e.g. the daemon was killed) leaves a + // draft revision open on the link, and every later attempt 409s with + // `AlreadyExists` forever. When the abandoned draft is our own client's + // (the server echoes `ConflictDraftClientUID`), delete it and retry — + // never touch another client's in-progress draft. Mirrors C# + // `NewRevisionDraftProvider`. + const MAX_DRAFT_ATTEMPTS: usize = 3; + let mut created: Option = None; + for _ in 0..MAX_DRAFT_ATTEMPTS { + match self + .http + .post::<_, RevisionCreationResponse>(&path, &request) + .await + { + Ok(response) => { + created = Some(response); + break; + } + Err(ProtonError::Api(ref e)) if e.code == ResponseCode::AlreadyExists => { + let conflict = e + .details + .as_ref() + .and_then(|d| serde_json::from_value::(d.clone()).ok()); + let Some(RevisionConflict { + draft_revision_id: Some(draft_id), + draft_client_uid: Some(owner), + }) = conflict + else { + // No detail, or a draft we can't attribute — do not delete + // a draft that might be another client's live upload. + return Err(ProtonError::invalid_operation( + "revision draft already exists and cannot be recovered", + )); + }; + if owner != client_uid { + return Err(ProtonError::invalid_operation( + "revision draft is held by another client", + )); + } + tracing::warn!(%file_uid, draft_id, "deleting our stale draft revision, then retrying"); + self.delete_revision(&volume_id, &file_uid.link_id, &draft_id) + .await?; + } + Err(other) => return Err(other), + } + } + let created = created.ok_or_else(|| { + ProtonError::invalid_operation("revision draft creation kept conflicting after retries") + })?; Ok(RevisionDraft { volume_id, @@ -3019,6 +3144,21 @@ impl ProtonDriveClient { }) } + /// Delete a revision (`DELETE v2/volumes/{vid}/files/{lid}/revisions/{rid}`). + /// Used to clear an abandoned draft revision left by an interrupted upload; + /// the server permits a writer to delete its own drafts. Mirrors C# + /// `FilesApiClient.DeleteRevisionAsync`. + async fn delete_revision( + &self, + volume_id: &VolumeId, + link_id: &LinkId, + revision_id: &str, + ) -> Result<()> { + let path = format!("v2/volumes/{volume_id}/files/{link_id}/revisions/{revision_id}"); + let _: proton_sdk::api::ApiResponse = self.http.delete(&path).await?; + Ok(()) + } + /// Encrypt, sign, verify and upload every content block of a draft revision, /// accumulating the content manifest and extended-attribute metadata. /// @@ -3470,7 +3610,7 @@ impl ProtonDriveClient { let mut cache = self.cache.lock().await; cache.main_volume_id = Some(volume_id); cache.my_files_root = Some(root_uid.clone()); - cache.folder_keys.insert(root_uid, decrypted_root.node_key); + cache.folder_keys.put(root_uid, decrypted_root.node_key); cache.my_files_share = Some(ShareKey { share_id, address_id: response.share.address_id.clone(), @@ -3632,7 +3772,7 @@ impl ProtonDriveClient { let mut cache = self.cache.lock().await; cache.photos_root = Some(root_uid.clone()); - cache.folder_keys.insert(root_uid, decrypted_root.node_key); + cache.folder_keys.put(root_uid, decrypted_root.node_key); cache.photos_share = Some(ShareKey { share_id, address_id: response.share.address_id.clone(), @@ -3878,7 +4018,7 @@ impl ProtonDriveClient { .lock() .await .folder_keys - .insert(uid.clone(), decrypted.node_key.clone()); + .put(uid.clone(), decrypted.node_key.clone()); Ok(decrypted.node_key) } @@ -3950,7 +4090,7 @@ impl ProtonDriveClient { .lock() .await .folder_keys - .insert(uid, decrypted.node_key.clone()); + .put(uid, decrypted.node_key.clone()); key = decrypted.node_key; } @@ -4037,7 +4177,7 @@ impl ProtonDriveClient { .lock() .await .folder_keys - .insert(uid.clone(), node_key.clone()); + .put(uid.clone(), node_key.clone()); NodeKind::Folder } LinkType::File => { @@ -4508,10 +4648,21 @@ fn verification_token(code: &[u8], ciphertext: &[u8]) -> Vec { #[cfg(test)] mod tests { - use super::{DriveEvent, alternate_names, epoch_to_iso8601, to_drive_event}; + use super::{ + DriveCache, DriveEvent, FOLDER_KEY_CACHE_CAP, alternate_names, epoch_to_iso8601, + to_drive_event, + }; use crate::dtos::{VolumeEventDto, VolumeEventLinkDto}; use proton_sdk::ids::{DriveEventId, LinkId, VolumeId}; + #[test] + fn folder_key_cache_is_bounded() { + // The LRU that plan #9 introduced must actually carry the cap, or the + // daemon leaks one decrypted folder key per folder ever visited. + let cache = DriveCache::default(); + assert_eq!(cache.folder_keys.cap().get(), FOLDER_KEY_CACHE_CAP); + } + #[test] fn epoch_formats_as_iso8601_utc() { // 2026-06-26T14:01:00Z diff --git a/crates/proton-drive-rs/src/dtos.rs b/crates/proton-drive-rs/src/dtos.rs index 7c96c66..84d1d30 100644 --- a/crates/proton-drive-rs/src/dtos.rs +++ b/crates/proton-drive-rs/src/dtos.rs @@ -692,6 +692,17 @@ pub struct RevisionCreationIdentity { pub revision_id: String, } +/// `Details` object attached to an `AlreadyExists` (2500) error from +/// `POST .../revisions`: the server names the draft revision already open on the +/// link, and which client left it there. Mirrors C# `RevisionConflict`. +#[derive(Debug, Deserialize)] +pub struct RevisionConflict { + #[serde(rename = "ConflictDraftRevisionID", default)] + pub draft_revision_id: Option, + #[serde(rename = "ConflictDraftClientUID", default)] + pub draft_client_uid: Option, +} + /// `GET v2/volumes/{vid}/links/{lid}/revisions/{rid}/verification`. #[derive(Debug, Deserialize)] pub struct BlockVerificationInputResponse { diff --git a/crates/proton-drive-rs/src/event_manager.rs b/crates/proton-drive-rs/src/event_manager.rs new file mode 100644 index 0000000..c3ef89c --- /dev/null +++ b/crates/proton-drive-rs/src/event_manager.rs @@ -0,0 +1,277 @@ +//! Managed Drive event polling. +//! +//! A background loop that drains a volume's event cursor, invalidates the +//! client's caches, persists the cursor, and fans each event out to +//! subscribers. Mirrors the TS `internal/events` stack (`eventManager` + +//! `eventScheduler`): a per-scope poll loop at a foreground/background interval, +//! event→cache invalidation, and a subscription API. +//! +//! Before this, every consumer rolled its own loop (the `proton-drive-linux` +//! FUSE daemon has one) and none invalidated the client's own `folder_keys` / +//! entity cache — the staleness SDK plan #9 describes. Routing events through an +//! [`EventManager`] closes that gap: [`ProtonDriveClient::invalidate_caches_for_event`] +//! runs for every event before it reaches subscribers, so the SDK's caches +//! converge on the server without each consumer re-implementing it. + +use std::sync::Arc; +use std::sync::atomic::{AtomicBool, Ordering}; +use std::time::Duration; + +use async_trait::async_trait; +use proton_sdk::error::Result; +use proton_sdk::ids::DriveEventId; +use tokio::sync::broadcast; +use tracing::{debug, info, warn}; + +use crate::client::ProtonDriveClient; +use crate::events::{DriveEvent, DriveEventScopeId}; + +/// Foreground poll interval — the scope is actively in use. Matches TS +/// `FOREGROUND_POLLING_INTERVAL_SECONDS` (30s). +pub const DEFAULT_FOREGROUND_INTERVAL: Duration = Duration::from_secs(30); + +/// Background poll interval — the scope is idle. Matches TS +/// `BACKGROUND_POLLING_INTERVAL_SECONDS` (10 min). +pub const DEFAULT_BACKGROUND_INTERVAL: Duration = Duration::from_secs(10 * 60); + +/// Smallest wait before retrying a failed poll, doubled on each consecutive +/// failure up to the current poll interval. +const RETRY_MIN: Duration = Duration::from_secs(1); + +/// Persistence for the enumeration cursor. +/// +/// The SDK owns no storage, so a consumer supplies one: a daemon persists to +/// disk (the FUSE daemon uses SQLite) so a restart *resumes* from where it left +/// off rather than reseeding from the server head — reseeding would silently +/// skip everything that changed while the process was stopped. +#[async_trait] +pub trait CursorStore: Send + Sync { + /// The last persisted cursor, or `None` for a first-ever run. + async fn load(&self) -> Result>; + /// Persist `cursor` as the resume point for the next run. + async fn save(&self, cursor: &DriveEventId) -> Result<()>; +} + +/// In-memory [`CursorStore`]. Loses the cursor when dropped, so a process using +/// it reseeds from the server head on each start. Fine for tests and one-shot +/// consumers; a long-running daemon wants a persistent store instead. +#[derive(Default)] +pub struct MemoryCursorStore { + cursor: tokio::sync::Mutex>, +} + +#[async_trait] +impl CursorStore for MemoryCursorStore { + async fn load(&self) -> Result> { + Ok(self.cursor.lock().await.clone()) + } + async fn save(&self, cursor: &DriveEventId) -> Result<()> { + *self.cursor.lock().await = Some(cursor.clone()); + Ok(()) + } +} + +/// Tuning for an [`EventManager`]. +#[derive(Debug, Clone)] +pub struct EventManagerConfig { + /// Poll interval while the scope is in the foreground. + pub foreground_interval: Duration, + /// Poll interval while the scope is in the background. + pub background_interval: Duration, + /// Broadcast buffer size. A subscriber that falls this far behind receives a + /// [`broadcast::error::RecvError::Lagged`] and should treat it like a + /// continuity loss — resync from current server state. + pub channel_capacity: usize, +} + +impl Default for EventManagerConfig { + fn default() -> Self { + Self { + foreground_interval: DEFAULT_FOREGROUND_INTERVAL, + background_interval: DEFAULT_BACKGROUND_INTERVAL, + channel_capacity: 256, + } + } +} + +/// Polls one event scope, invalidates the client's caches, and publishes events +/// to subscribers. +/// +/// Single-scope by design (the volume the caller cares about — typically their +/// own My Files volume). A caller that tracks several scopes runs one manager +/// each; the foreground/background switch then lets a shared scope poll lazily +/// until the user opens it. +pub struct EventManager { + client: ProtonDriveClient, + scope: DriveEventScopeId, + store: Arc, + config: EventManagerConfig, + events_tx: broadcast::Sender, + foreground: Arc, +} + +impl EventManager { + /// Build a manager for `scope`, persisting its cursor through `store`, with + /// default intervals. Starts in the foreground. + pub fn new( + client: ProtonDriveClient, + scope: DriveEventScopeId, + store: Arc, + ) -> Self { + Self::with_config(client, scope, store, EventManagerConfig::default()) + } + + /// As [`new`](Self::new), with explicit tuning. + pub fn with_config( + client: ProtonDriveClient, + scope: DriveEventScopeId, + store: Arc, + config: EventManagerConfig, + ) -> Self { + let (events_tx, _) = broadcast::channel(config.channel_capacity.max(1)); + Self { + client, + scope, + store, + config, + events_tx, + foreground: Arc::new(AtomicBool::new(true)), + } + } + + /// A receiver for the event feed. Each subscriber sees every event applied + /// after it subscribed. Cache invalidation already happened in the manager, + /// so a subscriber only needs to update *its own* state (e.g. a UI or FUSE + /// tree). + pub fn subscribe(&self) -> broadcast::Receiver { + self.events_tx.subscribe() + } + + /// Switch the scope between the foreground (frequent) and background (lazy) + /// poll interval. Takes effect on the next poll cycle. + pub fn set_foreground(&self, foreground: bool) { + self.foreground.store(foreground, Ordering::Relaxed); + } + + /// Whether the scope is currently polled at the foreground interval. + pub fn is_foreground(&self) -> bool { + self.foreground.load(Ordering::Relaxed) + } + + fn current_interval(&self) -> Duration { + if self.is_foreground() { + self.config.foreground_interval + } else { + self.config.background_interval + } + } + + /// Load the resume cursor, seeding from the server head on a first-ever run. + /// + /// Seeding needs the network and a manager may start while offline, so a + /// seed failure is retried with backoff rather than giving up (which would + /// disable sync for the life of the process). A persisted cursor resumes + /// immediately without a network call. + async fn resolve_start_cursor(&self) -> Option { + match self.store.load().await { + Ok(Some(cursor)) => return Some(cursor), + Ok(None) => {} + Err(e) => warn!(error = %e, "load event cursor failed; seeding from head"), + } + + let mut delay = RETRY_MIN; + loop { + match self.client.enumerate_events(&self.scope, None).await { + Ok(events) => { + let head = events.last().map(|e| e.id().clone()); + if let Some(cursor) = &head + && let Err(e) = self.store.save(cursor).await + { + warn!(error = %e, "persist seed cursor failed"); + } + return head; + } + Err(e) => { + warn!(error = %e, ?delay, "seed event cursor failed; retrying"); + tokio::time::sleep(delay).await; + delay = (delay * 2).min(self.config.background_interval); + } + } + } + } + + /// Poll the scope forever: at each interval, drain events from the cursor, + /// invalidate the client's caches, publish each event, and persist the new + /// cursor. Returns only if the cursor cannot be resolved at all. + /// + /// Runs as a task; hold the manager in an `Arc` and spawn a clone: + /// `let m = mgr.clone(); tokio::spawn(async move { m.run().await });`. + pub async fn run(&self) { + let mut cursor = self.resolve_start_cursor().await; + info!(scope = %self.scope, ?cursor, "event manager started"); + + let mut retry = RETRY_MIN; + loop { + tokio::time::sleep(self.current_interval()).await; + + let events = match self + .client + .enumerate_events(&self.scope, cursor.as_ref()) + .await + { + Ok(events) => events, + Err(e) => { + warn!(error = %e, ?retry, "event poll failed; backing off"); + tokio::time::sleep(retry).await; + retry = (retry * 2).min(self.current_interval()); + continue; + } + }; + retry = RETRY_MIN; + + if events.is_empty() { + continue; + } + debug!(count = events.len(), "applying remote events"); + for event in &events { + if let Err(e) = self.client.invalidate_caches_for_event(event).await { + warn!(error = %e, "cache invalidation for event failed"); + } + // A send with no live receivers returns Err; that is normal (the + // cache invalidation above is the load-bearing effect), so ignore + // it rather than treat "nobody is listening" as a failure. + let _ = self.events_tx.send(event.clone()); + } + + cursor = events.last().map(|e| e.id().clone()); + if let Some(cursor) = &cursor + && let Err(e) = self.store.save(cursor).await + { + warn!(error = %e, "persist event cursor failed"); + } + } + } +} + +#[cfg(test)] +mod tests { + use super::*; + + #[tokio::test] + async fn memory_cursor_store_roundtrips() { + let store = MemoryCursorStore::default(); + assert!(store.load().await.unwrap().is_none()); + + let id = DriveEventId::new("evt-42"); + store.save(&id).await.unwrap(); + assert_eq!(store.load().await.unwrap(), Some(id)); + } + + #[test] + fn config_defaults_match_ts_scheduler() { + let config = EventManagerConfig::default(); + assert_eq!(config.foreground_interval, Duration::from_secs(30)); + assert_eq!(config.background_interval, Duration::from_secs(600)); + assert!(config.channel_capacity > 0); + } +} diff --git a/crates/proton-drive-rs/src/lib.rs b/crates/proton-drive-rs/src/lib.rs index d0ec7fa..1f2bc5a 100644 --- a/crates/proton-drive-rs/src/lib.rs +++ b/crates/proton-drive-rs/src/lib.rs @@ -37,6 +37,7 @@ mod client; mod crypto; mod devices; mod dtos; +mod event_manager; mod events; mod node; mod photos; @@ -46,6 +47,10 @@ mod sharing; pub use cache::{CachedNodeInfo, DriveEntityCache}; pub use client::ProtonDriveClient; pub use devices::{Device, DeviceType}; +pub use event_manager::{ + CursorStore, DEFAULT_BACKGROUND_INTERVAL, DEFAULT_FOREGROUND_INTERVAL, EventManager, + EventManagerConfig, MemoryCursorStore, +}; pub use events::{DriveEvent, DriveEventScopeId}; pub use node::{Node, NodeKind, RevisionState, Thumbnail, ThumbnailType}; pub use photos::{PhotoTag, PhotoUploadMetadata, PhotosTimelineItem, ProtonPhotosClient}; diff --git a/crates/proton-sdk/src/api.rs b/crates/proton-sdk/src/api.rs index fdf8c9b..c062989 100644 --- a/crates/proton-sdk/src/api.rs +++ b/crates/proton-sdk/src/api.rs @@ -13,6 +13,12 @@ pub struct ApiResponse { #[serde(rename = "Error", default)] pub error_message: Option, + + /// Endpoint-specific error detail object, when the API attaches one (e.g. a + /// revision-creation conflict names the existing draft here). Kept raw so + /// each caller can deserialize the shape it expects. + #[serde(rename = "Details", default)] + pub details: Option, } impl ApiResponse { diff --git a/crates/proton-sdk/src/config.rs b/crates/proton-sdk/src/config.rs index 0459447..e38c2e6 100644 --- a/crates/proton-sdk/src/config.rs +++ b/crates/proton-sdk/src/config.rs @@ -9,8 +9,17 @@ pub const DEFAULT_BASE_URL: &str = "https://drive-api.proton.me/"; pub const DEFAULT_REFRESH_REDIRECT_URI: &str = "https://proton.me"; /// Default per-request timeout, matching `ProtonApiDefaults.DefaultTimeoutSeconds`. +/// Applies to the JSON API calls (link details, listings, revision creation) — +/// small requests where 30s is generous and a stall should fail fast. pub const DEFAULT_TIMEOUT: Duration = Duration::from_secs(30); +/// Default timeout for a single block-storage transfer (`get_storage_blob` / +/// `post_storage_blob`). A block is up to 4 MiB and rides a different host from +/// the API; on a slow uplink that transfer legitimately takes far longer than a +/// JSON call, so it gets its own, much larger budget rather than tripping the +/// 30s API timeout spuriously. +pub const DEFAULT_STORAGE_TIMEOUT: Duration = Duration::from_secs(300); + /// Default number of automatic retries on a retryable failure (rate-limit, /// transient transport error). The original attempt is not counted, so the /// total request budget is `1 + max_retries`. @@ -74,6 +83,10 @@ pub struct ProtonClientConfiguration { pub user_agent: String, pub refresh_redirect_uri: String, pub request_timeout: Duration, + /// Timeout for a single block-storage transfer. Separate from + /// `request_timeout` so a large block on a slow link does not fail against + /// the (short) API timeout. See [`DEFAULT_STORAGE_TIMEOUT`]. + pub storage_timeout: Duration, pub retry_policy: RetryPolicy, } @@ -86,6 +99,7 @@ impl ProtonClientConfiguration { user_agent: String::new(), refresh_redirect_uri: DEFAULT_REFRESH_REDIRECT_URI.to_owned(), request_timeout: DEFAULT_TIMEOUT, + storage_timeout: DEFAULT_STORAGE_TIMEOUT, retry_policy: RetryPolicy::default(), } } @@ -105,6 +119,11 @@ impl ProtonClientConfiguration { self } + pub fn with_storage_timeout(mut self, timeout: Duration) -> Self { + self.storage_timeout = timeout; + self + } + pub fn with_retry_policy(mut self, retry_policy: RetryPolicy) -> Self { self.retry_policy = retry_policy; self diff --git a/crates/proton-sdk/src/error.rs b/crates/proton-sdk/src/error.rs index b462470..321ed3b 100644 --- a/crates/proton-sdk/src/error.rs +++ b/crates/proton-sdk/src/error.rs @@ -44,6 +44,9 @@ pub struct ProtonApiError { pub http_status: u16, /// Human-readable message from the `Error` field, if present. pub message: String, + /// Raw `Details` object from the error envelope, when present. Endpoint + /// specific — e.g. a revision-creation conflict names the existing draft. + pub details: Option, } impl ProtonApiError { diff --git a/crates/proton-sdk/src/http.rs b/crates/proton-sdk/src/http.rs index c5801f6..6e22897 100644 --- a/crates/proton-sdk/src/http.rs +++ b/crates/proton-sdk/src/http.rs @@ -165,7 +165,14 @@ impl ApiHttpClient { pub async fn get_storage_blob(&self, url: &str, token: &str) -> Result { let mut timer = self.telemetry().start("storage_download"); let response = send_retrying(&self.inner.config.retry_policy, || { - let mut request = self.inner.http.get(url).header(STORAGE_TOKEN_HEADER, token); + // Override the client-level (API) timeout: a 4 MiB block on a slow + // uplink legitimately outruns the 30s JSON budget. + let mut request = self + .inner + .http + .get(url) + .timeout(self.inner.config.storage_timeout) + .header(STORAGE_TOKEN_HEADER, token); if !self.inner.config.user_agent.is_empty() { request = request.header(reqwest::header::USER_AGENT, &self.inner.config.user_agent); @@ -216,6 +223,7 @@ impl ApiHttpClient { .inner .http .post(url) + .timeout(self.inner.config.storage_timeout) .header(STORAGE_TOKEN_HEADER, token) .multipart(form); @@ -492,6 +500,7 @@ fn api_error(status: StatusCode, bytes: &[u8]) -> ProtonError { .as_ref() .map(|e| e.code) .unwrap_or(ResponseCode::Unknown); + let details = envelope.as_ref().and_then(|e| e.details.clone()); let message = envelope.and_then(|e| e.error_message).unwrap_or_else(|| { status .canonical_reason() @@ -503,6 +512,7 @@ fn api_error(status: StatusCode, bytes: &[u8]) -> ProtonError { code, http_status: status.as_u16(), message, + details, }) }