event manager, concurrency, optimizations

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",

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"

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

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,
+    RenameLinkRequest, RevisionConflict, RevisionCreationRequest, RevisionCreationResponse,
-    RevisionResponse, RevisionUpdateRequest, ShareInvitationDto, ShareInvitationsResponse,
+    RevisionDto, RevisionResponse, RevisionUpdateRequest, ShareInvitationDto,
-    ShareMembersResponse, ShareResponse, ShareTargetType, ShareUrlDto, ShareUrlsResponse,
+    ShareInvitationsResponse, ShareMembersResponse, ShareResponse, ShareTargetType, ShareUrlDto,
-    SharedByMeResponse, SharedWithMeResponse, ThumbnailBlockListRequest,
+    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<VolumeId>,
     my_files_share: Option<ShareKey>,
@@ -183,13 +192,34 @@ struct DriveCache {
     /// `v2/shares/photos` on every timeline page; `None` means "not yet checked".
     photos_volume_exists: Option<bool>,
     /// Decrypted node key per folder, used as the parent key for its children.
-    folder_keys: HashMap<NodeUid, PrivateKey>,
+    /// 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<NodeUid, PrivateKey>,
     /// 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<HashMap<NodeUid, ShareId>>,
 }
 
+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<Option<Node>> {
         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<Vec<u8>> {
         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<RevisionCreationResponse> = 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::<RevisionConflict>(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<u8> {
 
 #[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

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<String>,
+    #[serde(rename = "ConflictDraftClientUID", default)]
+    pub draft_client_uid: Option<String>,
+}
+
 /// `GET v2/volumes/{vid}/links/{lid}/revisions/{rid}/verification`.
 #[derive(Debug, Deserialize)]
 pub struct BlockVerificationInputResponse {

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<Option<DriveEventId>>;
+    /// 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<Option<DriveEventId>>,
+}
+
+#[async_trait]
+impl CursorStore for MemoryCursorStore {
+    async fn load(&self) -> Result<Option<DriveEventId>> {
+        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<dyn CursorStore>,
+    config: EventManagerConfig,
+    events_tx: broadcast::Sender<DriveEvent>,
+    foreground: Arc<AtomicBool>,
+}
+
+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<dyn CursorStore>,
+    ) -> 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<dyn CursorStore>,
+        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<DriveEvent> {
+        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<DriveEventId> {
+        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);
+    }
+}

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};

crates/proton-sdk/src/api.rs

@@ -13,6 +13,12 @@ pub struct ApiResponse {
 
     #[serde(rename = "Error", default)]
     pub error_message: Option<String>,
+
+    /// 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<serde_json::Value>,
 }
 
 impl ApiResponse {

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

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<serde_json::Value>,
 }
 
 impl ProtonApiError {

crates/proton-sdk/src/http.rs

@@ -165,7 +165,14 @@ impl ApiHttpClient {
     pub async fn get_storage_blob(&self, url: &str, token: &str) -> Result<Bytes> {
         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,
     })
 }