version 0.4.0

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.3.2" }
+proton-sdk = { path = "../proton-sdk", version = "0.4.0" }
 tokio.workspace = true
 reqwest.workspace = true
 serde.workspace = true

crates/proton-drive-rs/src/client.rs

@@ -63,13 +63,13 @@ use crate::dtos::{
     NodeNameAvailabilityRequest, NodeNameAvailabilityResponse, PhotoTagsRequest,
     PhotosAttributesDto, RenameLinkRequest, RevisionConflict, RevisionCreationRequest,
     RevisionCreationResponse, RevisionDto, RevisionListItemDto, RevisionListResponse,
-    RevisionMetadataResponse, RevisionResponse, RevisionUpdateRequest, ShareInvitationDto,
-    ShareInvitationsResponse, ShareMembersResponse, ShareMembershipSummaryDto, ShareResponse,
-    ShareTargetType, ShareUrlDto, ShareUrlsResponse, SharedAlbumsResponse, SharedByMeResponse,
-    SharedWithMeResponse, SmallFileUploadMetadataRequest, SmallRevisionUploadMetadataRequest,
-    SmallUploadResponse, ThumbnailBlockListRequest, ThumbnailBlockListResponse,
-    ThumbnailCreationRequest, ThumbnailDto, TimelinePhotoListResponse, UpdatePermissionsRequest,
-    VolumeCreationRequest, VolumeEventDto, VolumeEventListResponse, VolumeTrashResponse,
+    RevisionMetadataResponse, RevisionUpdateRequest, ShareInvitationDto, ShareInvitationsResponse,
+    ShareMembersResponse, ShareMembershipSummaryDto, ShareResponse, ShareTargetType, ShareUrlDto,
+    ShareUrlsResponse, SharedAlbumsResponse, SharedByMeResponse, SharedWithMeResponse,
+    SmallFileUploadMetadataRequest, SmallRevisionUploadMetadataRequest, SmallUploadResponse,
+    ThumbnailBlockListRequest, ThumbnailBlockListResponse, ThumbnailCreationRequest, ThumbnailDto,
+    TimelinePhotoListResponse, UpdatePermissionsRequest, VolumeCreationRequest, VolumeEventDto,
+    VolumeEventListResponse, VolumeTrashResponse,
 };
 use crate::events::{DriveEvent, DriveEventScopeId};
 use crate::node::{
@@ -87,19 +87,21 @@ use crate::sharing::{
     PublicLink, ShareInvitation, ShareMember, ShareMembership, SharedWithMeItem,
 };
 use crate::single_flight::SingleFlight;
-
-/// Content blocks are 4 MiB of plaintext each (C# `RevisionWriter.DefaultBlockSize`).
-const DEFAULT_BLOCK_SIZE: usize = 1 << 22;
+use crate::transport::{DEFAULT_BLOCK_SIZE, RevisionTransport, rank_block_sizes};
 
 /// Maximum encrypted multipart size accepted by the atomic upload endpoint.
 const SMALL_UPLOAD_SIZE_LIMIT: usize = 128 * 1024;
 
 /// Maximum links per batch trash/restore/delete request (C#
 /// `NodeOperations.MaximumBatchCount`).
-const MAX_BATCH_COUNT: usize = 150;
+pub(crate) const MAX_BATCH_COUNT: usize = 150;
+
+/// Thumbnail block ids resolved per `POST volumes/{vid}/thumbnails` request
+/// (C# `FileOperations.MaxThumbnailIdsPerRequest`).
+pub(crate) const MAX_THUMBNAIL_IDS_PER_REQUEST: usize = 30;
 
 /// Link-detail batches fetched at once, per volume, when enumerating nodes.
-const MAX_CONCURRENT_DETAIL_FETCHES: usize = 4;
+pub(crate) const MAX_CONCURRENT_DETAIL_FETCHES: usize = 4;
 
 /// Nodes decrypted at once within one batch of link details.
 ///
@@ -121,7 +123,7 @@ const MAX_CONCURRENT_NODE_BUILDS: usize = 8;
 /// Sized a little above one file's pipeline (an upload holds up to
 /// `MAX_BUFFERED_UPLOAD_BLOCKS + MAX_CONCURRENT_BLOCK_UPLOADS + 1` permits) so a
 /// lone transfer still saturates it and a second one is not starved.
-const DEFAULT_MAX_INFLIGHT_BLOCKS: usize = 16;
+pub(crate) const DEFAULT_MAX_INFLIGHT_BLOCKS: usize = 16;
 
 /// Content blocks uploaded concurrently within a single file.
 ///
@@ -515,13 +517,6 @@ impl ProtonDriveClient {
     }
 
     /// The account client backing this Drive client's key chain.
-    /// The underlying HTTP client, for sibling modules that issue their own
-    /// requests against the same session (e.g. [`RevisionReader`] fetching
-    /// block bodies from storage).
-    pub(crate) fn http(&self) -> &ApiHttpClient {
-        &self.http
-    }
-
     pub fn account(&self) -> &AccountClient {
         &self.account
     }
@@ -587,6 +582,16 @@ impl ProtonDriveClient {
         self.block_slots.clone()
     }
 
+    /// What a [`RevisionReader`] needs from this client, and nothing else.
+    ///
+    /// The session is `Static` because an ordinary bearer session refreshes its
+    /// own tokens inside [`ApiHttpClient`] on a 401; by the time an error
+    /// reaches the transport there is nothing left to retry. The visitor path
+    /// supplies a session that can genuinely be renewed.
+    pub(crate) fn revision_transport(&self) -> RevisionTransport {
+        RevisionTransport::authenticated(self.http.clone(), self.block_slots())
+    }
+
     /// Resolve (and cache) the user's "My Files" root folder.
     pub async fn get_my_files_folder(&self) -> Result<Node> {
         self.ensure_my_files().await?;
@@ -1361,7 +1366,7 @@ impl ProtonDriveClient {
 
         timer.success();
         Ok(RevisionReader::new(
-            self.clone(),
+            self.revision_transport(),
             uid.clone(),
             revision_id,
             content_key,
@@ -1639,26 +1644,11 @@ impl ProtonDriveClient {
 
     /// Plaintext size of each content block, in block order.
     ///
-    /// These sizes are not merely descriptive: [`RevisionReader`] accumulates
-    /// them to derive where each block *starts* in the plaintext. A vector that
-    /// overstates any block but the last therefore shifts every block after it,
-    /// and the reader splices those blocks at the wrong offsets — returning
-    /// full-length reads of the wrong bytes, with no error to notice. See
-    /// `revision.rs::padded_block_sizes_shift_later_blocks`, which pins that
-    /// consequence.
-    ///
-    /// So the sources are ranked by how much they can be trusted, and the
-    /// function refuses rather than guesses when none of them holds:
-    ///
-    /// 1. `Common.BlockSizes` from the revision's extended attributes — the
-    ///    authoritative value, written by the uploading client.
-    /// 2. `Common.Size` — the total, from which all-but-last full blocks and a
-    ///    final short block follow by subtraction. Sound for the standard
-    ///    layout the uploader produces.
-    /// 3. Nothing usable. A single-block file cannot be misaligned (there is no
-    ///    later block to shift), so it is assumed full-size and only its
-    ///    reported length can be wrong. A multi-block file is an error: serving
-    ///    it would mean serving misplaced bytes.
+    /// Decrypts the revision's extended attributes — *verified*, which is what
+    /// needs an account client and is therefore why this lives here rather than
+    /// in `transport` — and hands them to
+    /// [`rank_block_sizes`](crate::transport::rank_block_sizes), which owns the
+    /// ranking rules and the refusal.
     async fn resolve_block_sizes(
         &self,
         node_key: &PrivateKey,
@@ -1687,40 +1677,7 @@ impl ProtonDriveClient {
             None => None,
         };
 
-        let block = DEFAULT_BLOCK_SIZE as u64;
-
-        if let Some(sizes) = common.as_ref().and_then(|c| c.block_sizes.as_ref())
-            && sizes.len() == block_count
-        {
-            // A non-positive entry is as corrupting as a padded one — a
-            // zero-length block in the middle shifts everything after it — so a
-            // malformed vector is rejected outright rather than clamped.
-            if let Some(bad) = sizes.iter().position(|&n| n <= 0) {
-                return Err(ProtonError::invalid_operation(format!(
-                    "revision {} records a non-positive size for block {bad}",
-                    revision.id
-                )));
-            }
-            return Ok(sizes.iter().map(|&n| n as u64).collect());
-        }
-
-        if let Some(total) = common.and_then(|c| c.size).filter(|&n| n >= 0) {
-            let total = total as u64;
-            return Ok((0..block_count)
-                .map(|i| total.saturating_sub(block * i as u64).min(block))
-                .collect());
-        }
-
-        if block_count <= 1 {
-            return Ok(vec![block; block_count]);
-        }
-
-        Err(ProtonError::invalid_operation(format!(
-            "revision {} has {block_count} blocks but no usable size information \
-             (no BlockSizes, no Size); refusing to guess, since a wrong block \
-             layout would serve misaligned content",
-            revision.id
-        )))
+        rank_block_sizes(common.as_ref(), &revision.id, block_count)
     }
 
     /// Download and decrypt a file's thumbnail of the given type, if it has one.
@@ -1814,8 +1771,6 @@ impl ProtonDriveClient {
         thumbnail_type: ThumbnailType,
         for_photos: bool,
     ) -> Result<Vec<FileThumbnail>> {
-        const MAX_THUMBNAIL_IDS_PER_REQUEST: usize = 30;
-
         let mut results: Vec<FileThumbnail> = Vec::new();
 
         // Group link ids by volume, preserving first-seen volume order.
@@ -5087,50 +5042,21 @@ impl ProtonDriveClient {
         Ok(())
     }
 
+    /// The revision's metadata and its full, ordered block table.
+    ///
+    /// The paging, ordering and contiguity rules are identical on the visitor
+    /// path, so they live in
+    /// [`RevisionTransport::list_blocks`](crate::transport::RevisionTransport)
+    /// and this is the authenticated entry point to them.
     pub(crate) async fn fetch_revision_blocks(
         &self,
         volume_id: &VolumeId,
         link_id: &LinkId,
         revision_id: &str,
     ) -> Result<(RevisionDto, Vec<BlockDto>)> {
-        const PAGE_SIZE: i32 = 50;
-
-        let mut blocks: Vec<BlockDto> = Vec::new();
-        let mut metadata: Option<RevisionDto> = None;
-        let mut from_index: i32 = 1;
-
-        loop {
-            let path = format!(
-                "v2/volumes/{volume_id}/files/{link_id}/revisions/{revision_id}?FromBlockIndex={from_index}&PageSize={PAGE_SIZE}&NoBlockUrls=0"
-            );
-            let response: RevisionResponse = self.http.get(&path).await?;
-            let mut revision = response.revision;
-            let page = std::mem::take(&mut revision.blocks);
-            let page_len = page.len();
-
-            if metadata.is_none() {
-                metadata = Some(revision);
-            }
-            blocks.extend(page);
-
-            if page_len < PAGE_SIZE as usize {
-                break;
-            }
-            from_index = blocks.iter().map(|b| b.index).max().unwrap_or(from_index) + 1;
-        }
-
-        blocks.sort_by_key(|b| b.index);
-        for (offset, block) in blocks.iter().enumerate() {
-            if block.index != offset as i32 + 1 {
-                return Err(ProtonError::invalid_operation(
-                    "file contents are incomplete (non-contiguous blocks)",
-                ));
-            }
-        }
-
-        let metadata = metadata
-            .ok_or_else(|| ProtonError::invalid_operation("revision returned no metadata"))?;
-        Ok((metadata, blocks))
+        self.revision_transport()
+            .list_blocks(volume_id, link_id, revision_id)
+            .await
     }
 
     async fn ensure_my_files(&self) -> Result<()> {

crates/proton-drive-rs/src/crypto.rs

@@ -249,19 +249,34 @@ pub async fn decrypt_extended_attributes_verified(
     Ok((attrs, status))
 }
 
+/// Decrypt a revision's extended attributes without verifying their signature.
+///
+/// For callers that have the node key but no [`AccountClient`] to resolve the
+/// author's public key through — namely an anonymous public-link visitor, who
+/// has no address book and cannot look up a signer at all.
+///
+/// Skipping verification loses nothing a visitor could otherwise have used:
+/// [`decrypt_extended_attributes_verified`] returns the status as *metadata* and
+/// callers do not gate on it (see the SDK's rule that a failed verification is
+/// never an error). Decryption still proves the payload was encrypted to this
+/// node key.
+pub(crate) fn decrypt_extended_attributes_unverified(
+    node_key: &PrivateKey,
+    armored_xattr: &str,
+) -> Result<DecryptedExtendedAttributes> {
+    let json = node_key.decrypt_armored_message(armored_xattr)?;
+    serde_json::from_slice(&json).map_err(|e| {
+        ProtonError::invalid_operation(format!("deserialize extended attributes: {e}"))
+    })
+}
+
 #[cfg(test)]
 mod tests {
     use super::*;
     use proton_sdk::crypto::generate_node_key;
 
-    /// Decrypt + deserialize XAttr without verification (the production path
-    /// uses [`decrypt_extended_attributes_verified`], which needs an account to
-    /// resolve author keys).
     fn decrypt_xattr(node_key: &PrivateKey, armored: &str) -> DecryptedExtendedAttributes {
-        let json = node_key
-            .decrypt_armored_message(armored)
-            .expect("decrypt xattr");
-        serde_json::from_slice(&json).expect("deserialize xattr")
+        decrypt_extended_attributes_unverified(node_key, armored).expect("decrypt xattr")
     }
 
     #[test]

crates/proton-drive-rs/src/lib.rs

@@ -45,6 +45,7 @@ mod public_link;
 mod revision;
 mod sharing;
 mod single_flight;
+mod transport;
 
 pub use cache::{CachedNodeInfo, DriveEntityCache};
 pub use client::ProtonDriveClient;
@@ -55,7 +56,8 @@ pub use event_manager::{
 };
 pub use events::{DriveEvent, DriveEventScopeId};
 pub use node::{
-    AlbumProperties, Node, NodeKind, PhotoProperties, RevisionState, Thumbnail, ThumbnailType,
+    AlbumProperties, FileThumbnail, Node, NodeKind, PhotoProperties, RevisionState, Thumbnail,
+    ThumbnailType,
 };
 pub use photos::{
     AlbumItem, PhotoTag, PhotoTagsUpdate, PhotoUploadMetadata, PhotosTimelineItem,

crates/proton-drive-rs/src/public_link.rs

@@ -29,9 +29,15 @@
 //! claimed, rather than a verification that silently passed.
 
 use std::collections::HashMap;
+use std::num::NonZeroUsize;
+use std::sync::Arc;
 
+use async_trait::async_trait;
 use base64::Engine;
 use base64::engine::general_purpose::STANDARD as BASE64;
+use futures::stream::{self, StreamExt, TryStreamExt};
+use lru::LruCache;
+use tokio::sync::Semaphore;
 
 use proton_sdk::config::ProtonClientConfiguration;
 use proton_sdk::crypto::{
@@ -42,15 +48,21 @@ use proton_sdk::error::{ProtonError, Result};
 use proton_sdk::http::{ApiHttpClient, Tokens, get_unauthenticated};
 use proton_sdk::ids::{LinkId, NodeUid, SessionId, VolumeId};
 
-use crate::client::parse_public_link_url;
-use crate::crypto::decrypt_link;
+use crate::client::{
+    DEFAULT_MAX_INFLIGHT_BLOCKS, MAX_BATCH_COUNT, MAX_CONCURRENT_DETAIL_FETCHES,
+    MAX_THUMBNAIL_IDS_PER_REQUEST, parse_public_link_url,
+};
+use crate::crypto::{decrypt_extended_attributes_unverified, decrypt_link};
 use crate::dtos::{
     FileDto, FolderChildrenResponse, LinkDetailsDto, LinkDetailsRequest, LinkDetailsResponse,
     LinkType, PublicLinkAuthRequest, PublicLinkAuthResponse, PublicLinkInfoResponse,
-    RevisionResponse,
+    ThumbnailBlockListRequest, ThumbnailBlockListResponse,
 };
-use crate::node::{Node, NodeKind, RevisionState};
+use crate::node::{FileThumbnail, Node, NodeKind, RevisionState, ThumbnailType};
+use crate::revision::{MAX_CONCURRENT_BLOCK_DOWNLOADS, RevisionReader, decrypt_block_blocking};
 use crate::sharing::MemberRole;
+use crate::single_flight::SingleFlight;
+use crate::transport::{BlockSession, RevisionTransport, rank_block_sizes};
 
 /// Link metadata readable before authenticating, from `GET urls/{token}/info`.
 ///
@@ -69,11 +81,13 @@ pub struct PublicLinkInfo {
     pub vendor_type: i32,
 }
 
-/// An authenticated, anonymous session against one public link.
+/// Everything needed to run the handshake, and nothing that a completed
+/// handshake produced.
 ///
-/// Mirrors TS `SharingPublicLinkSession`. Holds the link password so an expired
-/// session can be re-established by replaying the handshake.
-struct PublicLinkSession {
+/// Split out from [`PublicLinkSession`] so a session can be *constructed* around
+/// an already-authenticated client while still being able to replay the
+/// handshake later — the credentials outlive any particular session.
+struct Credentials {
     config: ProtonClientConfiguration,
     token: String,
     /// URL fragment password concatenated with any custom password — the TS SDK
@@ -89,7 +103,7 @@ struct AuthOutcome {
     public_role: MemberRole,
 }
 
-impl PublicLinkSession {
+impl Credentials {
     async fn info(&self) -> Result<PublicLinkInfoResponse> {
         get_unauthenticated(&self.config, &format!("drive/urls/{}/info", self.token)).await
     }
@@ -159,7 +173,12 @@ impl PublicLinkSession {
         // scope. Sending them to `drive/` returns 403 `MissingScopes`.
         // The session routes (`drive/urls/{token}/…`) are exempt — they run
         // unauthenticated, before this client exists.
-        .with_base_route("drive/unauth/");
+        .with_base_route("drive/unauth/")
+        // The refresh token above is empty, so `ApiHttpClient`'s built-in 401
+        // handling would POST it to `auth/v4/refresh` and replace the real 401
+        // with an error about refreshing. Opt out, and let the 401 reach
+        // `RevisionTransport`, which recovers by replaying *this* handshake.
+        .without_token_refresh();
 
         Ok(AuthOutcome {
             http,
@@ -190,19 +209,133 @@ impl PublicLinkSession {
     }
 }
 
+/// The client a public-link session is currently issuing requests through,
+/// tagged with the handshake that produced it.
+struct LiveSession {
+    http: ApiHttpClient,
+    generation: u64,
+}
+
+/// An authenticated, anonymous session against one public link.
+///
+/// Mirrors TS `SharingPublicLinkSession`. Holds the link password so an expired
+/// session can be re-established by replaying the handshake — which matters more
+/// here than anywhere else in the SDK, because the anonymous session is minted
+/// with an **empty refresh token** and so cannot renew itself the ordinary way.
+/// A long read (a film, say) will outlive it.
+///
+/// The live client is swapped behind a lock rather than handed back to the
+/// caller as a new object, so every [`RevisionReader`](crate::RevisionReader)
+/// already opened over this session keeps working across the swap. Swapping the
+/// *whole* client is required, not just its tokens: a re-auth mints a new
+/// `x-pm-uid` as well as a new bearer, and an [`ApiHttpClient`] binds one
+/// session id for its lifetime.
+struct PublicLinkSession {
+    credentials: Credentials,
+    /// Read on every request, written only by a handshake. A `std` lock because
+    /// [`BlockSession::http`] is synchronous; it is never held across an await,
+    /// and never poisoned in practice, so poisoning is recovered from rather
+    /// than propagated.
+    live: std::sync::RwLock<LiveSession>,
+    /// Serializes handshakes, so a burst of expired-session failures replays one
+    /// SRP exchange rather than one per caller.
+    renewing: tokio::sync::Mutex<()>,
+}
+
+impl PublicLinkSession {
+    /// Authenticate and wrap the result in a renewable session.
+    ///
+    /// Returns the outcome alongside the session because the share key, root uid
+    /// and role belong to the *link*, not to the session, and the client keeps
+    /// them directly.
+    async fn open(credentials: Credentials) -> Result<(Arc<Self>, AuthOutcome)> {
+        let auth = credentials.auth().await?;
+        let session = Arc::new(Self {
+            credentials,
+            live: std::sync::RwLock::new(LiveSession {
+                http: auth.http.clone(),
+                generation: 0,
+            }),
+            renewing: tokio::sync::Mutex::new(()),
+        });
+        Ok((session, auth))
+    }
+
+    fn read_live<T>(&self, f: impl FnOnce(&LiveSession) -> T) -> T {
+        let live = self
+            .live
+            .read()
+            .unwrap_or_else(std::sync::PoisonError::into_inner);
+        f(&live)
+    }
+}
+
+#[async_trait]
+impl BlockSession for PublicLinkSession {
+    fn http(&self) -> (ApiHttpClient, u64) {
+        self.read_live(|live| (live.http.clone(), live.generation))
+    }
+
+    async fn renew(&self, seen: u64) -> Result<()> {
+        let _guard = self.renewing.lock().await;
+
+        if self.read_live(|live| live.generation) != seen {
+            // Someone else re-authenticated while we waited; their session is at
+            // least as fresh as one we would mint now.
+            return Ok(());
+        }
+
+        let auth = self.credentials.auth().await?;
+
+        let mut live = self
+            .live
+            .write()
+            .unwrap_or_else(std::sync::PoisonError::into_inner);
+        live.http = auth.http;
+        live.generation += 1;
+        Ok(())
+    }
+}
+
 /// Read access to one public share link, as a visitor.
 ///
 /// Obtained from [`open`](Self::open). The client holds an anonymous session and
 /// the link's share key; every operation is scoped to the shared subtree.
+///
+/// Cloning is cheap and shares the session, so a clone recovers from an expiry
+/// the original recovered from, and vice versa.
+#[derive(Clone)]
 pub struct ProtonDrivePublicLinkClient {
-    session: PublicLinkSession,
-    http: ApiHttpClient,
+    session: Arc<PublicLinkSession>,
+    /// Every request goes through here rather than through a bare
+    /// [`ApiHttpClient`], so an expired anonymous session is replayed once and
+    /// retried instead of surfacing as a 401.
+    transport: RevisionTransport,
     /// The link's share key — the parent key of the shared root node.
     share_key: PrivateKey,
     root_uid: NodeUid,
     public_role: MemberRole,
+    /// Unlocked node keys, so a folder's children do not each re-derive their
+    /// shared parent's key.
+    ///
+    /// Unlocking a key is an S2K derivation — tens of milliseconds. Listing a
+    /// folder of 500 episodes walked the ancestor chain and re-unlocked every
+    /// key on it *per child*, which is the difference between a catalog crawl
+    /// taking seconds and taking minutes.
+    node_keys: Arc<std::sync::Mutex<LruCache<NodeUid, PrivateKey>>>,
+    /// Collapses concurrent misses for the same key onto one load, so the
+    /// fanned-out detail fetches below do not each start the same ancestor walk
+    /// before any of them has populated the cache.
+    key_loads: Arc<SingleFlight<NodeUid, PrivateKey>>,
 }
 
+/// Unlocked node keys held per client.
+///
+/// A shared subtree's *distinct* keys are its folders, so this only has to
+/// outlive a crawl of one directory's worth of ancestors; it is a bound on
+/// pathological trees, not a working-set estimate.
+const NODE_KEY_CACHE_CAPACITY: usize = 512;
+
 impl ProtonDrivePublicLinkClient {
     /// Read a link's public metadata without opening it.
     ///
@@ -211,12 +344,12 @@ impl ProtonDrivePublicLinkClient {
     /// its `#password` fragment.
     pub async fn info(config: ProtonClientConfiguration, url: &str) -> Result<PublicLinkInfo> {
         let (token, password) = parse_public_link_url(url)?;
-        let session = PublicLinkSession {
+        let credentials = Credentials {
             config,
             token,
             password,
         };
-        let raw = session.info().await?;
+        let raw = credentials.info().await?;
         Ok(PublicLinkInfo {
             is_custom_password_protected: raw.is_custom_password_protected(),
             is_legacy: raw.is_legacy(),
@@ -242,22 +375,50 @@ impl ProtonDrivePublicLinkClient {
             _ => url_password,
         };
 
-        let session = PublicLinkSession {
+        let (session, auth) = PublicLinkSession::open(Credentials {
             config,
             token,
             password,
-        };
-        let auth = session.auth().await?;
+        })
+        .await?;
+
+        let transport = RevisionTransport::new(
+            session.clone(),
+            Arc::new(Semaphore::new(DEFAULT_MAX_INFLIGHT_BLOCKS)),
+        );
 
         Ok(Self {
             session,
-            http: auth.http,
+            transport,
             share_key: auth.share_key,
             root_uid: auth.root_uid,
             public_role: auth.public_role,
+            node_keys: Arc::new(std::sync::Mutex::new(LruCache::new(
+                NonZeroUsize::new(NODE_KEY_CACHE_CAPACITY).expect("capacity is non-zero"),
+            ))),
+            key_loads: Arc::new(SingleFlight::default()),
         })
     }
 
+    /// Cap the content blocks this client keeps in memory at once, across every
+    /// read it is running.
+    ///
+    /// Mirrors
+    /// [`ProtonDriveClient::with_max_inflight_blocks`](crate::ProtonDriveClient::with_max_inflight_blocks).
+    /// The default is sized for a background sync; a player that seeks wants it
+    /// higher. Must be non-zero; a zero cap would deadlock every read, so it is
+    /// clamped to 1.
+    ///
+    /// Applies to this client and every clone made from it *afterwards*; clones
+    /// already taken keep the previous cap.
+    pub fn with_max_inflight_blocks(mut self, blocks: usize) -> Self {
+        self.transport = RevisionTransport::new(
+            self.session.clone(),
+            Arc::new(Semaphore::new(blocks.max(1))),
+        );
+        self
+    }
+
     /// The role the link grants its visitors — [`MemberRole::Viewer`] for a
     /// read-only link, [`MemberRole::Editor`] for one that permits uploads.
     pub fn public_role(&self) -> MemberRole {
@@ -310,7 +471,7 @@ impl ProtonDrivePublicLinkClient {
                 path.push_str(&format!("?AnchorID={anchor_id}"));
             }
 
-            let page: FolderChildrenResponse = self.http.get(&path).await?;
+            let page: FolderChildrenResponse = self.transport.get(&path).await?;
             if page.link_ids.is_empty() {
                 break;
             }
@@ -334,6 +495,10 @@ impl ProtonDrivePublicLinkClient {
     /// A node that does not exist is omitted; one that fails to decrypt is
     /// logged and skipped, so the result may be shorter than `uids` — the same
     /// partial-listing behavior as the authenticated client.
+    ///
+    /// Link ids are batched at [`MAX_BATCH_COUNT`] per request — the server's
+    /// limit, which an unchunked call silently exceeded on any folder past 150
+    /// children — and the batches are fetched several at a time.
     pub async fn enumerate_nodes(&self, uids: &[NodeUid]) -> Result<Vec<Node>> {
         if uids.is_empty() {
             return Ok(Vec::new());
@@ -341,28 +506,39 @@ impl ProtonDrivePublicLinkClient {
 
         let mut nodes = Vec::with_capacity(uids.len());
         for (volume_id, link_ids) in group_by_volume(uids) {
-            let response: LinkDetailsResponse = self
-                .http
-                .post(
-                    &format!("v2/volumes/{volume_id}/links"),
-                    &LinkDetailsRequest {
-                        link_ids: &link_ids,
-                    },
-                )
-                .await?;
+            // The closure takes its batch *by value*. Taking `&[LinkId]` — which
+            // `chunks()` hands out directly — would give the fetch future a
+            // higher-ranked lifetime that `tokio::spawn` rejects in callers
+            // ("implementation of `FnOnce` is not general enough"), an error that
+            // surfaces only downstream. `tests/spawnable.rs` pins it.
+            let chunks: Vec<Vec<LinkId>> = link_ids
+                .chunks(MAX_BATCH_COUNT)
+                .map(<[LinkId]>::to_vec)
+                .collect();
+            let batches = chunks.into_iter().map(|link_ids| {
+                let client = self.clone();
+                let volume_id = volume_id.clone();
+                async move { client.fetch_link_details(&volume_id, &link_ids).await }
+            });
 
-            for details in &response.links {
-                let parent_key = match self.resolve_parent_key(&volume_id, details).await {
-                    Ok(key) => key,
-                    Err(e) => {
-                        tracing::warn!(link_id = %details.link.id, "skipping node: {e}");
-                        continue;
-                    }
-                };
-                match self.build_node(&volume_id, details, &parent_key) {
-                    Ok(node) => nodes.push(node),
-                    Err(e) => {
-                        tracing::warn!(link_id = %details.link.id, "skipping node: {e}")
+            // `buffered` yields in request order, so a listing keeps the order
+            // its uids were given in.
+            let mut responses = stream::iter(batches).buffered(MAX_CONCURRENT_DETAIL_FETCHES);
+
+            while let Some(response) = responses.try_next().await? {
+                for details in &response.links {
+                    let parent_key = match self.resolve_parent_key(&volume_id, details).await {
+                        Ok(key) => key,
+                        Err(e) => {
+                            tracing::warn!(link_id = %details.link.id, "skipping node: {e}");
+                            continue;
+                        }
+                    };
+                    match self.build_node(&volume_id, details, &parent_key) {
+                        Ok(node) => nodes.push(node),
+                        Err(e) => {
+                            tracing::warn!(link_id = %details.link.id, "skipping node: {e}")
+                        }
                     }
                 }
             }
@@ -371,6 +547,21 @@ impl ProtonDrivePublicLinkClient {
         Ok(nodes)
     }
 
+    /// One batch of link details. Callers must keep `link_ids` within
+    /// [`MAX_BATCH_COUNT`].
+    async fn fetch_link_details(
+        &self,
+        volume_id: &VolumeId,
+        link_ids: &[LinkId],
+    ) -> Result<LinkDetailsResponse> {
+        self.transport
+            .post(
+                &format!("v2/volumes/{volume_id}/links"),
+                &LinkDetailsRequest { link_ids },
+            )
+            .await
+    }
+
     /// Download a shared file's active revision.
     ///
     /// Unlike the authenticated client's
@@ -404,24 +595,35 @@ impl ProtonDrivePublicLinkClient {
         let decrypted = decrypt_link(&parent_key, &details.link)?;
         let content_key = self.content_key(&details, &decrypted.node_key)?;
 
-        let response: RevisionResponse = self
-            .http
-            .get(&format!(
-                "v2/volumes/{}/files/{}/revisions/{}",
-                uid.volume_id, uid.link_id, revision_id
-            ))
+        // Paged: a single un-paged GET returns only the first 50 blocks, which
+        // silently truncated any file past 200 MiB.
+        let (_, blocks) = self
+            .transport
+            .list_blocks(&uid.volume_id, &uid.link_id, &revision_id)
             .await?;
 
-        // Blocks arrive in index order; decrypt each in turn straight into the
-        // writer so a large file never has to be resident in full.
-        let mut blocks = response.revision.blocks;
-        blocks.sort_by_key(|block| block.index);
-        for block in blocks {
-            let ciphertext = self
-                .http
-                .get_storage_blob(&block.bare_url, &block.token)
-                .await?;
-            let plaintext = content_key.decrypt_block(&ciphertext)?;
+        // Fetched and decrypted several at a time but yielded in block order, so
+        // the writer still sees the file sequentially and no more than
+        // `MAX_CONCURRENT_BLOCK_DOWNLOADS` blocks are ever resident.
+        let mut plaintexts = stream::iter(blocks.into_iter().map(|block| {
+            let transport = self.transport.clone();
+            let content_key = content_key.clone();
+            async move {
+                let permit = transport.block_slots().acquire_owned().await.map_err(|e| {
+                    ProtonError::invalid_operation(format!("block slots closed: {e}"))
+                })?;
+                let ciphertext = transport
+                    .http()
+                    .get_storage_blob(&block.bare_url, &block.token)
+                    .await?;
+                let plaintext = decrypt_block_blocking(content_key, ciphertext).await?;
+                Ok::<_, ProtonError>((plaintext, permit))
+            }
+        }))
+        .buffered(MAX_CONCURRENT_BLOCK_DOWNLOADS);
+
+        // `_permit` frees once the block is written, not when it is decrypted.
+        while let Some((plaintext, _permit)) = plaintexts.try_next().await? {
             writer
                 .write_all(&plaintext)
                 .map_err(|e| ProtonError::invalid_operation(format!("write block: {e}")))?;
@@ -430,32 +632,347 @@ impl ProtonDrivePublicLinkClient {
         Ok(())
     }
 
+    /// Download and decrypt only the plaintext byte range
+    /// `[offset, offset + length)` of a shared file's active revision.
+    ///
+    /// A one-shot convenience over [`open_revision`](Self::open_revision), which
+    /// callers issuing more than one read should hold instead — this repeats the
+    /// whole resolution (link details, ancestor walk, node-key unlock, revision
+    /// listing) every time.
+    pub async fn download_range(&self, uid: &NodeUid, offset: u64, length: u64) -> Result<Vec<u8>> {
+        self.open_revision(uid).await?.read_at(offset, length).await
+    }
+
+    /// Open the active revision of a shared file for seekable reads.
+    ///
+    /// This is what makes a public link streamable: every content block is a
+    /// self-contained PGP packet under one session key, so a
+    /// [`RevisionReader`] can serve any byte range by fetching only the blocks
+    /// that overlap it, several at a time — no whole-file download, no
+    /// sequential decrypt.
+    ///
+    /// The reader holds this client's session and renews it on expiry, so it
+    /// keeps working past the anonymous session's lifetime.
+    ///
+    /// The reader is pinned to the revision that is active now; it does not
+    /// follow later revisions of the same file.
+    pub async fn open_revision(&self, uid: &NodeUid) -> Result<RevisionReader> {
+        self.open_revision_inner(uid, None).await
+    }
+
+    /// As [`open_revision`](Self::open_revision), on an explicit revision.
+    ///
+    /// A visitor cannot enumerate a file's revision history, so `revision_id`
+    /// has to come from somewhere that already knew it —
+    /// [`NodeKind::File::active_revision_id`](crate::NodeKind) recorded earlier,
+    /// typically, to reopen the same bytes a previous session was reading.
+    pub async fn open_revision_at(
+        &self,
+        uid: &NodeUid,
+        revision_id: &str,
+    ) -> Result<RevisionReader> {
+        self.open_revision_inner(uid, Some(revision_id)).await
+    }
+
+    async fn open_revision_inner(
+        &self,
+        uid: &NodeUid,
+        revision_id: Option<&str>,
+    ) -> Result<RevisionReader> {
+        let details = self.link_details(uid).await?;
+        let active = file_properties(&details).and_then(|file| file.active_revision.as_ref());
+
+        let revision_id = match revision_id {
+            Some(id) => id.to_string(),
+            None => active.map(|revision| revision.id.clone()).ok_or_else(|| {
+                ProtonError::invalid_operation("the shared file has no active revision")
+            })?,
+        };
+
+        let parent_key = self.resolve_parent_key(&uid.volume_id, &details).await?;
+        let node_key = decrypt_link(&parent_key, &details.link)?.node_key;
+        let content_key = self.content_key(&details, &node_key)?;
+
+        let (revision, blocks) = self
+            .transport
+            .list_blocks(&uid.volume_id, &uid.link_id, &revision_id)
+            .await?;
+
+        // Prefer the revision's own extended attributes. Fall back to the copy
+        // the link-details response above already delivered for the *active*
+        // revision — the visitor route has a history of stripping fields the
+        // authenticated one returns, and without a usable size the reader cannot
+        // place block boundaries and refuses to open a multi-block file at all.
+        let armored = revision.extended_attributes.as_deref().or_else(|| {
+            active
+                .filter(|candidate| candidate.id == revision_id)
+                .and_then(|candidate| candidate.extended_attributes.as_deref())
+        });
+
+        let common = armored.and_then(|xattr| {
+            match decrypt_extended_attributes_unverified(&node_key, xattr) {
+                Ok(attrs) => attrs.common,
+                Err(e) => {
+                    tracing::debug!(error = %e, "revision extended attributes did not decrypt");
+                    None
+                }
+            }
+        });
+
+        let block_sizes = rank_block_sizes(common.as_ref(), &revision_id, blocks.len())?;
+
+        Ok(RevisionReader::new(
+            self.transport.clone(),
+            uid.clone(),
+            revision_id,
+            content_key,
+            blocks,
+            block_sizes,
+        ))
+    }
+
+    /// Download and decrypt a shared file's thumbnail of the given type, if it
+    /// has one.
+    ///
+    /// Ports the authenticated
+    /// [`download_thumbnail`](crate::ProtonDriveClient::download_thumbnail): the
+    /// revision's thumbnail header names a block id, `POST volumes/{vid}/thumbnails`
+    /// resolves it to a download URL, and the bytes decrypt under the same
+    /// content key as the file's content blocks. Only the route prefix differs.
+    /// Returns `Ok(None)` when the file has no thumbnail of that type.
+    pub async fn download_thumbnail(
+        &self,
+        uid: &NodeUid,
+        thumbnail_type: ThumbnailType,
+    ) -> Result<Option<Vec<u8>>> {
+        let (content_key, thumbnail_id) = match self.thumbnail_target(uid, thumbnail_type).await? {
+            (content_key, Some(id)) => (content_key, id),
+            (_, None) => return Ok(None),
+        };
+
+        let response: ThumbnailBlockListResponse = self
+            .transport
+            .post(
+                &format!("volumes/{}/thumbnails", uid.volume_id),
+                &ThumbnailBlockListRequest {
+                    thumbnail_ids: vec![thumbnail_id.clone()],
+                },
+            )
+            .await?;
+
+        let block = response
+            .blocks
+            .into_iter()
+            .find(|block| block.thumbnail_id == thumbnail_id)
+            .ok_or_else(|| {
+                let reason = response
+                    .errors
+                    .iter()
+                    .find(|e| e.thumbnail_id == thumbnail_id)
+                    .map(|e| e.error.clone())
+                    .unwrap_or_else(|| "thumbnail block not returned".to_string());
+                ProtonError::invalid_operation(format!("resolve thumbnail block: {reason}"))
+            })?;
+
+        let ciphertext = self
+            .transport
+            .http()
+            .get_storage_blob(&block.bare_url, &block.token)
+            .await?;
+        Ok(Some(content_key.decrypt_thumbnail(&ciphertext)?))
+    }
+
+    /// Batch-download the thumbnails of `uids`.
+    ///
+    /// The shape a poster wall wants: block ids are resolved in batches of up to
+    /// [`MAX_THUMBNAIL_IDS_PER_REQUEST`], and a per-file failure — no thumbnail
+    /// of that type, a decrypt error, a node that is not a file — is reported in
+    /// that file's [`FileThumbnail`] instead of failing the batch. Returned
+    /// order does not track the input order.
+    pub async fn enumerate_thumbnails(
+        &self,
+        uids: &[NodeUid],
+        thumbnail_type: ThumbnailType,
+    ) -> Result<Vec<FileThumbnail>> {
+        let mut results: Vec<FileThumbnail> = Vec::new();
+
+        for (volume_id, link_ids) in group_by_volume(uids) {
+            // thumbnail id -> the file it belongs to and the key it decrypts under.
+            let mut targets: HashMap<String, (NodeUid, ContentKey)> = HashMap::new();
+
+            for link_id in link_ids {
+                let uid = NodeUid::new(volume_id.clone(), link_id);
+                match self.thumbnail_target(&uid, thumbnail_type).await {
+                    Ok((content_key, Some(id))) => {
+                        targets.insert(id, (uid, content_key));
+                    }
+                    Ok((_, None)) => results.push(FileThumbnail::err(
+                        uid.clone(),
+                        ProtonError::invalid_operation(format!(
+                            "node {uid} has no thumbnail of the requested type"
+                        )),
+                    )),
+                    Err(e) => results.push(FileThumbnail::err(uid, e)),
+                }
+            }
+
+            let thumbnail_ids: Vec<String> = targets.keys().cloned().collect();
+            for chunk in thumbnail_ids.chunks(MAX_THUMBNAIL_IDS_PER_REQUEST) {
+                self.drain_thumbnail_chunk(&volume_id, chunk, &mut targets, &mut results)
+                    .await;
+            }
+        }
+
+        Ok(results)
+    }
+
+    /// Resolve one chunk of thumbnail ids to blocks, download them, and record
+    /// an outcome for every id in the chunk — including the ones the server said
+    /// nothing about at all.
+    async fn drain_thumbnail_chunk(
+        &self,
+        volume_id: &VolumeId,
+        chunk: &[String],
+        targets: &mut HashMap<String, (NodeUid, ContentKey)>,
+        results: &mut Vec<FileThumbnail>,
+    ) {
+        let response: ThumbnailBlockListResponse = match self
+            .transport
+            .post(
+                &format!("volumes/{volume_id}/thumbnails"),
+                &ThumbnailBlockListRequest {
+                    thumbnail_ids: chunk.to_vec(),
+                },
+            )
+            .await
+        {
+            Ok(response) => response,
+            // The chunk request itself failed: every file in it gets that error.
+            Err(e) => {
+                for id in chunk {
+                    if let Some((uid, _)) = targets.remove(id) {
+                        results.push(FileThumbnail::err(
+                            uid,
+                            ProtonError::invalid_operation(format!(
+                                "resolve thumbnail blocks: {e}"
+                            )),
+                        ));
+                    }
+                }
+                return;
+            }
+        };
+
+        for block in response.blocks {
+            let Some((uid, content_key)) = targets.remove(&block.thumbnail_id) else {
+                continue;
+            };
+            let downloaded = match self
+                .transport
+                .http()
+                .get_storage_blob(&block.bare_url, &block.token)
+                .await
+            {
+                Ok(ciphertext) => content_key
+                    .decrypt_thumbnail(&ciphertext)
+                    .map_err(ProtonError::from),
+                Err(e) => Err(e),
+            };
+            results.push(match downloaded {
+                Ok(bytes) => FileThumbnail::ok(uid, bytes),
+                Err(e) => FileThumbnail::err(uid, e),
+            });
+        }
+
+        for error in response.errors {
+            if let Some((uid, _)) = targets.remove(&error.thumbnail_id) {
+                results.push(FileThumbnail::err(
+                    uid,
+                    ProtonError::invalid_operation(error.error),
+                ));
+            }
+        }
+
+        // Anything still in `targets` was neither returned nor errored on. Left
+        // unreported it would look like a file that was never asked about.
+        for id in chunk {
+            if let Some((uid, _)) = targets.remove(id) {
+                results.push(FileThumbnail::err(
+                    uid,
+                    ProtonError::invalid_operation("thumbnail not found".to_string()),
+                ));
+            }
+        }
+    }
+
+    /// A file's content key and the block id of its thumbnail of
+    /// `thumbnail_type`, if it has one.
+    async fn thumbnail_target(
+        &self,
+        uid: &NodeUid,
+        thumbnail_type: ThumbnailType,
+    ) -> Result<(ContentKey, Option<String>)> {
+        let details = self.link_details(uid).await?;
+        let revision_id = file_properties(&details)
+            .and_then(|file| file.active_revision.as_ref())
+            .map(|revision| revision.id.clone())
+            .ok_or_else(|| {
+                ProtonError::invalid_operation("the shared file has no active revision")
+            })?;
+
+        let parent_key = self.resolve_parent_key(&uid.volume_id, &details).await?;
+        let node_key = decrypt_link(&parent_key, &details.link)?.node_key;
+        let content_key = self.content_key(&details, &node_key)?;
+
+        // The thumbnail headers ride along with the revision's block listing.
+        let (revision, _blocks) = self
+            .transport
+            .list_blocks(&uid.volume_id, &uid.link_id, &revision_id)
+            .await?;
+        let wanted = thumbnail_type.as_i32();
+        let thumbnail_id = revision
+            .thumbnails
+            .iter()
+            .find(|thumbnail| thumbnail.thumbnail_type == wanted)
+            .and_then(|thumbnail| thumbnail.id.clone());
+
+        Ok((content_key, thumbnail_id))
+    }
+
     /// Re-run the handshake, replacing the session behind this client.
     ///
     /// The anonymous session carries no refresh token, so an expiry is recovered
     /// from by authenticating again (TS `SharingPublicLinkSession.reauth`).
-    /// Returns a fresh client rather than mutating this one, so an in-flight
-    /// caller never observes a half-swapped session.
+    ///
+    /// Rarely needed directly: every request this client makes already replays
+    /// the handshake once on an expired session, and so does every
+    /// [`RevisionReader`](crate::RevisionReader) it hands out. Reach for it to
+    /// renew *ahead* of a long read rather than to repair a failed one.
+    ///
+    /// The share key, root uid and role belong to the link and cannot change
+    /// under a successful re-auth, so only the session is replaced.
+    pub async fn refresh_session(&self) -> Result<()> {
+        let (_, generation) = self.session.http();
+        self.session.renew(generation).await
+    }
+
+    /// As [`refresh_session`](Self::refresh_session), returning a client for
+    /// callers written against the older shape.
+    ///
+    /// The returned client shares this one's session rather than owning a
+    /// separate one — which is the point of the change: readers opened from
+    /// *either* handle recover from the expiry, where previously the original
+    /// client and everything it had handed out were stranded.
     pub async fn reauth(&self) -> Result<Self> {
-        let auth = self.session.auth().await?;
-        Ok(Self {
-            session: PublicLinkSession {
-                config: self.session.config.clone(),
-                token: self.session.token.clone(),
-                password: self.session.password.clone(),
-            },
-            http: auth.http,
-            share_key: auth.share_key,
-            root_uid: auth.root_uid,
-            public_role: auth.public_role,
-        })
+        self.refresh_session().await?;
+        Ok(self.clone())
     }
 
     /// The key that decrypts `link`: the share key at the shared root, otherwise
-    /// the parent node's key, resolved recursively up to the root.
+    /// the parent node's key.
     ///
-    /// The recursion terminates at the shared root because the visitor cannot
-    /// see past it — a link whose parent chain does not reach the root is not
+    /// The walk terminates at the shared root because the visitor cannot see
+    /// past it — a link whose parent chain does not reach the root is not
     /// readable with this session's material.
     async fn resolve_parent_key(
         &self,
@@ -470,24 +987,88 @@ impl ProtonDrivePublicLinkClient {
             ProtonError::invalid_operation("node has no parent and is not the shared root")
         })?;
 
-        let parent = self
-            .link_details(&NodeUid::new(volume_id.clone(), parent_id))
-            .await?;
-        // Recursion is boxed: the chain length is the tree depth, which the
-        // compiler cannot bound for an `async fn` that calls itself.
-        let grandparent_key = Box::pin(self.resolve_parent_key(volume_id, &parent)).await?;
-        Ok(decrypt_link(&grandparent_key, &parent.link)?.node_key)
+        self.node_key(&NodeUid::new(volume_id.clone(), parent_id))
+            .await
+    }
+
+    /// A node's own unlocked key, from cache when it is there.
+    ///
+    /// Walks *up* to the nearest cached ancestor (or the shared root, whose
+    /// parent key is the share key), then unlocks back *down*, caching every key
+    /// on the way. Written iteratively rather than recursively so it is one
+    /// finite future — an `async fn` that awaited itself would need boxing at
+    /// the recursion point, and its `Send`-ness would depend on its own, which
+    /// the compiler cannot resolve.
+    async fn node_key(&self, uid: &NodeUid) -> Result<PrivateKey> {
+        if let Some(key) = self.cached_node_key(uid) {
+            return Ok(key);
+        }
+
+        let client = self.clone();
+        let target = uid.clone();
+        self.key_loads
+            .run(uid.clone(), async move {
+                // Up: collect the chain from `target` towards the root, stopping
+                // at the first key already unlocked.
+                let mut chain: Vec<LinkDetailsDto> = Vec::new();
+                let mut cursor = target.clone();
+
+                let mut key = loop {
+                    if let Some(key) = client.cached_node_key(&cursor) {
+                        break key;
+                    }
+
+                    let details = client.link_details(&cursor).await?;
+
+                    if details.link.id == client.root_uid.link_id {
+                        let key = decrypt_link(&client.share_key, &details.link)?.node_key;
+                        client.cache_node_key(&cursor, &key);
+                        break key;
+                    }
+
+                    let parent_id = details.link.parent_id.clone().ok_or_else(|| {
+                        ProtonError::invalid_operation(
+                            "node has no parent and is not the shared root",
+                        )
+                    })?;
+                    let parent = NodeUid::new(cursor.volume_id.clone(), parent_id);
+                    chain.push(details);
+                    cursor = parent;
+                };
+
+                // Down: each node's key unlocks the next, and every one is worth
+                // keeping — the siblings still to be listed will ask for them.
+                while let Some(details) = chain.pop() {
+                    key = decrypt_link(&key, &details.link)?.node_key;
+                    client.cache_node_key(
+                        &NodeUid::new(target.volume_id.clone(), details.link.id.clone()),
+                        &key,
+                    );
+                }
+
+                Ok(key)
+            })
+            .await
+    }
+
+    fn cached_node_key(&self, uid: &NodeUid) -> Option<PrivateKey> {
+        self.node_keys
+            .lock()
+            .unwrap_or_else(std::sync::PoisonError::into_inner)
+            .get(uid)
+            .cloned()
+    }
+
+    fn cache_node_key(&self, uid: &NodeUid, key: &PrivateKey) {
+        self.node_keys
+            .lock()
+            .unwrap_or_else(std::sync::PoisonError::into_inner)
+            .put(uid.clone(), key.clone());
     }
 
     async fn link_details(&self, uid: &NodeUid) -> Result<LinkDetailsDto> {
-        let response: LinkDetailsResponse = self
-            .http
-            .post(
-                &format!("v2/volumes/{}/links", uid.volume_id),
-                &LinkDetailsRequest {
-                    link_ids: std::slice::from_ref(&uid.link_id),
-                },
-            )
+        let response = self
+            .fetch_link_details(&uid.volume_id, std::slice::from_ref(&uid.link_id))
             .await?;
         response
             .links
@@ -522,20 +1103,42 @@ impl ProtonDrivePublicLinkClient {
             NodeKind::Folder
         } else {
             let file = file_properties(details);
+            let active = file.and_then(|f| f.active_revision.as_ref());
+
+            // The link-details response already carries the active revision's
+            // `XAttr`, so the uploader's claimed size, mtime and digest cost
+            // nothing beyond one decrypt — no extra round-trip. Unverified,
+            // because a visitor has no address book to resolve the signer
+            // through; see the module note.
+            let common = active
+                .and_then(|revision| revision.extended_attributes.as_deref())
+                .and_then(|xattr| {
+                    match decrypt_extended_attributes_unverified(&decrypted.node_key, xattr) {
+                        Ok(attrs) => attrs.common,
+                        // Present but unreadable is worth a line — absent is
+                        // ordinary, and says nothing.
+                        Err(e) => {
+                            tracing::debug!(link_id = %link.id, error = %e,
+                                "revision extended attributes did not decrypt");
+                            None
+                        }
+                    }
+                });
+
             NodeKind::File {
                 media_type: file.map(|f| f.media_type.clone()).unwrap_or_default(),
                 total_size_on_storage: file.map(|f| f.total_size_on_storage).unwrap_or_default(),
-                active_revision_state: file
-                    .and_then(|f| f.active_revision.as_ref())
+                active_revision_state: active
                     .map(|revision| RevisionState::from_raw(revision.state)),
-                active_revision_id: file
-                    .and_then(|f| f.active_revision.as_ref())
-                    .map(|revision| revision.id.clone()),
-                // Claimed metadata lives in the revision's extended attributes,
-                // which this read surface does not fetch.
-                claimed_size: None,
-                claimed_modification_time: None,
-                content_sha1: None,
+                active_revision_id: active.map(|revision| revision.id.clone()),
+                claimed_size: common.as_ref().and_then(|c| c.size),
+                claimed_modification_time: common
+                    .as_ref()
+                    .and_then(|c| c.modification_time.clone()),
+                content_sha1: common
+                    .as_ref()
+                    .and_then(|c| c.digests.as_ref())
+                    .and_then(|d| d.sha1.clone()),
             }
         };
 

crates/proton-drive-rs/src/revision.rs

@@ -14,8 +14,8 @@ use crate::node::RevisionState;
 use sha2::{Digest, Sha256};
 use tokio::sync::{Mutex, OwnedSemaphorePermit, RwLock};
 
-use crate::client::ProtonDriveClient;
 use crate::dtos::BlockDto;
+use crate::transport::RevisionTransport;
 
 /// Content blocks fetched concurrently within a single file.
 ///
@@ -134,7 +134,11 @@ struct BlockTable {
 /// file gains a new revision, this reader keeps serving the old one; reopen to
 /// follow the change.
 pub struct RevisionReader {
-    client: ProtonDriveClient,
+    /// Everything the read needs from whoever opened it: block permits, an HTTP
+    /// client already carrying the right route prefix, and — for a session that
+    /// can die under the reader — the renewal path. Deliberately *not* a whole
+    /// client, so an anonymous public-link visitor can have one too.
+    transport: RevisionTransport,
     uid: NodeUid,
     revision_id: String,
     content_key: ContentKey,
@@ -149,7 +153,7 @@ pub struct RevisionReader {
 
 impl RevisionReader {
     pub(crate) fn new(
-        client: ProtonDriveClient,
+        transport: RevisionTransport,
         uid: NodeUid,
         revision_id: String,
         content_key: ContentKey,
@@ -158,7 +162,7 @@ impl RevisionReader {
     ) -> Self {
         let file_size = block_sizes.iter().sum();
         Self {
-            client,
+            transport,
             uid,
             revision_id,
             content_key,
@@ -249,7 +253,7 @@ impl RevisionReader {
         // Held across fetch *and* decrypt: both halves are resident 4 MiB
         // buffers, and the plaintext outlives the ciphertext.
         let permit = self
-            .client
+            .transport
             .block_slots()
             .acquire_owned()
             .await
@@ -257,12 +261,14 @@ impl RevisionReader {
 
         let (url, token, generation) = self.block_location(index).await?;
 
-        let ciphertext = match self.client.http().get_storage_blob(&url, &token).await {
+        // A block fetch carries a `pm-storage-token` and no session credential
+        // at all, so it cannot fail for session reasons — only the URL expires.
+        let ciphertext = match self.transport.http().get_storage_blob(&url, &token).await {
             Ok(bytes) => bytes,
             Err(e) if is_expired_block_url(&e) => {
                 self.refresh_blocks(generation).await?;
                 let (url, token, _) = self.block_location(index).await?;
-                self.client.http().get_storage_blob(&url, &token).await?
+                self.transport.http().get_storage_blob(&url, &token).await?
             }
             Err(e) => return Err(e),
         };
@@ -300,8 +306,8 @@ impl RevisionReader {
         }
 
         let (_, blocks) = self
-            .client
-            .fetch_revision_blocks(&self.uid.volume_id, &self.uid.link_id, &self.revision_id)
+            .transport
+            .list_blocks(&self.uid.volume_id, &self.uid.link_id, &self.revision_id)
             .await?;
 
         if blocks.len() != self.block_sizes.len() {

crates/proton-drive-rs/src/transport.rs

@@ -0,0 +1,492 @@
+//! What a revision read needs from its client, and nothing else.
+//!
+//! [`RevisionReader`](crate::RevisionReader) originally held a whole
+//! [`ProtonDriveClient`](crate::ProtonDriveClient), which is why an anonymous
+//! public-link visitor could not have one. It only ever used three things: the
+//! host-wide block permits, an [`ApiHttpClient`] to fetch block bodies through,
+//! and the paginated revision listing. All three are satisfied by an HTTP client
+//! plus a semaphore — and `ApiHttpClient` already carries the route prefix
+//! (`drive/` vs `drive/unauth/`), which is the only thing that differs between
+//! the authenticated and visitor paths. The endpoints, payloads, paging and
+//! contiguity rules are identical.
+//!
+//! So the seam here is deliberately narrow: [`RevisionTransport`] is *concrete*,
+//! and the one genuine axis of variation — what to do when the session dies —
+//! is the two-method [`BlockSession`] trait.
+//!
+//! ## The session asymmetry
+//!
+//! Worth stating once, because it is easy to re-derive wrongly: block bodies are
+//! fetched with [`ApiHttpClient::get_storage_blob`], which sends a
+//! `pm-storage-token` and **no bearer and no `x-pm-uid`**. Block fetches are
+//! therefore immune to session expiry — a 401 there means an expired *block
+//! URL*. The revision listing is the only session-authenticated call in a read,
+//! so it is the only one that can meet a dead session.
+
+use std::sync::Arc;
+
+use async_trait::async_trait;
+use proton_sdk::error::{ProtonError, Result};
+use proton_sdk::http::ApiHttpClient;
+use proton_sdk::ids::{LinkId, VolumeId};
+use serde::Serialize;
+use serde::de::DeserializeOwned;
+use tokio::sync::Semaphore;
+
+use crate::dtos::{BlockDto, DecryptedCommonExtendedAttributes, RevisionDto, RevisionResponse};
+
+/// Content blocks are 4 MiB of plaintext each (C# `RevisionWriter.DefaultBlockSize`).
+pub(crate) const DEFAULT_BLOCK_SIZE: usize = 1 << 22;
+
+/// The live session a revision read issues its authenticated calls through, and
+/// how to replace it when it dies.
+///
+/// Generation-versioned for the same reason
+/// [`RevisionReader::refresh_blocks`](crate::RevisionReader) versions its block
+/// table: a burst of failures must replay **one** recovery, not one per caller.
+/// A loser re-checks the generation and returns.
+#[async_trait]
+pub(crate) trait BlockSession: Send + Sync + 'static {
+    /// The client to use right now, paired with the generation it belongs to.
+    ///
+    /// Deliberately synchronous — this is on the per-request path, and it is a
+    /// lock read plus two refcount bumps. The lock is never held across an await.
+    fn http(&self) -> (ApiHttpClient, u64);
+
+    /// Replace the session, unless somebody has already replaced `seen`.
+    async fn renew(&self, seen: u64) -> Result<()>;
+}
+
+/// A session that renews itself.
+///
+/// An ordinary bearer session refreshes its own tokens inside [`ApiHttpClient`]
+/// on a 401, so by the time an error surfaces here there is nothing left to try:
+/// `renew` is a no-op and the generation never moves.
+pub(crate) struct StaticSession(ApiHttpClient);
+
+impl StaticSession {
+    pub(crate) fn new(http: ApiHttpClient) -> Self {
+        Self(http)
+    }
+}
+
+#[async_trait]
+impl BlockSession for StaticSession {
+    fn http(&self) -> (ApiHttpClient, u64) {
+        (self.0.clone(), 0)
+    }
+
+    async fn renew(&self, _seen: u64) -> Result<()> {
+        Ok(())
+    }
+}
+
+/// Everything a revision read needs that is not the revision itself.
+#[derive(Clone)]
+pub(crate) struct RevisionTransport {
+    session: Arc<dyn BlockSession>,
+    /// Host-wide in-flight block permits, shared with the owning client's other
+    /// transfers.
+    block_slots: Arc<Semaphore>,
+}
+
+impl RevisionTransport {
+    pub(crate) fn new(session: Arc<dyn BlockSession>, block_slots: Arc<Semaphore>) -> Self {
+        Self {
+            session,
+            block_slots,
+        }
+    }
+
+    /// A transport over a session that manages its own tokens.
+    pub(crate) fn authenticated(http: ApiHttpClient, block_slots: Arc<Semaphore>) -> Self {
+        Self::new(Arc::new(StaticSession::new(http)), block_slots)
+    }
+
+    /// The current HTTP client. Block fetches go straight through this — they
+    /// carry no session credential, so they need no renewal path.
+    pub(crate) fn http(&self) -> ApiHttpClient {
+        self.session.http().0
+    }
+
+    pub(crate) fn block_slots(&self) -> Arc<Semaphore> {
+        self.block_slots.clone()
+    }
+
+    /// `GET path` through the current session, replaying the handshake once if
+    /// the session turns out to be gone.
+    pub(crate) async fn get<T: DeserializeOwned>(&self, path: &str) -> Result<T> {
+        let (http, generation) = self.session.http();
+        match http.get::<T>(path).await {
+            Err(e) if is_session_expired(&e) => {
+                self.session.renew(generation).await?;
+                self.session.http().0.get::<T>(path).await
+            }
+            other => other,
+        }
+    }
+
+    /// `POST path` through the current session, with the same one-shot renewal.
+    pub(crate) async fn post<B: Serialize + Sync, T: DeserializeOwned>(
+        &self,
+        path: &str,
+        body: &B,
+    ) -> Result<T> {
+        let (http, generation) = self.session.http();
+        match http.post::<B, T>(path, body).await {
+            Err(e) if is_session_expired(&e) => {
+                self.session.renew(generation).await?;
+                self.session.http().0.post::<B, T>(path, body).await
+            }
+            other => other,
+        }
+    }
+
+    /// The revision's metadata and its full, ordered block table.
+    pub(crate) async fn list_blocks(
+        &self,
+        volume_id: &VolumeId,
+        link_id: &LinkId,
+        revision_id: &str,
+    ) -> Result<(RevisionDto, Vec<BlockDto>)> {
+        const PAGE_SIZE: i32 = 50;
+
+        let mut blocks: Vec<BlockDto> = Vec::new();
+        let mut metadata: Option<RevisionDto> = None;
+        let mut from_index: i32 = 1;
+
+        loop {
+            let path = format!(
+                "v2/volumes/{volume_id}/files/{link_id}/revisions/{revision_id}?FromBlockIndex={from_index}&PageSize={PAGE_SIZE}&NoBlockUrls=0"
+            );
+            let response: RevisionResponse = self.get(&path).await?;
+            let mut revision = response.revision;
+            let page = std::mem::take(&mut revision.blocks);
+            let page_len = page.len();
+
+            if metadata.is_none() {
+                metadata = Some(revision);
+            }
+            blocks.extend(page);
+
+            if page_len < PAGE_SIZE as usize {
+                break;
+            }
+            from_index = blocks.iter().map(|b| b.index).max().unwrap_or(from_index) + 1;
+        }
+
+        blocks.sort_by_key(|b| b.index);
+        for (offset, block) in blocks.iter().enumerate() {
+            if block.index != offset as i32 + 1 {
+                return Err(ProtonError::invalid_operation(
+                    "file contents are incomplete (non-contiguous blocks)",
+                ));
+            }
+        }
+
+        let metadata = metadata
+            .ok_or_else(|| ProtonError::invalid_operation("revision returned no metadata"))?;
+        Ok((metadata, blocks))
+    }
+}
+
+/// Whether a failure means the *session* is gone.
+///
+/// Only meaningful for session-authenticated calls. A block fetch carries a
+/// `pm-storage-token` rather than the bearer, so a 401 there is an expired block
+/// URL — see `revision::is_expired_block_url`, which deliberately also accepts
+/// 403 and 404.
+///
+/// 403 is excluded here on purpose: on the visitor path it means
+/// `MissingScopes`, i.e. the request went to `drive/` instead of
+/// `drive/unauth/`. Re-authenticating would not fix that and would loop.
+pub(crate) fn is_session_expired(error: &ProtonError) -> bool {
+    matches!(error, ProtonError::Api(e) if e.http_status == 401)
+}
+
+/// Rank a revision's block-size sources and produce the per-block plaintext
+/// sizes.
+///
+/// These sizes are not merely descriptive: [`RevisionReader`](crate::RevisionReader)
+/// accumulates them to derive where each block *starts* in the plaintext. A
+/// vector that overstates any block but the last therefore shifts every block
+/// after it, and the reader splices those blocks at the wrong offsets —
+/// returning full-length reads of the wrong bytes, with no error to notice. See
+/// `revision.rs::padded_block_sizes_shift_later_blocks`, which pins that
+/// consequence.
+///
+/// So the sources are ranked by how much they can be trusted, and this refuses
+/// rather than guesses when none of them holds:
+///
+/// 1. `Common.BlockSizes` from the revision's extended attributes — the
+///    authoritative value, written by the uploading client.
+/// 2. `Common.Size` — the total, from which all-but-last full blocks and a final
+///    short block follow by subtraction. Sound for the standard layout the
+///    uploader produces.
+/// 3. Nothing usable. A single-block file cannot be misaligned (there is no
+///    later block to shift), so it is assumed full-size and only its reported
+///    length can be wrong. A multi-block file is an error: serving it would mean
+///    serving misplaced bytes.
+///
+/// **Not a rung:** `FileDto.total_size_on_storage` (wire `TotalEncryptedSize`).
+/// That is the *ciphertext* size — PGP framing per block, varying with the
+/// encryption mode — so deriving plaintext block boundaries from it produces
+/// exactly the shifted-splice failure above. Do not add it.
+pub(crate) fn rank_block_sizes(
+    common: Option<&DecryptedCommonExtendedAttributes>,
+    revision_id: &str,
+    block_count: usize,
+) -> Result<Vec<u64>> {
+    let block = DEFAULT_BLOCK_SIZE as u64;
+
+    if let Some(sizes) = common.and_then(|c| c.block_sizes.as_ref())
+        && sizes.len() == block_count
+    {
+        // A non-positive entry is as corrupting as a padded one — a zero-length
+        // block in the middle shifts everything after it — so a malformed vector
+        // is rejected outright rather than clamped.
+        if let Some(bad) = sizes.iter().position(|&n| n <= 0) {
+            return Err(ProtonError::invalid_operation(format!(
+                "revision {revision_id} records a non-positive size for block {bad}"
+            )));
+        }
+        return Ok(sizes.iter().map(|&n| n as u64).collect());
+    }
+
+    if let Some(total) = common.and_then(|c| c.size).filter(|&n| n >= 0) {
+        let total = total as u64;
+        return Ok((0..block_count)
+            .map(|i| total.saturating_sub(block * i as u64).min(block))
+            .collect());
+    }
+
+    if block_count <= 1 {
+        return Ok(vec![block; block_count]);
+    }
+
+    Err(ProtonError::invalid_operation(format!(
+        "revision {revision_id} has {block_count} blocks but no usable size information \
+         (no block sizes and no total in its extended attributes)"
+    )))
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use std::sync::atomic::{AtomicUsize, Ordering};
+
+    fn common(
+        size: Option<i64>,
+        block_sizes: Option<Vec<i64>>,
+    ) -> DecryptedCommonExtendedAttributes {
+        DecryptedCommonExtendedAttributes {
+            size,
+            block_sizes,
+            ..Default::default()
+        }
+    }
+
+    const MIB: u64 = 1 << 20;
+
+    /// The authoritative rung: whatever the uploader recorded is used as-is,
+    /// including a short final block.
+    #[test]
+    fn block_sizes_from_extended_attributes_are_used_verbatim() {
+        let attrs = common(Some(9 * MIB as i64), Some(vec![4 << 20, 4 << 20, 1 << 20]));
+        let sizes = rank_block_sizes(Some(&attrs), "rev", 3).expect("sizes resolve");
+        assert_eq!(sizes, vec![4 * MIB, 4 * MIB, MIB]);
+    }
+
+    /// A vector that does not describe every block is not partial information,
+    /// it is untrustworthy information — fall through to the total rather than
+    /// pad it out.
+    #[test]
+    fn a_block_sizes_vector_of_the_wrong_length_falls_through_to_the_total() {
+        let attrs = common(Some(9 * MIB as i64), Some(vec![4 << 20, 4 << 20]));
+        let sizes = rank_block_sizes(Some(&attrs), "rev", 3).expect("sizes resolve");
+        assert_eq!(sizes, vec![4 * MIB, 4 * MIB, MIB], "derived from the total");
+    }
+
+    /// A zero-length block in the middle shifts every block after it just as a
+    /// padded one does, so it is refused rather than clamped to something
+    /// plausible.
+    #[test]
+    fn a_non_positive_block_size_is_rejected_rather_than_clamped() {
+        let attrs = common(Some(9 * MIB as i64), Some(vec![4 << 20, 0, 1 << 20]));
+        let error = rank_block_sizes(Some(&attrs), "rev", 3).expect_err("must refuse");
+        assert!(
+            error.to_string().contains("non-positive size for block 1"),
+            "names the offending block: {error}"
+        );
+    }
+
+    /// The total alone is enough, because the uploader's layout is all-but-last
+    /// full blocks plus a remainder.
+    #[test]
+    fn the_total_size_derives_full_blocks_and_a_short_final_one() {
+        let attrs = common(Some(9 * MIB as i64), None);
+        let sizes = rank_block_sizes(Some(&attrs), "rev", 3).expect("sizes resolve");
+        assert_eq!(sizes, vec![4 * MIB, 4 * MIB, MIB]);
+        assert_eq!(sizes.iter().sum::<u64>(), 9 * MIB, "sums to the total");
+    }
+
+    /// A file that happens to be an exact multiple of the block size gets no
+    /// short tail, and no phantom extra block.
+    #[test]
+    fn a_total_that_is_a_whole_number_of_blocks_has_no_short_tail() {
+        let attrs = common(Some(8 * MIB as i64), None);
+        let sizes = rank_block_sizes(Some(&attrs), "rev", 2).expect("sizes resolve");
+        assert_eq!(sizes, vec![4 * MIB, 4 * MIB]);
+    }
+
+    /// With one block there is no later block to displace, so assuming the full
+    /// block size can only misreport the length — never misplace bytes.
+    #[test]
+    fn a_single_block_revision_is_assumed_full_size() {
+        let sizes = rank_block_sizes(None, "rev", 1).expect("sizes resolve");
+        assert_eq!(sizes, vec![4 * MIB]);
+        assert_eq!(
+            rank_block_sizes(None, "rev", 0).expect("empty resolves"),
+            Vec::<u64>::new()
+        );
+    }
+
+    /// The refusal that keeps the reader honest: guessing here would serve
+    /// full-length reads of the wrong bytes.
+    #[test]
+    fn a_multi_block_revision_with_no_size_information_is_an_error() {
+        let error = rank_block_sizes(None, "rev-42", 3).expect_err("must refuse");
+        let message = error.to_string();
+        assert!(message.contains("rev-42"), "names the revision: {message}");
+        assert!(
+            message.contains("no usable size information"),
+            "says why: {message}"
+        );
+    }
+
+    /// A dead session is a 401 and nothing else. 403 on the visitor path means
+    /// `MissingScopes` — a wrong route prefix, which re-authenticating would
+    /// only loop on — and 404 means an expired block URL.
+    #[test]
+    fn only_a_401_reads_as_an_expired_session() {
+        use proton_sdk::api::ResponseCode;
+        use proton_sdk::error::ProtonApiError;
+
+        let api = |status: u16| {
+            ProtonError::Api(ProtonApiError {
+                code: ResponseCode::Unknown,
+                http_status: status,
+                message: "nope".into(),
+                details: None,
+            })
+        };
+
+        assert!(is_session_expired(&api(401)));
+        assert!(
+            !is_session_expired(&api(403)),
+            "MissingScopes is not expiry"
+        );
+        assert!(
+            !is_session_expired(&api(404)),
+            "an expired block url is not"
+        );
+        assert!(!is_session_expired(&ProtonError::invalid_operation("x")));
+    }
+
+    /// A session that counts its handshakes, so renewal can be tested without a
+    /// network.
+    struct CountingSession {
+        http: ApiHttpClient,
+        generation: std::sync::Mutex<u64>,
+        handshakes: AtomicUsize,
+    }
+
+    #[async_trait]
+    impl BlockSession for CountingSession {
+        fn http(&self) -> (ApiHttpClient, u64) {
+            (self.http.clone(), *self.generation.lock().unwrap())
+        }
+
+        async fn renew(&self, seen: u64) -> Result<()> {
+            let mut generation = self.generation.lock().unwrap();
+            if *generation != seen {
+                // Somebody else already replaced the session we were told about.
+                return Ok(());
+            }
+            self.handshakes.fetch_add(1, Ordering::SeqCst);
+            *generation += 1;
+            Ok(())
+        }
+    }
+
+    fn counting_session() -> Arc<CountingSession> {
+        use proton_sdk::config::ProtonClientConfiguration;
+        use proton_sdk::http::Tokens;
+        use proton_sdk::ids::SessionId;
+
+        let http = ApiHttpClient::new(
+            ProtonClientConfiguration::new("test@1.0"),
+            SessionId::from("anonymous"),
+            Tokens {
+                access_token: "access".into(),
+                refresh_token: String::new(),
+            },
+        )
+        .expect("build client");
+
+        Arc::new(CountingSession {
+            http,
+            generation: std::sync::Mutex::new(0),
+            handshakes: AtomicUsize::new(0),
+        })
+    }
+
+    /// Concurrent readers that all saw the same dead session replay **one**
+    /// handshake between them — the generation check makes the losers no-ops.
+    /// Without this, a seek that spans ten blocks would run ten SRP handshakes.
+    #[tokio::test]
+    async fn concurrent_renewals_of_one_generation_run_a_single_handshake() {
+        let session = counting_session();
+
+        let mut set = tokio::task::JoinSet::new();
+        for _ in 0..8 {
+            let session = session.clone();
+            set.spawn(async move { session.renew(0).await });
+        }
+        while let Some(result) = set.join_next().await {
+            result.expect("task").expect("renew");
+        }
+
+        assert_eq!(
+            session.handshakes.load(Ordering::SeqCst),
+            1,
+            "eight callers, one handshake"
+        );
+        assert_eq!(*session.generation.lock().unwrap(), 1);
+    }
+
+    /// A caller arriving with a stale generation has already been rescued by
+    /// somebody else's handshake, so it must not start another.
+    #[tokio::test]
+    async fn renewing_a_generation_that_was_already_replaced_is_a_no_op() {
+        let session = counting_session();
+
+        session.renew(0).await.expect("first renew");
+        assert_eq!(session.handshakes.load(Ordering::SeqCst), 1);
+
+        session.renew(0).await.expect("stale renew");
+        assert_eq!(
+            session.handshakes.load(Ordering::SeqCst),
+            1,
+            "the stale caller did not re-handshake"
+        );
+    }
+
+    /// The transport is what `RevisionReader` stores, so it has to survive being
+    /// moved into a spawned task alongside the reader.
+    #[test]
+    fn a_revision_transport_is_send_and_sync() {
+        fn assert_send_sync<T: Send + Sync>() {}
+        assert_send_sync::<RevisionTransport>();
+    }
+}

crates/proton-drive-rs/tests/live_sharing.rs

@@ -19,7 +19,8 @@
 mod common;
 
 use proton_drive_rs::{
-    MemberRole, ProtonDriveClient, ProtonDrivePublicLinkClient, ProtonPhotosClient,
+    MemberRole, ProtonDriveClient, ProtonDrivePublicLinkClient, ProtonPhotosClient, Thumbnail,
+    ThumbnailType,
 };
 use proton_sdk::ids::NodeUid;
 
@@ -439,3 +440,318 @@ async fn custom_password_link_requires_the_password() {
 
     cleanup(client, &folder).await;
 }
+
+// ---------------------------------------------------------------------------
+// Public link — seekable reads, the streaming path
+// ---------------------------------------------------------------------------
+
+/// A multi-block file, in blocks of `4 MiB`, whose every byte encodes its own
+/// offset. A range served from the wrong block is then identifiable rather than
+/// merely unequal — which matters here, because a wrong block-size vector
+/// returns a full-length read of the wrong bytes with no error at all.
+fn positional_payload(len: usize) -> Vec<u8> {
+    (0..len).map(|i| (i % 251) as u8).collect()
+}
+
+/// **The load-bearing test for streaming over a public link.**
+///
+/// Everything `proton-stream` does rests on one uncertain fact: that the
+/// `drive/unauth/` revision endpoint returns a usable `XAttr` to an anonymous
+/// visitor. Without it `rank_block_sizes` cannot place block boundaries and
+/// refuses to open a multi-block file, and seeking is impossible. The visitor
+/// route has form here — `?NoBlockUrls=true` already strips `BareURL`/`Token` —
+/// so this is verified, not assumed.
+///
+/// Then the reads themselves: a range spanning a block boundary, a range wholly
+/// inside a later block (the seek case), and the short final block.
+#[tokio::test]
+#[ignore = "live: needs test-account credentials"]
+async fn a_visitor_can_seek_within_a_shared_multi_block_file() {
+    let Some(live) = common::live_client().await else {
+        return;
+    };
+    let client = &live.client;
+
+    let folder = scratch_folder(client, "public-seek").await;
+
+    // 9 MiB: two full 4 MiB blocks and a 1 MiB tail, so both the boundary case
+    // and the short-tail case are reachable.
+    const BLOCK: usize = 4 << 20;
+    let contents = positional_payload(9 << 20);
+    let name = format!("seekable-{}.bin", common::unique_suffix());
+    let file = client
+        .upload_file(&folder, &name, "application/octet-stream", &contents)
+        .await
+        .expect("upload_file");
+
+    let link = client
+        .create_public_link(&folder, MemberRole::Viewer, None, None)
+        .await
+        .expect("create_public_link");
+    let url = link.url.clone().expect("a created link must carry its URL");
+
+    let visitor = ProtonDrivePublicLinkClient::open(common::config(), &url, None)
+        .await
+        .expect("open public link");
+
+    let reader = visitor
+        .open_revision(&file)
+        .await
+        .expect("a visitor must be able to open a revision for reading");
+
+    // If this fails, the unauth revision response carried no usable size and the
+    // whole streaming design needs revisiting — so it is asserted before the
+    // reads, with a message that says which rung was missing.
+    assert_eq!(
+        reader.block_sizes(),
+        &[BLOCK as u64, BLOCK as u64, 1 << 20],
+        "the visitor must resolve true per-block sizes; a padded or empty vector \
+         means the unauth revision response omitted its XAttr"
+    );
+    assert_eq!(
+        reader.size(),
+        contents.len() as u64,
+        "and they must sum to the real file size"
+    );
+
+    // A range straddling the block 0 / block 1 boundary.
+    let straddle = reader
+        .read_at(BLOCK as u64 - 1024, 2048)
+        .await
+        .expect("read across a block boundary");
+    assert_eq!(
+        straddle,
+        &contents[BLOCK - 1024..BLOCK + 1024],
+        "a read spanning two blocks must splice them at the right offset"
+    );
+
+    // The seek case: a range wholly inside the *last* full block, touching
+    // neither of the ones before it.
+    let seeked = reader
+        .read_at(6 << 20, 64 << 10)
+        .await
+        .expect("read from the middle of the file");
+    let want = &contents[(6 << 20)..(6 << 20) + (64 << 10)];
+    assert_eq!(
+        seeked, want,
+        "a mid-file read must return the bytes at that offset, not an earlier block's"
+    );
+
+    // The short tail, and a read that runs past EOF: clamped, never padded.
+    let tail = reader
+        .read_at(8 << 20, 4 << 20)
+        .await
+        .expect("read the short final block");
+    assert_eq!(
+        tail,
+        &contents[8 << 20..],
+        "the final block is 1 MiB, and a read past EOF is clamped to it"
+    );
+    assert!(
+        reader
+            .read_at(contents.len() as u64, 4096)
+            .await
+            .expect("a read at EOF is not an error")
+            .is_empty(),
+        "a read starting at EOF yields nothing"
+    );
+
+    // `download_range` is the one-shot form of the same thing.
+    let ranged = visitor
+        .download_range(&file, 5 << 20, 4096)
+        .await
+        .expect("download_range");
+    assert_eq!(ranged, &contents[(5 << 20)..(5 << 20) + 4096]);
+
+    // And the whole-file path must still agree with all of it — it now drives a
+    // paged, concurrent block pipeline rather than one un-paged sequential loop.
+    let whole = visitor
+        .download_file(&file)
+        .await
+        .expect("visitor download");
+    assert_eq!(
+        whole, contents,
+        "the whole-file download must match the uploaded bytes"
+    );
+
+    cleanup(client, &folder).await;
+}
+
+/// A visitor's node listing must carry the uploader's claimed metadata, which it
+/// previously left as `None` — a catalog has nothing to show without a size.
+#[tokio::test]
+#[ignore = "live: needs test-account credentials"]
+async fn a_visitor_sees_claimed_size_and_digest_on_shared_files() {
+    let Some(live) = common::live_client().await else {
+        return;
+    };
+    let client = &live.client;
+
+    let folder = scratch_folder(client, "public-xattr").await;
+
+    let contents = positional_payload(64 << 10);
+    let name = format!("claimed-{}.bin", common::unique_suffix());
+    let file = client
+        .upload_file(&folder, &name, "application/octet-stream", &contents)
+        .await
+        .expect("upload_file");
+
+    let link = client
+        .create_public_link(&folder, MemberRole::Viewer, None, None)
+        .await
+        .expect("create_public_link");
+    let url = link.url.clone().expect("a created link must carry its URL");
+
+    let visitor = ProtonDrivePublicLinkClient::open(common::config(), &url, None)
+        .await
+        .expect("open public link");
+
+    let node = visitor
+        .get_node(&file)
+        .await
+        .expect("visitor node lookup")
+        .expect("the shared file must exist");
+
+    match &node.kind {
+        proton_drive_rs::NodeKind::File {
+            claimed_size,
+            content_sha1,
+            ..
+        } => {
+            assert_eq!(
+                *claimed_size,
+                Some(contents.len() as i64),
+                "the visitor must read the uploader's claimed size from the link details XAttr"
+            );
+            assert!(
+                content_sha1.is_some(),
+                "and the content digest alongside it"
+            );
+        }
+        other => panic!("expected a file, got {other:?}"),
+    }
+
+    cleanup(client, &folder).await;
+}
+
+/// Renewing the session must not strand a reader that was already handed out —
+/// the whole reason the session is swapped in place rather than replaced.
+///
+/// Simulates the expiry rather than waiting hours for one: an explicit
+/// `refresh_session` puts the reader on a session it was not opened with, which
+/// is exactly the state a real expiry leaves it in once recovery has run.
+#[tokio::test]
+#[ignore = "live: needs test-account credentials"]
+async fn refreshing_the_session_keeps_an_open_reader_working() {
+    let Some(live) = common::live_client().await else {
+        return;
+    };
+    let client = &live.client;
+
+    let folder = scratch_folder(client, "public-renew").await;
+
+    let contents = positional_payload(5 << 20);
+    let name = format!("renew-{}.bin", common::unique_suffix());
+    let file = client
+        .upload_file(&folder, &name, "application/octet-stream", &contents)
+        .await
+        .expect("upload_file");
+
+    let link = client
+        .create_public_link(&folder, MemberRole::Viewer, None, None)
+        .await
+        .expect("create_public_link");
+    let url = link.url.clone().expect("a created link must carry its URL");
+
+    let visitor = ProtonDrivePublicLinkClient::open(common::config(), &url, None)
+        .await
+        .expect("open public link");
+
+    let reader = visitor.open_revision(&file).await.expect("open_revision");
+
+    let before = reader.read_at(0, 4096).await.expect("read before refresh");
+    assert_eq!(before, &contents[..4096]);
+
+    visitor
+        .refresh_session()
+        .await
+        .expect("refresh_session must replay the handshake");
+
+    // Past the first block, so this genuinely fetches under the new session
+    // rather than replaying anything cached.
+    let after = reader
+        .read_at(4 << 20, 4096)
+        .await
+        .expect("a reader opened before the refresh must keep working");
+    assert_eq!(after, &contents[(4 << 20)..(4 << 20) + 4096]);
+
+    cleanup(client, &folder).await;
+}
+
+/// Thumbnails on the visitor path: the same block-storage fetch and content-key
+/// decrypt as the authenticated one, over `drive/unauth/`.
+#[tokio::test]
+#[ignore = "live: needs test-account credentials"]
+async fn a_visitor_can_download_thumbnails_of_shared_files() {
+    let Some(live) = common::live_client().await else {
+        return;
+    };
+    let client = &live.client;
+
+    let folder = scratch_folder(client, "public-thumb").await;
+
+    let contents = positional_payload(32 << 10);
+    let thumbnail = b"not a real jpeg, but it round-trips".to_vec();
+    let name = format!("thumbed-{}.bin", common::unique_suffix());
+    let file = client
+        .upload_file_from(
+            &folder,
+            &name,
+            "application/octet-stream",
+            std::io::Cursor::new(contents.clone()),
+            contents.len() as i64,
+            vec![
+                Thumbnail::new(ThumbnailType::Thumbnail, thumbnail.clone())
+                    .expect("small thumbnail"),
+            ],
+            None,
+            false,
+        )
+        .await
+        .expect("upload_file_from with a thumbnail");
+
+    let link = client
+        .create_public_link(&folder, MemberRole::Viewer, None, None)
+        .await
+        .expect("create_public_link");
+    let url = link.url.clone().expect("a created link must carry its URL");
+
+    let visitor = ProtonDrivePublicLinkClient::open(common::config(), &url, None)
+        .await
+        .expect("open public link");
+
+    let single = visitor
+        .download_thumbnail(&file, ThumbnailType::Thumbnail)
+        .await
+        .expect("visitor thumbnail download");
+    assert_eq!(
+        single.as_deref(),
+        Some(thumbnail.as_slice()),
+        "the visitor must decrypt the thumbnail it was shown"
+    );
+
+    let batch = visitor
+        .enumerate_thumbnails(std::slice::from_ref(&file), ThumbnailType::Thumbnail)
+        .await
+        .expect("visitor thumbnail batch");
+    assert_eq!(batch.len(), 1, "one file in, one result out");
+    assert_eq!(
+        batch[0]
+            .result
+            .as_deref()
+            .expect("the batch entry succeeded"),
+        thumbnail.as_slice()
+    );
+
+    cleanup(client, &folder).await;
+}

crates/proton-drive-rs/tests/spawnable.rs

@@ -16,7 +16,7 @@ use std::sync::Arc;
 
 use tokio::task::JoinHandle;
 
-use proton_drive_rs::{ProtonDriveClient, RevisionReader};
+use proton_drive_rs::{ProtonDriveClient, ProtonDrivePublicLinkClient, RevisionReader};
 use proton_sdk::ids::NodeUid;
 
 /// `download_file` inside a spawned task.
@@ -128,3 +128,49 @@ fn reader_read_at_is_spawnable(
         Ok::<_, proton_sdk::error::ProtonError>(())
     })
 }
+
+/// The same guard on the visitor path, which is the one that actually matters
+/// for a streaming player: `open_revision` over a public link, then several
+/// concurrent `read_at`s off one shared reader inside spawned tasks.
+///
+/// The public client reaches this through a different route than the
+/// authenticated one — its transport wraps a renewable session behind a trait
+/// object — so `Send`ness has to be re-proven, not assumed.
+fn public_link_reader_read_at_is_spawnable(
+    client: ProtonDrivePublicLinkClient,
+    uid: NodeUid,
+) -> JoinHandle<proton_sdk::error::Result<()>> {
+    tokio::spawn(async move {
+        let reader: Arc<RevisionReader> = Arc::new(client.open_revision(&uid).await?);
+        let mut set = tokio::task::JoinSet::new();
+        for i in 0..4u64 {
+            let reader = reader.clone();
+            set.spawn(async move { reader.read_at(i * 4096, 4096).await });
+        }
+        while let Some(joined) = set.join_next().await {
+            joined.expect("task panicked")?;
+        }
+        Ok::<_, proton_sdk::error::ProtonError>(())
+    })
+}
+
+/// `download_file_to` on the visitor path, which now drives the same `buffered`
+/// block pipeline the authenticated one does.
+fn public_link_download_file_to_is_spawnable(
+    client: ProtonDrivePublicLinkClient,
+    uid: NodeUid,
+) -> JoinHandle<proton_sdk::error::Result<()>> {
+    tokio::spawn(async move {
+        let mut sink: Vec<u8> = Vec::new();
+        client.download_file_to(&uid, &mut sink).await
+    })
+}
+
+/// `enumerate_nodes` on the visitor path: chunked and fanned out with
+/// `buffered` over cloned clients.
+fn public_link_enumerate_nodes_is_spawnable(
+    client: ProtonDrivePublicLinkClient,
+    uids: Vec<NodeUid>,
+) -> JoinHandle<proton_sdk::error::Result<Vec<proton_drive_rs::Node>>> {
+    tokio::spawn(async move { client.enumerate_nodes(&uids).await })
+}