From 9e53a9bbdb9e0e16bb589f3689af9cdc0bf550da Mon Sep 17 00:00:00 2001 From: Nils Pukropp Date: Mon, 12 Oct 2026 21:32:05 +0200 Subject: [PATCH] fix(core,app,android): refresh an expired visitor session on the desktop too The age check and the one retry behind a refresh (B22) lived only in the Android bridge. The desktop opened streams, offline copies and crawls against whatever session the share was opened with, so a window left open for a few hours failed to play or recrawl until it was restarted. The logic moves to pstr-core as SessionAges, owned by SharedLibrary and reached through SharedLibrary::reauthenticating. Both engines go through it for play, warm, offline copy and (desktop) crawl. --- crates/pstr-android/src/lib.rs | 62 ++---------- crates/pstr-app/src/engine.rs | 30 ++++-- crates/pstr-core/src/shares.rs | 180 ++++++++++++++++++++++++++++++++- 3 files changed, 209 insertions(+), 63 deletions(-) diff --git a/crates/pstr-android/src/lib.rs b/crates/pstr-android/src/lib.rs index 442d4b8..1aa4c31 100644 --- a/crates/pstr-android/src/lib.rs +++ b/crates/pstr-android/src/lib.rs @@ -949,12 +949,6 @@ pub struct AndroidEngine { share_generation: AtomicU64, /// How many Proton thumbnails may be in flight at once. thumbnails: tokio::sync::Semaphore, - /// When each share's visitor session was last refreshed. - /// - /// Sessions are short-lived and the app is routinely backgrounded for hours - /// between one episode and the next, so age — not just a share-store - /// mutation — is a reason to reauthenticate before opening anything. - session_refreshed: Mutex>, /// The streaming cache's budget in bytes, for the next connection and, /// through [`AndroidEngine::set_stream_cache_budget`], the open one. cache_budget: AtomicU64, @@ -982,13 +976,6 @@ const OFFLINE_DOWNLOAD_CONCURRENCY: usize = 2; /// How often a download waiting for a slot asks whether it was cancelled. const DOWNLOAD_WAIT_POLL: std::time::Duration = std::time::Duration::from_millis(500); -/// How long a visitor session is assumed good for without a refresh. -/// -/// A refresh costs one round trip. Not refreshing costs an authentication -/// failure that, before this, the viewer could only clear with a manual -/// pull-to-refresh. -const SESSION_FRESH_FOR: std::time::Duration = std::time::Duration::from_secs(5 * 60); - #[uniffi::export] impl AndroidEngine { #[uniffi::constructor] @@ -1034,7 +1021,6 @@ impl AndroidEngine { share_publication: Mutex::new(()), share_generation: AtomicU64::new(0), thumbnails: tokio::sync::Semaphore::new(THUMBNAIL_CONCURRENCY), - session_refreshed: Mutex::new(HashMap::new()), cache_budget: AtomicU64::new(DiskCacheConfig::DEFAULT_BUDGET_BYTES), download_permits: Arc::new(tokio::sync::Semaphore::new(OFFLINE_DOWNLOAD_CONCURRENCY)), accounts, @@ -2281,7 +2267,6 @@ impl AndroidEngine { } } drop(reusable); - self.session_refreshed.lock().remove(share_id); self.share_generation.fetch_add(1, Ordering::AcqRel); } @@ -2445,39 +2430,17 @@ impl AndroidEngine { uid: &NodeUid, copy: bool, ) -> Result { - let open = || async { - if copy { - connection.source.open_for_copy(share_id, uid).await - } else { - connection.source.open(share_id, uid).await - } - }; - let stale = self - .session_refreshed - .lock() - .get(share_id) - .is_none_or(|at| at.elapsed() >= SESSION_FRESH_FOR); - if stale { - self.refresh_session(connection, share_id).await; - } - match open().await { - Ok(stream) => Ok(stream), - Err(_) if !stale => { - self.refresh_session(connection, share_id).await; - open().await.map_err(BridgeError::from_display) - } - Err(error) => Err(BridgeError::from_display(error)), - } - } - - /// Best-effort: a refresh that fails leaves the open to report the real - /// problem, which is more useful than a refresh error standing in for it. - async fn refresh_session(&self, connection: &Connection, share_id: &str) { - if connection.library.refresh_session(share_id).await.is_ok() { - self.session_refreshed - .lock() - .insert(share_id.to_owned(), std::time::Instant::now()); - } + connection + .library + .reauthenticating(share_id, || async { + if copy { + connection.source.open_for_copy(share_id, uid).await + } else { + connection.source.open(share_id, uid).await + } + }) + .await + .map_err(BridgeError::from_display) } async fn crawl_inner(&self, share_id: Option) -> Result<(), BridgeError> { @@ -2494,9 +2457,6 @@ impl AndroidEngine { .map_err(|error| BridgeError::Failure { reason: format!("refresh session for {id}: {error}"), })?; - self.session_refreshed - .lock() - .insert(id.clone(), std::time::Instant::now()); let before = self .catalog .lock() diff --git a/crates/pstr-app/src/engine.rs b/crates/pstr-app/src/engine.rs index ba01399..9573077 100644 --- a/crates/pstr-app/src/engine.rs +++ b/crates/pstr-app/src/engine.rs @@ -1371,7 +1371,9 @@ impl Engine { // share's rows and its stale downloads, and stopping halfway through // that would leave the two disagreeing. let nodes = tokio::select! { - result = library.crawl_reporting(&share_id, report) => match result { + result = library.reauthenticating(&share_id, || { + library.crawl_reporting(&share_id, &report) + }) => match result { Ok(nodes) => nodes, Err(error) => { self.fail(&format!("crawl {share_id}"), error); @@ -1549,11 +1551,14 @@ impl Engine { return; } } - let Some((_, source)) = engine.ensure_connected().await else { + let Some((library, source)) = engine.ensure_connected().await else { return; }; let uid = node_uid(&target.volume_id, &target.link_id); - match source.open(&target.share_id, &uid).await { + let opened = library + .reauthenticating(&target.share_id, || source.open(&target.share_id, &uid)) + .await; + match opened { Ok(stream) if stream.size() == 0 => { engine.emit(Event::Error(format!("{} has no content", target.name))); } @@ -1594,10 +1599,13 @@ impl Engine { if offline { return; } - let Some(source) = engine.opened().map(|(_, source)| source) else { + let Some((library, source)) = engine.opened() else { return; }; - match source.open(&share_id, &uid).await { + let opened = library + .reauthenticating(&share_id, || source.open(&share_id, &uid)) + .await; + match opened { Ok(stream) => stream.warm_edges(), Err(error) => tracing::debug!(%error, "warming the next episode failed"), } @@ -1930,14 +1938,14 @@ impl Engine { } }; self.update_download(&key, |item| item.state = DownloadState::Running); - let Some((_, source)) = self.ensure_connected().await else { + let Some((library, source)) = self.ensure_connected().await else { anyhow::bail!("share is not connected"); }; - let stream = source - .open_for_copy( - &target.share_id, - &node_uid(&target.volume_id, &target.link_id), - ) + let uid = node_uid(&target.volume_id, &target.link_id); + let stream = library + .reauthenticating(&target.share_id, || { + source.open_for_copy(&target.share_id, &uid) + }) .await?; let revision_id = stream.revision_id().to_owned(); let block_sizes = stream.block_sizes().to_vec(); diff --git a/crates/pstr-core/src/shares.rs b/crates/pstr-core/src/shares.rs index 5189990..1515c87 100644 --- a/crates/pstr-core/src/shares.rs +++ b/crates/pstr-core/src/shares.rs @@ -607,6 +607,75 @@ async fn open_link(url: &str, custom_password: Option<&str>) -> Result, + sessions: SessionAges, +} + +/// How long a visitor session is assumed good for without a refresh. +/// +/// A refresh costs one round trip. Not refreshing costs an authentication +/// failure that the viewer can only clear by restarting or recrawling. +pub const SESSION_FRESH_FOR: std::time::Duration = std::time::Duration::from_secs(5 * 60); + +/// When each share's visitor session was last known good. +/// +/// Sessions are short-lived and an app routinely sits open, or in the +/// background, for hours between one episode and the next, so age is a reason +/// to reauthenticate before opening anything. A share with no entry is stale. +#[derive(Debug, Default)] +pub struct SessionAges { + refreshed: parking_lot::Mutex>, +} + +impl SessionAges { + /// Record that `share_id`'s session was good just now. + pub fn mark(&self, share_id: &str) { + self.refreshed + .lock() + .insert(share_id.to_owned(), std::time::Instant::now()); + } + + fn is_stale(&self, share_id: &str) -> bool { + self.refreshed + .lock() + .get(share_id) + .is_none_or(|at| at.elapsed() >= SESSION_FRESH_FOR) + } + + /// Run `operation`, reauthenticating around it. + /// + /// Two guards, because there are two ways to lose a visitor session. Age is + /// the common one, and a timestamp catches it before the request. Anything + /// else — a session dropped early, a machine that slept through a clock + /// change — only shows up as a failure, so a fresh-looking session gets one + /// retry behind a refresh. `refresh` is best-effort: when it fails the + /// operation reports the real problem, which is more useful than a refresh + /// error standing in for it. + pub async fn run( + &self, + share_id: &str, + refresh: R, + operation: O, + ) -> std::result::Result + where + R: Fn() -> RF, + RF: std::future::Future, + O: Fn() -> OF, + OF: std::future::Future>, + { + let stale = self.is_stale(share_id); + if stale && refresh().await { + self.mark(share_id); + } + match operation().await { + Err(_) if !stale => { + if refresh().await { + self.mark(share_id); + } + operation().await + } + result => result, + } + } } /// One opened share, whichever kind it is. @@ -728,6 +797,9 @@ impl SharedLibrary { ) -> Result<(Self, Vec<(Share, Error)>)> { let mut clients = BTreeMap::new(); let mut failures = Vec::new(); + // A reused client's session is of unknown age, so it is left stale; one + // opened here has just completed its handshake. + let sessions = SessionAges::default(); for share in store.list()? { if let Some(client) = reusable.remove(&share.id) { @@ -736,13 +808,14 @@ impl SharedLibrary { } match store.open(&share, account).await { Ok(client) => { + sessions.mark(&share.id); clients.insert(share.id.clone(), client); } Err(e) => failures.push((share, e)), } } - Ok((Self { clients }, failures)) + Ok((Self { clients, sessions }, failures)) } /// The opened clients, minus `dropping`, ready to hand to @@ -783,9 +856,37 @@ impl SharedLibrary { .client(share_id) .ok_or_else(|| Error::NotFound(format!("share {share_id} is not open")))?; client.refresh_session().await?; + self.sessions.mark(share_id); Ok(()) } + /// Run `operation` against one share, refreshing its visitor session first + /// when it is old and once more if the operation fails. See + /// [`SessionAges::run`]. Every open of something on a share goes through + /// here, so an app left running for hours does not need restarting to play. + pub async fn reauthenticating( + &self, + share_id: &str, + operation: O, + ) -> std::result::Result + where + O: Fn() -> OF, + OF: std::future::Future>, + { + self.sessions + .run( + share_id, + || async { + match self.client(share_id) { + Some(client) => client.refresh_session().await.is_ok(), + None => false, + } + }, + operation, + ) + .await + } + /// Walk one share's whole subtree, level by level, yielding every node. /// /// There is no recursion helper on the visitor path — the authenticated @@ -927,6 +1028,83 @@ fn token_from_url(url: &str) -> Result { #[cfg(test)] mod tests { use super::*; + use std::sync::atomic::{AtomicUsize, Ordering}; + + /// An operation that fails its first `failures` calls, counting every call. + fn flaky( + failures: usize, + calls: &AtomicUsize, + ) -> impl Fn() -> std::future::Ready> { + move || { + let call = calls.fetch_add(1, Ordering::SeqCst); + std::future::ready(if call < failures { Err(()) } else { Ok(()) }) + } + } + + #[tokio::test] + async fn a_share_never_refreshed_is_refreshed_before_the_operation() { + let ages = SessionAges::default(); + let (refreshes, calls) = (AtomicUsize::new(0), AtomicUsize::new(0)); + let refresh = || { + refreshes.fetch_add(1, Ordering::SeqCst); + std::future::ready(true) + }; + ages.run("s", refresh, flaky(0, &calls)).await.unwrap(); + assert_eq!(refreshes.load(Ordering::SeqCst), 1); + assert_eq!(calls.load(Ordering::SeqCst), 1); + } + + #[tokio::test] + async fn a_fresh_session_is_not_refreshed_when_the_operation_succeeds() { + let ages = SessionAges::default(); + ages.mark("s"); + let (refreshes, calls) = (AtomicUsize::new(0), AtomicUsize::new(0)); + let refresh = || { + refreshes.fetch_add(1, Ordering::SeqCst); + std::future::ready(true) + }; + ages.run("s", refresh, flaky(0, &calls)).await.unwrap(); + assert_eq!(refreshes.load(Ordering::SeqCst), 0); + } + + /// A session Proton dropped early looks fresh by age, and only the failure + /// gives it away. + #[tokio::test] + async fn a_fresh_session_that_fails_is_refreshed_once_and_retried() { + let ages = SessionAges::default(); + ages.mark("s"); + let (refreshes, calls) = (AtomicUsize::new(0), AtomicUsize::new(0)); + let refresh = || { + refreshes.fetch_add(1, Ordering::SeqCst); + std::future::ready(true) + }; + ages.run("s", refresh, flaky(1, &calls)).await.unwrap(); + assert_eq!(refreshes.load(Ordering::SeqCst), 1); + assert_eq!(calls.load(Ordering::SeqCst), 2); + } + + /// A refresh just happened, so a second one would learn nothing: the + /// failure is the operation's own and is reported as it is. + #[tokio::test] + async fn a_stale_session_that_still_fails_after_its_refresh_is_not_retried() { + let ages = SessionAges::default(); + let calls = AtomicUsize::new(0); + let result = ages + .run("s", || std::future::ready(true), flaky(5, &calls)) + .await; + assert!(result.is_err()); + assert_eq!(calls.load(Ordering::SeqCst), 1); + } + + #[tokio::test] + async fn a_failed_refresh_leaves_the_session_stale() { + let ages = SessionAges::default(); + let calls = AtomicUsize::new(0); + ages.run("s", || std::future::ready(false), flaky(0, &calls)) + .await + .unwrap(); + assert!(ages.is_stale("s")); + } #[test] fn a_share_url_yields_its_token() {