mirror of
https://github.com/narrrl/proton-stream.git
synced 2026-11-05 17:12:10 +01:00
fix(core): list a folder again when its connection drops mid-crawl
A crawl is hundreds of listings and the first failure ended it. The SDK replays a timeout or a refused connection, not a request whose connection dropped, which is what a phone sees after the OS tears down its sockets or hands over between networks. A listing is read-only, so the crawl retries it after 1, 3 and 9 seconds.
This commit is contained in:
1 parent
fb3a97b408
commit
4f130d1d6e
1 file changed
+87
-9
@@ -27,11 +27,13 @@
|
|||||||
|
|
||||||
use std::collections::BTreeMap;
|
use std::collections::BTreeMap;
|
||||||
use std::sync::Arc;
|
use std::sync::Arc;
|
||||||
|
use std::time::Duration;
|
||||||
|
|
||||||
use proton_drive_rs::{
|
use proton_drive_rs::{
|
||||||
Node, ProtonDriveClient, ProtonDrivePublicLinkClient, RevisionReader, ThumbnailType,
|
Node, ProtonDriveClient, ProtonDrivePublicLinkClient, RevisionReader, ThumbnailType,
|
||||||
};
|
};
|
||||||
use proton_sdk::config::ProtonClientConfiguration;
|
use proton_sdk::config::ProtonClientConfiguration;
|
||||||
|
use proton_sdk::error::ProtonError;
|
||||||
use proton_sdk::ids::{LinkId, NodeUid, VolumeId};
|
use proton_sdk::ids::{LinkId, NodeUid, VolumeId};
|
||||||
use serde::{Deserialize, Serialize};
|
use serde::{Deserialize, Serialize};
|
||||||
|
|
||||||
@@ -973,7 +975,10 @@ impl SharedLibrary {
|
|||||||
.client(share_id)
|
.client(share_id)
|
||||||
.ok_or_else(|| Error::NotFound(format!("share {share_id} is not open")))?;
|
.ok_or_else(|| Error::NotFound(format!("share {share_id} is not open")))?;
|
||||||
|
|
||||||
let root = client.root_node().await?;
|
let root = retrying(&CRAWL_RETRY_DELAYS, ProtonError::is_retriable, || {
|
||||||
|
client.root_node()
|
||||||
|
})
|
||||||
|
.await?;
|
||||||
let mut level = vec![root.uid.clone()];
|
let mut level = vec![root.uid.clone()];
|
||||||
let mut found = vec![root];
|
let mut found = vec![root];
|
||||||
|
|
||||||
@@ -981,14 +986,10 @@ impl SharedLibrary {
|
|||||||
let listings = level.into_iter().map(|folder| {
|
let listings = level.into_iter().map(|folder| {
|
||||||
let client = client.clone();
|
let client = client.clone();
|
||||||
async move {
|
async move {
|
||||||
let child_uids: Vec<NodeUid> =
|
retrying(&CRAWL_RETRY_DELAYS, ProtonError::is_retriable, || {
|
||||||
client.enumerate_folder_children_node_uids(&folder).await?;
|
list_folder(&client, &folder)
|
||||||
if child_uids.is_empty() {
|
})
|
||||||
return Ok(Vec::new());
|
.await
|
||||||
}
|
|
||||||
// Chunked and fanned out inside the SDK too, so this is a
|
|
||||||
// multiplier on that concurrency rather than the only one.
|
|
||||||
client.enumerate_nodes(&child_uids).await
|
|
||||||
}
|
}
|
||||||
});
|
});
|
||||||
|
|
||||||
@@ -1010,6 +1011,59 @@ impl SharedLibrary {
|
|||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
/// One folder's children, with their details.
|
||||||
|
async fn list_folder(
|
||||||
|
client: &ShareClient,
|
||||||
|
folder: &NodeUid,
|
||||||
|
) -> proton_sdk::error::Result<Vec<Node>> {
|
||||||
|
let child_uids: Vec<NodeUid> = client.enumerate_folder_children_node_uids(folder).await?;
|
||||||
|
if child_uids.is_empty() {
|
||||||
|
return Ok(Vec::new());
|
||||||
|
}
|
||||||
|
// Chunked and fanned out inside the SDK too, so this is a multiplier on
|
||||||
|
// that concurrency rather than the only one.
|
||||||
|
client.enumerate_nodes(&child_uids).await
|
||||||
|
}
|
||||||
|
|
||||||
|
/// Run `operation`, and again after each delay while it fails in a way
|
||||||
|
/// `retriable` says another attempt could fix.
|
||||||
|
///
|
||||||
|
/// A crawl is hundreds of listings, and one that fails ends the crawl. The SDK
|
||||||
|
/// replays a timeout or a refused connection itself, but not a connection that
|
||||||
|
/// dropped mid-request — the request may have been applied, which matters for
|
||||||
|
/// a write and not at all for a listing. A phone sees exactly that: a pooled
|
||||||
|
/// connection the OS tore down while the app was off screen, or a hand-over
|
||||||
|
/// from Wi-Fi to mobile data.
|
||||||
|
async fn retrying<T, E, O, OF>(
|
||||||
|
delays: &[Duration],
|
||||||
|
retriable: impl Fn(&E) -> bool,
|
||||||
|
operation: O,
|
||||||
|
) -> std::result::Result<T, E>
|
||||||
|
where
|
||||||
|
O: Fn() -> OF,
|
||||||
|
OF: std::future::Future<Output = std::result::Result<T, E>>,
|
||||||
|
{
|
||||||
|
let mut delays = delays.iter();
|
||||||
|
loop {
|
||||||
|
match operation().await {
|
||||||
|
Err(error) if retriable(&error) => match delays.next() {
|
||||||
|
Some(delay) => tokio::time::sleep(*delay).await,
|
||||||
|
None => return Err(error),
|
||||||
|
},
|
||||||
|
done => return done,
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
/// How long a crawl waits before listing a folder again. About a quarter of a
|
||||||
|
/// minute in all: long enough for a network to come back, short enough that a
|
||||||
|
/// share that is really gone still says so.
|
||||||
|
const CRAWL_RETRY_DELAYS: [Duration; 3] = [
|
||||||
|
Duration::from_secs(1),
|
||||||
|
Duration::from_secs(3),
|
||||||
|
Duration::from_secs(9),
|
||||||
|
];
|
||||||
|
|
||||||
/// How many folders of one level to list at once.
|
/// How many folders of one level to list at once.
|
||||||
///
|
///
|
||||||
/// Each of these fans out again inside the SDK — up to four detail batches per
|
/// Each of these fans out again inside the SDK — up to four detail batches per
|
||||||
@@ -1087,6 +1141,30 @@ mod tests {
|
|||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
#[tokio::test]
|
||||||
|
async fn a_listing_that_drops_is_tried_again_until_it_comes_back() {
|
||||||
|
let calls = AtomicUsize::new(0);
|
||||||
|
let result = retrying(&[Duration::ZERO; 3], |_| true, flaky(2, &calls)).await;
|
||||||
|
assert_eq!(result, Ok(()));
|
||||||
|
assert_eq!(calls.load(Ordering::SeqCst), 3);
|
||||||
|
}
|
||||||
|
|
||||||
|
#[tokio::test]
|
||||||
|
async fn a_listing_that_keeps_failing_reports_its_failure_once_the_delays_run_out() {
|
||||||
|
let calls = AtomicUsize::new(0);
|
||||||
|
let result = retrying(&[Duration::ZERO; 3], |_| true, flaky(9, &calls)).await;
|
||||||
|
assert_eq!(result, Err(()));
|
||||||
|
assert_eq!(calls.load(Ordering::SeqCst), 4);
|
||||||
|
}
|
||||||
|
|
||||||
|
#[tokio::test]
|
||||||
|
async fn a_failure_another_attempt_cannot_fix_is_reported_at_once() {
|
||||||
|
let calls = AtomicUsize::new(0);
|
||||||
|
let result = retrying(&[Duration::ZERO; 3], |_| false, flaky(1, &calls)).await;
|
||||||
|
assert_eq!(result, Err(()));
|
||||||
|
assert_eq!(calls.load(Ordering::SeqCst), 1);
|
||||||
|
}
|
||||||
|
|
||||||
#[tokio::test]
|
#[tokio::test]
|
||||||
async fn a_share_never_refreshed_is_refreshed_before_the_operation() {
|
async fn a_share_never_refreshed_is_refreshed_before_the_operation() {
|
||||||
let ages = SessionAges::default();
|
let ages = SessionAges::default();
|
||||||
|
|||||||
Reference in new issue
Block a user